1. 创建项目

  1. <properties>
  2.   <flink.version>1.13.0</flink.version>
  3.   <java.version>1.8</java.version>
  4.   <scala.binary.version>2.12</scala.binary.version>
  5.   <slf4j.version>1.7.30</slf4j.version>
  6. </properties>
  7. ...
  8. <dependency>
  9.   <groupId>org.apache.flink</groupId>
  10.   <artifactId>flink-java</artifactId>
  11.   <version>${flink.version}</version>
  12. </dependency>
  13. <dependency>
  14.   <groupId>org.apache.flink</groupId>
  15.   <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  16.   <version>${flink.version}</version>
  17. </dependency>
  18. <dependency>
  19.   <groupId>org.apache.flink</groupId>
  20.   <artifactId>flink-clients_${scala.binary.version}</artifactId>
  21.   <version>${flink.version}</version>
  22. </dependency>

2. 编写代码

2.1 批处理

批处理读取到文件后，统一进行处理

  1. import org.apache.flink.api.common.typeinfo.Types;
  2. import org.apache.flink.api.java.ExecutionEnvironment;
  3. import org.apache.flink.api.java.operators.AggregateOperator;
  4. import org.apache.flink.api.java.operators.DataSource;
  5. import org.apache.flink.api.java.operators.FlatMapOperator;
  6. import org.apache.flink.api.java.operators.UnsortedGrouping;
  7. import org.apache.flink.api.java.tuple.Tuple2;
  8. import org.apache.flink.util.Collector;
  10. public class BatchWordCount {
  11.     public static void main(String[] args) throws Exception {
  12.         // 1. 创建执行环境 --> DataSet 级别 API（批处理）
  13.         ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
  14.         // 2. 从文件读取数据  按行读取(存储的元素就是每行的文本)
  15.         DataSource<String> lineDS = env.readTextFile("input/words.txt");
  16.         // 3. 转换数据格式
  17.         FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS
  18.                 .flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
  19.                     String[] words = line.split(" ");
  20.                     for (String word : words) {
  21.                         out.collect(Tuple2.of(word, 1L));
  22.                     }
  23.                 })
  24.                 .returns(Types.TUPLE(Types.STRING, Types.LONG));  //当Lambda表达式使用 Java 泛型的时候, 由于泛型擦除的存在, 需要显示的声明类型信息
  26.         // 4. 按照 word 进行分组
  27.         UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG = wordAndOne.groupBy(0);
  28.         // 5. 分组内聚合统计
  29.         AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1);
  31.         // 6. 打印结果
  32.         sum.print();
  33.     }
  34. }
  35. ------------------------------
  36. (java,1)
  37. (flink,1)
  38. (world,1)
  39. (hello,3)

2.2 流处理

一行一行读取输入的内容，一行一行的处理

2.2.1 读取文件（有界流）

  1. package com.atguigu.chapter02;
  3. /**
  4.  * Copyright (c) 2020-2030 尚硅谷 All Rights Reserved
  5.  * <p>
  6.  * Project:  FlinkTutorial
  7.  * <p>
  8.  * Created by  wushengran
  9.  */
  11. import org.apache.flink.api.common.typeinfo.Types;
  12. import org.apache.flink.api.java.tuple.Tuple2;
  13. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  14. import org.apache.flink.streaming.api.datastream.KeyedStream;
  15. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  16. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  17. import org.apache.flink.util.Collector;
  19. import java.util.Arrays;
  21. public class BoundedStreamWordCount {
  22.     public static void main(String[] args) throws Exception {
  23.         // 1. 创建流式执行环境 --> DataStream 级别 API
  24.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  25.         // 2. 读取文件
  26.         DataStreamSource<String> lineDSS = env.readTextFile("input/words.txt");
  27.         // 3. 转换数据格式
  28.         SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS
  29.                 .flatMap((String line, Collector<String> words) -> {
  30.                     Arrays.stream(line.split(" ")).forEach(words::collect);
  31.                 })
  32.                 .returns(Types.STRING)
  33.                 .map(word -> Tuple2.of(word, 1L))
  34.                 .returns(Types.TUPLE(Types.STRING, Types.LONG));
  35.         // 4. 分组
  36.         KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne
  37.                 .keyBy(t -> t.f0);
  38.         // 5. 求和
  39.         SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
  40.                 .sum(1);
  41.         // 6. 打印
  42.         result.print();
  43.         // 7. 执行
  44.         env.execute();
  45.     }
  46. }
  47. ------------------------------
  48. Tip：group by 后，同一个分组同一个线程处理；flink使用多线程进行异步处理。
  49. 3> (world,1)
  50. 2> (hello,1)
  51. 4> (flink,1)
  52. 2> (hello,2)
  53. 2> (hello,3)
  54. 1> (java,1)

Flink 是一个分布式处理引擎，所以我们的程序应该也是分布式运行的。在开发环境里，会通过多线程来模拟 Flink 集群运行。所以这里结果前的数字，其实就指示了本地执行的不同线程，对应着 Flink 运行时不同的并行资源。这样第一个乱序的问题也就解决了：既然是并行执行，不同线程的输出结果，自然也就无法保持输入的顺序了。

2.2.2 读取文本流（无界流）

  1. import org.apache.flink.api.common.typeinfo.Types;
  2. import org.apache.flink.api.java.tuple.Tuple2;
  3. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  4. import org.apache.flink.streaming.api.datastream.KeyedStream;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.util.Collector;
  9. import java.util.Arrays;
  11. public class StreamWordCount {
  12.     public static void main(String[] args) throws Exception {
  13.         // 1. 创建流式执行环境
  14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  15.         // 2. 读取文本流
  16.         DataStreamSource<String> lineDSS = env.socketTextStream("hadoop102", 7777);
  17.         // 3. 转换数据格式
  18.         SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS
  19.                 .flatMap((String line, Collector<String> words) -> {
  20.                     Arrays.stream(line.split(" ")).forEach(words::collect);
  21.                 })
  22.                 .returns(Types.STRING)
  23.                 .map(word -> Tuple2.of(word, 1L))
  24.                 .returns(Types.TUPLE(Types.STRING, Types.LONG));
  25.         // 4. 分组
  26.         KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne
  27.                 .keyBy(t -> t.f0);
  28.         // 5. 求和
  29.         SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
  30.                 .sum(1);
  31.         // 6. 打印
  32.         result.print();
  33.         // 7. 执行
  34.         env.execute();
  35.     }
  36. }
  37. ------------------------------
  38. 从 服务器 发送数据：
  39. hello flink
  40. hello world
  41. hello java
  42. 可以看到控制台输出结果如下：
  43. 4> (flink,1)
  44. 2> (hello,1)
  45. 3> (world,1)
  46. 2> (hello,2)
  47. 2> (hello,3)
  48. 1> (java,1)
  1. 代码说明和注意事项：
  • socket 文本流的读取需要配置两个参数：发送端主机名和端口号。这里代码中指定了主机“hadoop102”的 7777 端口作为发送数据的 socket 端口，读者可以根据测试环境自行配置；
  • 在实际项目应用中，主机名和端口号这类信息往往可以通过配置文件，或者传入程序运行参数的方式来指定；
  • socket文本流数据的发送，可以通过Linux系统自带的netcat工具进行模拟。
  1. 在 Linux 环境的主机 hadoop102 上，执行下列命令，发送数据进行测试：
    1. $ nc -lk 7777

3. 总结

本章主要实现一个 Flink 开发的入门程序——词频统计 WordCount。通过批处理和流处理两种不同模式的实现，可以对 Flink 的 API 风格和编程方式有所熟悉，并且更加深刻地理解批处理和流处理的不同。另外，通过读取有界数据（文件）和无界数据（socket 文本流）进行流处理的比较，我们也可以更加直观地体会到 Flink 流处理的方式和特点。