- flatMap：一对多转换操作，输入句子，输出分词后的每个词
- groupBy：按Key分组，0代表选择第1列作为Key
- sum：求和，1代表按照第2列进行累加
- print：打印最终结果
1.WordCount代码编写
需求：统计一段文字中，每个单词出现的频次。
环境准备：在src/main/java目录下，新建一个包，命名为com.atguigu.wc。
1.1 批处理
批处理基本思路：
①.先逐行读入文件数据
②.然后将每一行文字拆分成单词
③.接着按照单词分组
④.统计每组数据的个数
⑤.就是对应单词的频次。
1.2 创建项目
1）创建工程
（1）打开IntelliJ IDEA，创建一个Maven工程。
（2）将这个Maven工程命名为Flinkdemo。
2）添加项目依赖
在项目的pom文件中，添加Flink的依赖，包括flink-java、flink-streaming-java，以及flink-clients（客户端，也可以省略）。
<properties>
<flink.version>1.17.0</flink.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>${flink.version}</version>
</dependency>
</dependencies>3)数据准备
（1）在工程根目录下新建一个Data文件夹，并在下面创建文本文件words.txt
（2）在words.txt中输入一些文字，例如：
hello flink
hello world
hello java4）代码编写
（1）在com.atguigu.wc包下新建Java类Demo01_BatchProcess，在静态main方法中编写代码。具体代码实现如下：
package com.atguigu.wordcount;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
/**as
* Created by Smexy on 2023/9/3
*
* 计算的套路:
* ①计算的环境
* spark:SparkContext
* mr: Driver
* flink：ExecutionEnvironment
* ②把要计算的数据封装为计算模型
* spark: RDD（spark core）
* DataFrame|DataSet（sparksql）
* DStream(sparkstreaming)
* mr: K-V
* flink: DataSource
*
* ③调用计算api
* RDD.转换算子()
* mr: 自己去编写Mapper，Reducer
* flink: DataSource.算子()
*
* 使用的是DataSetAPI(批处理)
*
* -------------------------
* 了解。后续不用了！
*
*
*/
public class Demo01_BatchProcess
{
public static void main(String[] args) throws Exception {
//创建支持flink计算的环境
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
//使用环境去读数据，封装为计算模型
DataSource<String> source = env.readTextFile("data/words.txt");
//调用计算api
source
/*
hello hi hi hi
变为 (hello,1)
(h1,1)
(h1,1)
(h1,1)
输出到下游
*/
.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>()
{/*
String value: 输入的一行内容
Collector<String> out： 输出结果的收集器。帮你把结果自动收集，输出到下游。
单词,1: 输出的数据是多列，此时就应该使用集合或Bean来封装。
flink提供了Tuple的集合。用于封装多个列。
Tuple2： 用来封装2列
Tuple3： 用来封装3列
....
Tuple25: 用来封装25列
*/
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
String[] words = value.split(" ");
for (String word : words) {
//Tuple2<String, Integer> data = new Tuple2<>(word, 1);
Tuple2<String, Integer> data = Tuple2.of(word, 1);
//收集要输出的数据
out.collect(data);
}
}
})
/*
收到的是 (单词,1) 格式
计算： 得到 (单词，N)
groupBy(int fileds): 适用于 对Tuple类型的数据进行聚合。传入
int N，N代表Tuple中的列的索引。
groupBy(String fileds): 适用于对Bean类型的数据进行聚合，传入的String就是
Bean中的属性名。
*/
.groupBy(0)
// 对tuple2分组后的第二列进行sum运算
.sum(1)
//在控制台打印输出
.print();
}
}
5).输出
(flink,1)
(world,1)
(hello,3)
(java,1)1.3 常见问题
问题1.
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
解决方式：maven项目的 pom.xml安装依赖：
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.32</version>
</dependency>问题2.
log4j:WARN No appenders could be found for logger (org.apache.flink.api.java.utils.PlanGenerator).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
解决办法：log4j没有配置日志记录的位置，需要配置log4j.properties，在src目录main目录resources文件夹下下新建log4j.properties
log4j.properties配置文件：
log4j.rootLogger=warn,CONSOLE,File
#Console
log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout
log4j.appender.CONSOLE.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss SSS} [%t] [%c] [%p] - %m%n
#File DailyRollingFileAppender
log4j.logger.File=info
log4j.appender.File=org.apache.log4j.DailyRollingFileAppender
log4j.appender.File.layout=org.apache.log4j.PatternLayout
log4j.appender.File.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss SSS} [%t] [%c] [%p] - %m%n
log4j.appender.File.datePattern='.'yyyy-MM-dd
log4j.appender.File.Threshold = info
log4j.appender.File.append=true
log4j.appender.File.File=d://code/logs/flink/disk.log此时再次执行成功
