- 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");//调用计算apisource/*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列*/@Overridepublic 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此时再次执行成功
