大数据-108 Flink 快速应用案例 重回Hello WordCount！方案1批数据 方案2流数据

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

• Hadoop（已更完）
• HDFS（已更完）
• MapReduce（已更完）
• Hive（已更完）
• Flume（已更完）
• Sqoop（已更完）
• Zookeeper（已更完）
• HBase（已更完）
• Redis （已更完）
• Kafka（已更完）
• Spark（已更完）
• Flink（正在更新！）

章节内容

上节完成了如下的内容：

• Flink 基本介绍
• 基本概述 适用场景 框架特点 核心组成 生态发展 处理模型 组件架构

再次回到最初的起点，Hello Word Count！

Flink 流处理 (Stream Processing)

定义

流处理是指对持续不断的数据流进行实时处理。Flink 的流处理模式非常适合处理持续产生的数据，例如来自传感器、日志记录系统或金融交易的数据流。

核心概念

• 无界数据流：流处理通常处理无界数据流，即数据流没有明确的结束点，持续不断地产生。
• 事件时间：Flink 支持基于事件时间的处理，能够处理乱序和延迟数据，使得处理更加精确。事件时间指的是数据在其产生源头的时间。
• 窗口操作：在流处理过程中，通常需要将数据按时间窗口（如滑动窗口、滚动窗口、会话窗口）进行分组，以便执行聚合或其他操作。
• 状态管理：Flink 支持有状态的流处理，这意味着处理每条数据时，可以记住之前的数据状态。例如，在流中计算一个累积的总和或频率。

Flink 批处理 (Batch Processing)

定义

批处理是指对静态的、有界的数据集进行处理。这种处理通常用于一次性的大规模数据分析，如定期的业务报告生成、数据转换和加载任务。

核心概念

• 有界数据集：批处理通常处理有界数据集，即数据集是固定大小的，有明确的开始和结束点。
• 任务并行化：在批处理模式下，Flink 会将数据集划分为多个子任务，并行执行这些任务，以加快处理速度。
• DataSet API：Flink 的 DataSet API 提供了一组高层次的操作符，用于对批数据集执行各种操作，如映射（map）、过滤（filter）、联接（join）和聚合（aggregate）。

单词统计（批数据）

需求说明

统计一个文件中各个单词出现的次数，把统计结果输出到文件

• 读取数据源
• 处理数据源
• 将读取到的数据源文件中的每一行根据空格切分
• 将切分好的每个单词拼接1
• 根据单词聚合（将相同的单词放到一起）
• 累加相同的单词（单词后面的1进行累加）
• 保存处理结果

导入依赖

<?xml version="1.0" encoding="UTF-8"?><projectxmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>flink-test</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>11</maven.compiler.source><maven.compiler.target>11</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>1.11.1</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.12</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-scala_2.12</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.12</artifactId><version>1.11.1</version><scope>provided</scope></dependency></dependencies></project>

编写代码

packageicu.wzk;importorg.apache.flink.api.common.functions.FlatMapFunction;importorg.apache.flink.api.java.DataSet;importorg.apache.flink.api.java.ExecutionEnvironment;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.core.fs.FileSystem;importorg.apache.flink.util.Collector;publicclassWordCount{publicstaticvoidmain(String[] args)throwsException{String inPath ="word-count/word-count.txt";String outPath ="word-count/word-count-result.csv";// 获取Flink批处理执行环境ExecutionEnvironment env =ExecutionEnvironment.getExecutionEnvironment();// 读取文件中的内容DataSet<String> text = env.readTextFile(inPath);// 对数据进行处理DataSet<Tuple2<String,Integer>> dataSet = text
                .flatMap(newFlatMapFunction<String,Tuple2<String,Integer>>(){@OverridepublicvoidflatMap(String line,Collector<Tuple2<String,Integer>> collector)throwsException{for(String word : line.split(" ")){
                            collector.collect(newTuple2<>(word,1));}}}).groupBy(0).sum(1);
        dataSet
                .writeAsCsv(outPath,"\n"," ",FileSystem.WriteMode.OVERWRITE).setParallelism(1);// 触发执行程序
        env.execute("Word Count");}}

测试数据

Stateful Computations over Data Streams
Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams.
Flink has been designed to run in all common cluster environments, perform computations at in-memory speed and at any scale.
Correctness guarantees
Exactly-once state consistency
Event-time processing
Sophisticated late data handling
SQL on Stream & Batch Data
DataStream API & DataSet API
ProcessFunction (Time & State)
Flexible deployment
High-availability setup
Savepoints

运行测试

结果数据

查看 word-count/word-count-result.csv 打开即可看到以下内容：

Stateful 1
any 1
common 1
computations 2
on 1
setup 1
state 1
streams. 1
unbounded 1&3
Data 2
DataStream 1
High-availability 1for1
perform 1
run 1
to 1
Event-time 1
Flexible 1
Sophisticated 1
framework 1
is 1
scale. 1
Exactly-once 1
ProcessFunction 1
Stream 1
a 1
been 1
handling 1in1
late 1
processing 2
Batch 1
DataSet 1
at 2
bounded 1
consistency 1
deployment 1
distributed 1
engine 1
has 1
API 2
Apache 1
Flink 2
SQL 1
Streams 1
all 1
designed 1
over 2
Computations 1
Savepoints 1
and 3
data 2
environments, 1
in-memory 1
speed 1
stateful 1(Time 1
Correctness 1
State)1
cluster 1
guarantees 1

单词统计（流数据）

需求说明

Socket模拟实时发送单词，使用Flink实时接收数据，对指定时间窗口内（如5秒）的数据进行聚合统计，每隔1秒汇总计算一次，并且把时间窗口内计算结果打印出来。

编写代码

Server部分

编写一个Socket服务，提供一定的数据流。

packageicu.wzk;importjava.io.IOException;importjava.io.OutputStream;importjava.io.PrintWriter;importjava.net.InetSocketAddress;importjava.net.ServerSocket;importjava.net.Socket;importjava.util.Random;publicclassWordCountServer{publicstaticvoidmain(String[] args)throwsIOException,InterruptedException{String ip ="localhost";int port =9999;Random random =newRandom();ServerSocket serverSocket =newServerSocket();InetSocketAddress address =newInetSocketAddress(ip, port);
        serverSocket.bind(address);Socket socket = serverSocket.accept();OutputStream outputStream = socket.getOutputStream();PrintWriter writer =newPrintWriter(outputStream,true);for(int i =0; i <1000; i ++){int number = random.nextInt(100);System.out.println(number);
            writer.println(number);Thread.sleep((random.nextInt(900)+100));}
        socket.close();
        serverSocket.close();}}

Flink部分

连接到上述的Server部分

packageicu.wzk;importorg.apache.flink.api.common.functions.FlatMapFunction;importorg.apache.flink.api.java.functions.KeySelector;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.streaming.api.datastream.DataStreamSource;importorg.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.streaming.api.windowing.time.Time;importorg.apache.flink.util.Collector;publicclassWordCount2{publicstaticvoidmain(String[] args)throwsException{String ip ="localhost";int port =9999;// 获取 Flink 执行环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();// 获取 Socket 输入数据DataStreamSource<String> textStream = env.socketTextStream(ip, port,"\n");SingleOutputStreamOperator<Tuple2<String,Integer>> wordCount = textStream
                .flatMap(newFlatMapFunction<String,Tuple2<String,Integer>>(){@OverridepublicvoidflatMap(String value,Collector<Tuple2<String,Integer>> out)throwsException{String[] splits = value.split("\\s");for(String word : splits){
                            out.collect(newTuple2<>(word,1));}}});SingleOutputStreamOperator<Tuple2<String,Integer>> word = wordCount
                .keyBy(newKeySelector<Tuple2<String,Integer>,Object>(){@OverridepublicObjectgetKey(Tuple2<String,Integer> value)throwsException{return value.f0;}}).timeWindow(Time.seconds(5),Time.seconds(1)).sum(1);// 输出并运行
        word.print();
        env.execute("Word Count");}}

观察结果

Server部分

351884722451151365985568228417

Flink部分

3>(35,1)4>(18,1)3>(35,1)5>(84,1)4>(18,1)6>(72,1)3>(35,1)5>(84,1)5>(24,1)3>(35,1)6>(72,1)4>(18,1)7>(51,1)5>(24,1)5>(84,1)4>(15,1)6>(72,1)7>(51,1)3>(35,1)4>(15,1)4>(18,1)

运行结果过程截图如下所示：

过程总结

• 获得一个执行环境
• 加载、创建 初始化环境
• 指定数据操作的算子
• 指定结果数据存放位置
• 调用Execute触发执行程序

注意：Flink程序是延迟计算的，只有最后调用execute()方法的时候才会真正的触发执行程序。