Flink实时统计单词【入门】

前言

单词统计【word count】是flink的最最最基础的入门案例，就犹如学习java的第一堂课是运行一个hello world程序同样告急。
这里通过使用netcat发送数据来模仿无界数据流。
代码案例

在代码案例之前，我们必要先介绍一下netcat这个工具。
netcat介绍

netcat（简称nc）是一个强盛的网络工具，被称之为网络瑞士军刀。他能通过TCP/UDP协议进行数据传输，支持一下核心功能

• 端口扫描：测试端口是否开放
  
• 数据传输：作为客户端、服务端发送或接收数据
  
• 网络调试：模仿socket服务端或客户端
  

这个案例中，我们就使用netcat来模仿socket服务端来发送数据。
大多数的linux发行版本默认预装了netcat，直接使用nc命令就行了

1. # 安装（如未预装）
2. sudo apt install netcat  # Debian/Ubuntu
3. sudo yum install nc      # CentOS/RHEL
5. # 启动服务端（监听端口 9999）
6. nc -lk 9999

复制代码

笔者使用的是腾讯云轻量级服务器ubuntu，已经预装了netcat，记得要在开放对应的端口权限，比如

创建maven工程

创建一个简单的maven工程，对应的pom文件如下

1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
2.   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
3.   <modelVersion>4.0.0</modelVersion>
5.   <groupId>com.tml</groupId>
6.   <artifactId>flink-demo</artifactId>
7.   <version>1.0-SNAPSHOT</version>
8.   <packaging>jar</packaging>
10.   <name>flink-demo</name>
11.   <url>http://maven.apache.org</url>
13.   <properties>
14.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
15.     <flink.version>1.18.0</flink.version> <!-- 根据你的 Flink 版本进行调整 -->
16.     <maven.compiler.source>1.8</maven.compiler.source>
17.     <maven.compiler.target>1.8</maven.compiler.target>
18.   </properties>
20.   <dependencies>
21.     <dependency>
22.       <groupId>junit</groupId>
23.       <artifactId>junit</artifactId>
24.       <version>3.8.1</version>
25.       <scope>test</scope>
26.     </dependency>
28.     <dependency>
29.       <groupId>org.apache.flink</groupId>
30.       <artifactId>flink-java</artifactId>
31.       <version>${flink.version}</version>
32.     </dependency>
33.     <!-- Flink Streaming API -->
34.     <dependency>
35.       <groupId>org.apache.flink</groupId>
36.       <artifactId>flink-streaming-java</artifactId>
37.       <version>${flink.version}</version>
38.     </dependency>
39.     <!-- Flink Table API and SQL -->
40.     <dependency>
41.       <groupId>org.apache.flink</groupId>
42.       <artifactId>flink-clients</artifactId>
43.       <version>${flink.version}</version>
44.     </dependency>
45.   </dependencies>
46. </project>

复制代码

代码案例 

这里的需求就是，每来了一条消息，我就按照空字符进行切分，进行统计，比如hello world如许一条消息，我会根据空格来进行切分，对hello、world分别计数为1，最后进行累加，从而实时统计单词的数量

1. package com.tml;
3. import org.apache.flink.api.java.tuple.Tuple2;
4. import org.apache.flink.streaming.api.datastream.DataStream;
5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
8. /**
9. * 从socket流读取数据
10. */
11. public class WordCountFromSocket {
13.     public static void main(String[] args) throws Exception {
14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
15.         env.setParallelism(1);
16.         DataStream<String> socketTextStream = env.socketTextStream("xxx", 9999);
17.         //数据流处理
18.         SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketTextStream.flatMap(new Tokenizer()).keyBy(t -> t.f0).sum(1);
19.         sum.print();
20.         env.execute("Socket Stream WordCount~");
21.     }
22. }

复制代码

真正的逻辑处理是在Tokenizer类中，对应的代码如下

1. package com.tml;
3. import org.apache.flink.api.common.functions.FlatMapFunction;
4. import org.apache.flink.api.java.tuple.Tuple2;
5. import org.apache.flink.util.Collector;
8. /**
9. * 自定义 FlatMapFunction 实现单词拆分
10. */
11. public class Tokenizer implements FlatMapFunction<String, Tuple2<String,Integer>> {
12.     @Override
13.     public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
14.         //按照空格或者制表符分割单词
15.         String[] words = s.split("\\s+");
16.         for (String word : words) {
17.             collector.collect(new Tuple2<>(word, 1));
18.         }
19.     }
20. }

复制代码

 运行案例

先启动服务端

再启动客户端

运行main主类，发现程序再阻塞中，等待数据的流入，从这里也可以看出，flink是基于变乱驱动的。

模仿socket输入

运行结果查看

 从控制台的结果输出可以看到，flink的实时统计的结果是没有题目的！
总结

flink入门级别的案例，从运行过程到运行结果，可以感受到flink实时计算的强盛！完整的代码已上传至github【flink-demo】，欢迎围观！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。