• Flink1.17 DataStream API

    目录

    一.执行环境（Execution Environment）

    1.1 创建执行环境

    1.2 执行模式

    1.3 触发程序执行

    二.源算子（Source）

    2.1 从集合中读取数据

    2.2 从文件读取数据

    2.3 从 RabbitMQ 中读取数据

    2.4 从数据生成器读取数据 

    2.5 Flink支持的数据类型

    2.5.1 Flink的类型系统

    2.5.2 Flink支持的数据类型

    2.5.3 类型提示（Type Hints）

    三.转换算子（Transformation）

    3.1 基本转换算子（map/ filter/ flatMap）

    3.1.1 映射（map）

    3.1.2 过滤（filter）

    3.1.2 扁平映射（flatMap）

    3.2 聚合算子（Aggregation）

    3.2.1 按键分区（keyBy）

     3.2.2 简单聚合（sum/min/max/minBy/maxBy）

    3.2.3 归约聚合（reduce）

    3.3 用户自定义函数（UDF）

    3.3.1 函数类（Function Classes）

    3.3.2 富函数类（Rich Function Classes）

    3.4 物理分区算子（Physical Partitioning）

    3.4.1 随机分区（shuffle）

    3.4.2 轮询分区（Round-Robin）

    3.4.3 重缩放分区（rescale）

     3.4.4 广播（broadcast）

    3.4.5 全局分区（global）

    3.4.6 自定义分区（Custom） 

    3.5 分流

    3.5.1 Filter 实现分流

     3.5.2 使用侧输出流

    3.6 基本合流操作

    3.6.1 联合（Union）

    3.6.2 连接（Connect）

    3.6.2.1 连接流（ConnectedStreams）

    3.6.2.2 CoProcessFunction

    四.输出算子（Sink）

     4.1 连接到外部系统

    4.2 输出到文件 

    4.3 输出到RabbitMQ

    4.4 输出到MySQL（JDBC） 

     4.5 自定义Sink输出

    DataStream API是Flink的核心层API。一个Flink程序，其实就是对DataStream的各种转换。具体来说，代码基本上都由以下几部分构成：

    一.执行环境（Execution Environment）

            Flink程序可以在各种上下文环境中运行：既可以可以在本地JVM中执行程序，也可以提交到远程集群上运行。

    1.1 创建执行环境

    获取的执行环境是StreamExecutionEnvironment类的对象（流处理，批处理已经标记为过时），创建执行环境一般有以下三种方式：

    1. // 创建一个本地执行环境并返回，可传入并行度，默认是本地CPU核心数
    2. StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
    3.  
    4.  
    5. // 返回远程集群执行环境，需传入远程IobManager的主机名与端口、及在集群中需运行的Jar包
    6. StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("JobManager ip", "JobManager port","提交给JobManager的JAR包");
    7.  
    8.  
    9. // （推荐）根据当前环境自动选择执行环境，无脑选这个即可
    10. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

     当使用 getExecutionEnvironment() 创建环境时，可以传入 org.apache.flink.configuration.Configuration 类来手动指定默认的参数，例如端口等。

    1. Configuration conf = new Configuration();
    2. conf.set(RestOptions.PORT,8088);
    3. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

    1.2 执行模式

    从Flink 1.12开始，官方推荐的做法是直接使用DataStream API，在提交任务时通过将执行模式设为BATCH来进行批处理。不建议使用DataSet API（已过时）。

    通过代码指定：

    1. // 流 执行模式
    2. env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
    3. // 批 执行模式
    4. env.setRuntimeMode(RuntimeExecutionMode.BATCH);
    5. // 自动模式，根据数据源是否有界自动选择执行模式
    6. env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

    提交任务时命令行指定（推荐）：

    bin/flink run -Dexecution.runtime-mode=BATCH ...

    同一套代码/API，既可以指定流处理也可以指定批处理，这就是“流批一体”的其中一个解释。

    1.3 触发程序执行

    1. // 程序执行
    2. env.execute();

    写完输出（sink）操作并不代表程序已经结束。因为当 main() 方法被调用时，其实只是定义了作业的每个执行操作，然后添加到数据流图中；这时并没有真正处理数据——因为数据可能还没来。Flink是由事件驱动的，只有等到数据到来，才会触发真正的计算，这也被称为“延迟执行”或“懒执行”。

    默认 main 方法的一个 env.execute() 会触发一个 Flink Job，并且一个 main 方法可以调用多个 env.execute() ，但无意义，因为第一个会阻塞住。可使用 env.executeAsync() 可以异步触发，而且不会产生阻塞。

    在application模式下，代码中有多少个 env.executeAsync() ,就会有多少个Job，对应就会有多少个 JboManager。

    二.源算子（Source）

    Flink可以从各种来源获取数据，然后构建DataStream进行转换处理。一般将数据的输入来源称为数据源（data source），而读取数据的算子就是源算子（source operator）。所以，source就是我们整个处理程序的输入端。

    从Flink1.12开始，主要使用流批统一的新Source架构：

    DataStreamSource stream = env.fromSource(…)

    Flink直接提供了很多预实现的接口，此外还有很多外部连接工具也帮我们实现了对应的Source，通常情况下足以应对我们的实际需求。

    2.1 从集合中读取数据

            最简单的读取数据的方式，就是在代码中直接创建一个Java集合，然后调用执行环境的fromCollection方法进行读取。这相当于将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用，一般用于测试。

    1.  public static void main(String[] args) throws Exception {
    2.         
    3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    4.  
    5.         // 从集合中读取数据
    6.         DataStreamSource source = env.fromCollection(Arrays.asList(1, 10, 99, 53));
    7.  
    8.         source.print();
    9.  
    10.         env.execute();
    11.  
    12.     }

    输出结果：

    1. 6> 1
    2. 8> 99
    3. 1> 53
    4. 7> 10

    2.2 从文件读取数据

    在实际场景中，可能要读取、处理日志文件这样的需求，这也是批处理最常见的读取方式。 

    读取文件，需要添加文件连接器依赖:

    1. <dependency>
    2.     <groupId>org.apache.flinkgroupId>
    3.     <artifactId>flink-connector-filesartifactId>
    4.     <version>${flink.version}version>
    5. dependency>

     代码如下：

    1. public static void main(String[] args) throws Exception {
    2.  
    3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    4.  
    5.         
    6.         FileSource fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("input/words.txt")).build();
    7.  
    8.         // !使用官方推荐的新的 Source 架构  => env.fromSource(Source实现类，Watermark，资源名称)
    9.         env.fromSource(fileSource,WatermarkStrategy.noWatermarks(),"file").print();
    10.  
    11.         env.execute();
    12.  
    13.     }

    输出结果：

    1. 3> hello flink
    2. 3> hello world
    3. 3> hello java

    2.3 从 RabbitMQ 中读取数据

    导入相关依赖：

    2. <dependency>
    3.     <groupId>org.apache.flinkgroupId>
    4.     <artifactId>flink-connector-rabbitmqartifactId>
    5.     <version>3.0.1-1.17version>
    6. dependency>
    8. <dependency>
    9.     <groupId>com.rabbitmqgroupId>
    10.     <artifactId>amqp-clientartifactId>
    11.     <version>5.14.1version>
    12. dependency>

    相关代码：

    1. import org.apache.flink.api.common.serialization.SimpleStringSchema;
    2. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    4. import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
    5. import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
    6.  
    7. /**
    8.  * 从 RabbitMQ读取数据
    9.  */
    10. public class RabbitMQSourceDemo {
    11.  
    12.     public static void main(String[] args) throws Exception {
    13.  
    14.         // 获取执行环境
    15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    16.  
    17.         // 配置 RabbitMQ 连接信息
    18.         RMQConnectionConfig config = new RMQConnectionConfig.Builder()
    19.                 .setHost("RabbitMQ服务器地址")
    20.                 .setPort(RabbitMQ端口)
    21.                 .setUserName(用户名)
    22.                 .setPassword(密码)
    23.                 .setVirtualHost(虚拟主机)
    24.                 .build();
    25.  
    26.         // 添加 RabbitMQ 数据源（Flink 1.17 并不支持使用 env.fromSource 在 RabbitMQ 读取数据！）
    27.         RMQSource<String> source = new RMQSource<>(
    28.                 config,                     //  连接配置
    29.                 "test_queue",               // 队列名称
    30.                 new SimpleStringSchema());  // 反序列化器
    31.  
    32.         // 添加数据源
    33.         DataStreamSource<String> rabbitMQStream = env.addSource(source);
    34.  
    35.         // 打印
    36.         rabbitMQStream.print();
    37.         
    38.         // 执行
    39.         env.execute("RabbitMQ job");
    40.     }
    41. }

    进入 RabbitMQ Web 页面，在对应的虚拟主机下创建相关的队列，进入队列中，使用 Web 中的 Publish message给队列发送消息：

    输出结果：

    2.4 从数据生成器读取数据 

    Flink从1.11开始提供了一个内置的DataGen 连接器，主要是用于生成一些随机数，用于在没有数据源的时候，进行流任务的测试以及性能测试等。1.17提供了新的Source写法，需要导入依赖：

    1. <dependency>
    2.     <groupId>org.apache.flinkgroupId>
    3.     <artifactId>flink-connector-datagenartifactId>
    4.     <version>${flink.version}version>
    5. dependency>

     代码：

    1. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    2. import org.apache.flink.api.common.typeinfo.Types;
    3. import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
    4. import org.apache.flink.connector.datagen.source.DataGeneratorSource;
    5. import org.apache.flink.connector.datagen.source.GeneratorFunction;
    6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    7.  
    8. /**
    9.  * Flink 数据生成器
    10.  */
    11. public class DataGeneratorDemo {
    12.     public static void main(String[] args) throws Exception {
    13.         // 获取执行环境
    14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    15.  
    16.         /**
    17.          * 数据生成器的四个参数：
    18.          * 1、GeneratorFunction的map实现。重写返回值
    19.          * 2、返回的个数 会从0开始依次返回（使用Long.MAX_VALUE可模拟出无界流）
    20.          * 3、限速，每秒多少个数据
    21.          * 4、返回值类型
    22.          */
    23.         DataGeneratorSource dataGeneratorSource = new DataGeneratorSource<>(
    24.                 (GeneratorFunction) num -> "Number:" + num,
    25.                 30,
    26.                 RateLimiterStrategy.perSecond(3),
    27.                 Types.STRING
    28.         );
    29.  
    30.         env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(),"datagenerator-source").print();
    31.  
    32.         // 执行
    33.         env.execute();
    34.     }
    35. }

    输出：

    1. 1> Number:8
    2. 8> Number:12
    3. 4> Number:27
    4. 3> Number:0
    5. 2> Number:4
    6. 6> Number:24
    7. 5> Number:16
    8. 7> Number:20
    9. 3> Number:1
    10. 7> Number:21
    11. 5> Number:17
    12. .
    13. .
    14. .

    2.5 Flink支持的数据类型

    2.5.1 Flink的类型系统

    Flink使用“类型信息”（TypeInformation）来统一表示数据类型。TypeInformation类是Flink中所有类型描述符的基类。它涵盖了类型的一些基本属性，并为每个数据类型生成特定的序列化器、反序列化器和比较器。

    2.5.2 Flink支持的数据类型

    对于常见的Java和Scala数据类型，Flink都是支持的。Flink在内部，Flink对支持不同的类型进行了划分，这些类型可以在Types工具类中找到：

    其中包含 Java 基本类型及包装类、数组类型、复合数据类型、辅助类型（List、Map等）、泛型类型（GENERIC）。

    符合类型又包括：

    • Java元组类型（TUPLE）：这是Flink内置的元组类型，是Java API的一部分。最多25个字段，也就是从Tuple0~Tuple25，不支持空字段。
    • Scala 样例类及Scala元组：不支持空字段。
    • 行类型（ROW）：可以认为是具有任意个字段的元组，并支持空字段。
    • POJO：Flink自定义的类似于Java bean模式的类。（POJO的类和属性是公有的、有一个无参构造、属性可序列化）

    2.5.3 类型提示（Type Hints）

    Flink还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应的序列化器和反序列化器。例如：

    1. .map(word -> Tuple2.of(word, 1L))
    2. .returns(Types.TUPLE(Types.STRING, Types.LONG));

    可写作：

    1. .map(word -> Tuple2.of(word, 1L))
    2. .returns(new TypeHint>(){})

    三.转换算子（Transformation）

    数据源读入数据之后，我们就可以使用各种转换算子，将一个或多个DataStream转换为新的DataStream。

    3.1 基本转换算子（map/ filter/ flatMap）

    准备工作

    为了方便练习，这里使用WaterSensor作为数据模型。

    字段名

    数据类型

    说明

    id

    String

    水位传感器类型

    ts

    Long

    传感器记录时间戳

    vc

    Integer

    水位记录

    代码如下：

    1. public class WaterSensor {
    2.     public String id;
    3.     public Long ts;
    4.     public Integer vc;
    5.  
    6.     // 省略getter、setter、构造器、toString
    7. }

    3.1.1 映射（map）

    与 JDK1.8 中的Stream中的 Map 类似。Map 就是将一个元素映射成另一个元素。基于DataStream调用map()方法就可以进行转换处理。

    例子：需要提取 WaterSensor 中的 id 属性：

    1. public class MapDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.  
    6.         // 构造数据
    7.         DataStreamSource source = env.fromElements(
    8.                 new WaterSensor("id_1", 1l, 1),
    9.                 new WaterSensor("id_2", 2l, 2),
    10.                 new WaterSensor("id_3", 3l, 3)
    11.         );
    12.  
    13.         // 方法1：实现匿名内部类
    14.         source.map(new MapFunction() {
    15.             @Override
    16.             public String map(WaterSensor waterSensor) throws Exception {
    17.                 return waterSensor.getId();
    18.             }
    19.         }).print();
    20.  
    21.  
    22.         // 方法2：Lambda 表达式
    23.         source.map(WaterSensor::getId).print();
    24.  
    25.         // 方法三：定义 MapFunction 实现类
    26.         source.map(new MyMapFunction()).print();
    27.         env.execute();
    28.     }
    29.  
    30.     // 实现 MapFunction ， 可以复用
    31.     static class MyMapFunction implements MapFunction{
    32.         @Override
    33.         public String map(WaterSensor waterSensor) throws Exception {
    34.             return waterSensor.getId();
    35.         }
    36.     }
    37. }

    结果输出：

    1. 2> id_3
    2. 8> id_1
    3. 1> id_2

    3.1.2 过滤（filter）

    与 JDK1.8 中的Stream中的 Fliter类似。对数据流进行过滤，满足条件的元素则会被输出，不满足则被过滤。

    例子：过滤掉 WaterSensor 中 id 不为 “id_1” 的元素。

    1. public class FilterDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.  
    6.         // 构造数据
    7.         DataStreamSource source = env.fromElements(
    8.                 new WaterSensor("id_1", 1l, 1),
    9.                 new WaterSensor("id_1", 12l, 2),
    10.                 new WaterSensor("id_2", 2l, 2),
    11.                 new WaterSensor("id_3", 3l, 3)
    12.         );
    13.  
    14.         // 过滤数据中 id 不为 id_1 的元素
    15.         source.filter(new FilterFunction() {
    16.             @Override
    17.             public boolean filter(WaterSensor waterSensor) throws Exception {
    18.                 return "id_1".equals(waterSensor.getId());
    19.             }
    20.         }).print();
    21.  
    22.         env.execute();
    23.     }
    24.  
    25. }

    结果输出：

    1. 3> WaterSensor{id='id_1', ts=1, vc=1}
    2. 4> WaterSensor{id='id_1', ts=12, vc=2}

    3.1.2 扁平映射（flatMap）

    flatMap操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生0到多个元素。flatMap可以认为是“扁平化”（flatten）和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理。

    例子：如果 id 为 id_1 则输出 vc 属性，如果 id 为 id_2 则输出 ts、vc 属性。

    1. public class FlatmapDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.  
    6.         // 构造数据
    7.         DataStreamSource source = env.fromElements(
    8.                 new WaterSensor("id_1", 1l, 1),
    9.                 new WaterSensor("id_1", 12l, 2),
    10.                 new WaterSensor("id_2", 2l, 2),
    11.                 new WaterSensor("id_3", 3l, 3)
    12.         );
    13.  
    14.         /**
    15.          * 如果 id 为 id_1 则输出 vc 属性
    16.          * 如果 id 为 id_2 则输出 ts、vc 属性
    17.          */
    18.         source.flatMap(new FlatMapFunction() {
    19.             @Override
    20.             public void flatMap(WaterSensor waterSensor, Collector collector) throws Exception {
    21.                 if("id_1".equals(waterSensor.getId())){
    22.                     // 将 vc 放入采集器
    23.                     collector.collect(waterSensor.getVc().toString());
    24.                 } else if ("id_2".equals(waterSensor.getId())) {
    25.                     // 将 ts、vc 放入采集器
    26.                     collector.collect(waterSensor.getVc().toString());
    27.                     collector.collect(waterSensor.getTs().toString());
    28.                 }
    29.             }
    30.         }).print();
    31.  
    32.         env.execute();
    33.     }
    34. }

    结果输出：

    1. 2> 1
    2. 3> 2
    3. 4> 2
    4. 4> 2

    3.2 聚合算子（Aggregation）

    计算的结果不仅依赖当前数据，还跟之前的数据有关，相当于要把所有数据聚在一起进行汇总合并——这就是所谓的“聚合”（Aggregation），类似于MapReduce中的reduce操作。

    3.2.1 按键分区（keyBy）

    在Flink中，要做聚合，需要先进行分区；这个操作就是通过keyBy来完成的。keyBy通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务。 

    • 基于不同的key，流中的数据将被分配到不同的分区中去；这样一来，所有具有相同的key的数据，都将被发往同一个分区。一个子任务就可以理解为一个分区。
    • KeyBy 返回的是 KeyedStream 键控流。
    • KeyBy 不是转换算子，只是对数据做重分区，不能设置并行度。
    • 分区是通过对 Key 进行 Hash 再对分区数取模来实现的。

    例子：以 id 作为 Key 进行分区：

    1. public class KeyByDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.  
    6.         // 构造数据
    7.         DataStreamSource source = env.fromElements(
    8.                 new WaterSensor("id_1", 1l, 1),
    9.                 new WaterSensor("id_1", 12l, 2),
    10.                 new WaterSensor("id_2", 2l, 2),
    11.                 new WaterSensor("id_3", 3l, 3)
    12.         );
    13.  
    14.        // 以 id 为 Key 进行分区
    15.         source.keyBy(new KeySelector() {
    16.             @Override
    17.             public String getKey(WaterSensor waterSensor) throws Exception {
    18.                 return waterSensor.getId();
    19.             }
    20.         }).print();
    21.  
    22.         env.execute();
    23.     }
    24.     
    25. }

     结果输出：

    1. 2> WaterSensor{id='id_2', ts=2, vc=2}
    2. 3> WaterSensor{id='id_1', ts=1, vc=1}
    3. 3> WaterSensor{id='id_1', ts=12, vc=2}
    4. 3> WaterSensor{id='id_3', ts=3, vc=3}

     3.2.2 简单聚合（sum/min/max/minBy/maxBy）

    所有的聚合操作都要基于按键分区的数据流KeyedStream。 Flink为我们内置实现了一些最基本、最简单的聚合API，主要有以下几种：

    • sum()：在输入流上，对指定的字段做叠加求和的操作。
    • min()：在输入流上，对指定的字段求最小值。
    • max()：在输入流上，对指定的字段求最大值。
    • sumBy()、minBy()、maxBy()：功能类似，xxxBy() 会返回包含符合要求的整条数据。而不加 By 只会保留第一次的非比较字段。

    例子：

    1. public class AggrDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         env.setParallelism(2);
    6.         // 构造数据
    7.         DataStreamSource source = env.fromElements(
    8.                 new WaterSensor("id_1", 1l, 1),
    9.                 new WaterSensor("id_1", 12l, 22),
    10.                 new WaterSensor("id_2", 2l, 2),
    11.                 new WaterSensor("id_3", 3l, 3)
    12.         );
    13.  
    14.        // 以 id 为 Key 进行分区
    15.         KeyedStream keyBySource = source.keyBy(new KeySelector() {
    16.             @Override
    17.             public String getKey(WaterSensor waterSensor) throws Exception {
    18.                 return waterSensor.getId();
    19.             }
    20.         });
    21.  
    22.  
    23.         keyBySource.sum("vc").print();
    24.         keyBySource.min("vc").print();
    25.         
    26.         keyBySource.max("vc").print();
    27.         /**
    28.          * max结果：
    29.          * 1> WaterSensor{id='id_1', ts=1, vc=1}
    30.          * 1> WaterSensor{id='id_1', ts=1, vc=22}
    31.          * 1> WaterSensor{id='id_2', ts=2, vc=2}
    32.          * 1> WaterSensor{id='id_3', ts=3, vc=3}
    33.          * ts 还是 第一次的值
    34.          */
    35.  
    36.         keyBySource.maxBy("vc").print();
    37.         /**
    38.          * max结果：
    39.          * 1> WaterSensor{id='id_1', ts=1, vc=1}
    40.          * 1> WaterSensor{id='id_1', ts=12, vc=22}
    41.          * 1> WaterSensor{id='id_2', ts=2, vc=2}
    42.          * 1> WaterSensor{id='id_3', ts=3, vc=3}
    43.          * 取当前整列值 
    44.          */
    45.  
    46.         env.execute();
    47.     }
    48.  
    49. }

    3.2.3 归约聚合（reduce）

    reduce可以对已有的数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。

    ReduceFunction接口里需要实现reduce()方法，这个方法接收两个输入事件，经过转换处理之后输出一个相同类型的事件。在流处理的底层实现过程中，实际上是将中间“合并的结果”作为任务的一个状态保存起来的；之后每来一个新的数据，就和之前的聚合状态进一步做归约。

    例子：只保存每个分组中 VC 最大的那条数据

    1. public class ReduceDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         env.setParallelism(1);
    6.         // 构造数据
    7.         DataStreamSource source = env.fromElements(
    8.                 new WaterSensor("id_1", 1l, 1),
    9.                 new WaterSensor("id_1", 21l, 21),
    10.                 new WaterSensor("id_1", 31l, 31),
    11.                 new WaterSensor("id_2", 2l, 2),
    12.                 new WaterSensor("id_3", 3l, 3)
    13.         );
    14.  
    15.         // 以 id 为 Key 进行分区
    16.         KeyedStream sensorKs = source.keyBy(new KeySelector() {
    17.             @Override
    18.             public String getKey(WaterSensor waterSensor) throws Exception {
    19.                 return waterSensor.getId();
    20.             }
    21.         });
    22.  
    23.         /**
    24.          * reduce:
    25.          *  1.必须在KeyBy后调用
    26.          *  2.输入类型 = 输出类型
    27.          *  3.每个分区的第一条数据来的时候不会执行reduce，但是会存起来保存状态，直接输出，，“Flink有状态的体现”
    28.          *  4.reduce( value1,  value2)
    29.          *     a.value1 是上一次的计算结果
    30.          *     b.value2 是当前进入的数据
    31.          */
    32.         SingleOutputStreamOperator reduce = sensorKs.reduce(new ReduceFunction() {
    33.             @Override
    34.             public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
    35.                 if(value2.getVc() > value1.getVc()){
    36.                     return new WaterSensor(value2.getId(), value2.getTs(), value2.getVc());
    37.                 }else {
    38.                     return value1;
    39.                 }
    40.  
    41.             }
    42.         });
    43.  
    44.         reduce.print();
    45.  
    46.         env.execute();
    47.     }
    48. }

    结果输出：

    1. WaterSensor{id='id_1', ts=1, vc=1} // 分组的第一条数据直接返回
    2. WaterSensor{id='id_1', ts=21, vc=21}
    3. WaterSensor{id='id_1', ts=31, vc=31}
    4. WaterSensor{id='id_2', ts=2, vc=2} // 分组的第一条数据直接返回
    5. WaterSensor{id='id_3', ts=3, vc=3} // 分组的第一条数据直接返回

    reduce同简单聚合算子一样，也要针对每一个key保存状态。因为状态不会清空，所以我们需要将reduce算子作用在一个有限key的流上。

    3.3 用户自定义函数（UDF）

    用户自定义函数（user-defined function，UDF），即用户可以根据自身需求，重新实现算子的逻辑。

    用户自定义函数分为：函数类、匿名函数、富函数类。

    3.3.1 函数类（Function Classes）

    Flink暴露了所有UDF函数的接口，具体实现方式为接口或者抽象类，例如MapFunction、FilterFunction、ReduceFunction等。所以用户可以自定义一个函数类，实现对应的接口。

    匿名内部类实现：

    1.  source.filter(new FilterFunction() {
    2.             @Override
    3.             public boolean filter(WaterSensor waterSensor) throws Exception {
    4.                 return "id_1".equals(waterSensor.getId());
    5.             }
    6.         }).print();

    Lambda表达式实现:

    source.filter((FilterFunction) waterSensor -> "id_1".equals(waterSensor.getId())).print();

    实现 XxxFunction 接口：

    1. public class FilterFunctionImpl implements FilterFunction {
    2.  
    3.     public String id ;
    4.  
    5.     public FilterFunctionImpl(String id) {
    6.         this.id = id;
    7.     }
    8.  
    9.     @Override
    10.     public boolean filter(WaterSensor waterSensor) throws Exception {
    11.         return this.id.equals(waterSensor.getId());
    12.     }
    13. }

    3.3.2 富函数类（Rich Function Classes）

    “富函数类”也是DataStream API提供的一个函数类的接口，所有的Flink函数类都有其Rich版本。

    RichXxxFunction 与 XxxFunction 的区别是可以获取到任务运行时的一些上下文信息、环境信息以及对任务生命周期的管理。

    典型的生命周期方法有：

    • 重写open()方法，每个子任务在启动时，会调用一次。
    • 重写close()方法，每个子任务在结束时会调用一次。
      • 程序异常退出不会调用 close() 方法。
      • 手动取消任务会调用 close() 方法。

    在open、close中可以使用 getRuntimeContext() 来获取运行时上下文信息。

    1. public class RichMapFunctionDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         env.setParallelism(1);
    6.  
    7.         DataStreamSource source = env.fromElements(1, 2, 3, 4);
    8.  
    9.         SingleOutputStreamOperator map = source.map(new RichMapFunction() {
    10.  
    11.             @Override
    12.             public void open(Configuration parameters) throws Exception {
    13.                 System.out.println("open:子任务名称"+getRuntimeContext().getTaskNameWithSubtasks());
    14.                 System.out.println("open:子任务编号"+getRuntimeContext().getIndexOfThisSubtask());
    15.                 super.open(parameters);
    16.  
    17.             }
    18.  
    19.             @Override
    20.             public void close() throws Exception {
    21.                 System.out.println("close:子任务名称"+getRuntimeContext().getTaskNameWithSubtasks());
    22.                 System.out.println("close:子任务编号"+getRuntimeContext().getIndexOfThisSubtask());
    23.                 super.close();
    24.  
    25.             }
    26.  
    27.             @Override
    28.             public Integer map(Integer value) throws Exception {
    29.                 return value + 1;
    30.             }
    31.         });
    32.  
    33.         map.print();
    34.  
    35.         env.execute();
    36.     }
    37. }

    输出结果：

    1. open:子任务名称Source: Collection Source -> Map -> Sink: Print to Std. Out (1/1)#0
    2. open:子任务编号0
    3. 2
    4. 3
    5. 4
    6. 5
    7. close:子任务名称Source: Collection Source -> Map -> Sink: Print to Std. Out (1/1)#0
    8. close:子任务编号0

    3.4 物理分区算子（Physical Partitioning）

    Flink 为我们提供了7种分区策略和一个用户自定义分区器。常见的物理分区策略有：随机分配（Random）、轮询分配（Round-Robin）、重缩放（Rescale）和广播（Broadcast）。

    分区算子就是将数据按照某种策略分配到下游算子的子任务分区中。 

    3.4.1 随机分区（shuffle）

    通过调用DataStream的.shuffle()方法，将数据随机地分配到下游算子的并行任务分区中去。

    shuffle底层实现采用的是 生成随机数

    1. public static void main(String[] args) throws Exception {
    2.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    3.         env.setParallelism(2);
    4.  
    5.         DataStreamSource source = env.socketTextStream("ip", 端口);
    6.  
    7.         // 随机分区 random.nextInt(下游算子并行度)
    8.         source.shuffle().print();
    9.  
    10.         env.execute();
    11.     }
    1. 输入：
    2. [root@VM-55-24-centos ~]# nc -lk 1234
    3. 1
    4. 2
    5. 3
    6. 4
    7. 5
    8.  
    9. 输出：
    10. 1> 1
    11. 2> 2
    12. 2> 3
    13. 1> 4
    14. 1> 5

    从控制台输出的左侧子任务编号可以看出子任务分区是随机分配的。

    3.4.2 轮询分区（Round-Robin）

    通过调用DataStream的.rebalance()方法，就可以实现轮询重分区。rebalance采用的是对并行度取模，可以将输入流数据平均分配到下游的并行任务中去。可以解决 数据源数据倾斜 的问题。

    1.  // 轮询重分区
    2.  source.rebalance().print();
    1. 输入：
    2. [root@VM-55-24-centos ~]# nc -lk 1234
    3. 1
    4. 2
    5. 3
    6. 4
    7. 5
    8.  
    9. 输出：
    10. 1> 1
    11. 2> 2
    12. 1> 3
    13. 2> 4
    14. 1> 5

    3.4.3 重缩放分区（rescale）

    与 rebalance 类似，也是轮询的效果，不过比轮询更加高效。rescale的做法是将数据在固定的几个分区中进行轮询，而不是轮询所有分区。

    1. // 缩放轮询
    2. source.rescale().print();

     3.4.4 广播（broadcast）

    通过调用DataStream的broadcast()方法，会将数据发送到下游算子的所有并行任务中去。慎用！

    1.  // 广播
    2.  source.broadcast().print();

    3.4.5 全局分区（global）

    全局分区做法非常极端，通过调用.global()方法，会将所有的输入流数据都发送到下游算子的第一个并行子任务中去。这就相当于强行让下游任务并行度变成了1。慎用，可能对程序造成很大的压力！

    1.  // 全局分区
    2.  source.global().print();

    3.4.6 自定义分区（Custom） 

    当Flink提供的所有分区策略都不能满足用户的需求时，我们可以通过使用partitionCustom()方法来自定义分区策略。

    例子：实现将奇数与偶数分配到不同的分区

    自定义分区器实现 Partitioner：

    1. public class MyPartitioner implements Partitioner {
    2.  
    3.     @Override
    4.     public int partition(String key, int numPartitions) {
    5.         // key 为当前数据，numPartitions 为下游并行度
    6.         return Integer.parseInt(key) % numPartitions;
    7.     }
    8. }

    使用自定义分区

    1. public class PartitionCustomDemo {
    2.     
    3.     public static void main(String[] args) throws Exception {
    4.         
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         
    7.         env.setParallelism(2);
    8.  
    9.         DataStreamSource source = env.socketTextStream("IP", 端口);
    10.  
    11.         source.partitionCustom(new MyPartitioner(), v -> v).print();
    12.         
    13.         env.execute();
    14.     }
    15. }

    结果输出：

    1. 输入：
    2. [root@VM-55-24-centos ~]# nc -lk 1234
    3. 2
    4. 3
    5. 4
    6. 6
    7. 8
    8. 10
    9.  
    10. 输出：
    11. 2> 1
    12. 1> 2
    13. 2> 3
    14. 1> 4
    15. 1> 6
    16. 1> 8
    17. 1> 10

    3.5 分流

    所谓“分流”，就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于一个DataStream，定义一些筛选条件，将符合条件的数据拣选出来放到对应的流里。

    与分区不同的是，分流是是将一条数据流拆分成多条流。而分区是将数据分配到下游算子的子任务中。

    3.5.1 Filter 实现分流

    例子：读取一个整数数字流，将数据流划分为奇数流和偶数流。

    1. public class SplitByFilterDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         env.setParallelism(2);
    6.  
    7.         DataStreamSource source = env.socketTextStream("IP", 端口);
    8.  
    9.         source.filter(num -> Integer.parseInt(num) % 2 == 0).print("偶数流：");
    10.  
    11.         source.filter(num -> Integer.parseInt(num) % 2 == 1).print("奇数流：");
    12.  
    13.         env.execute();
    14.     }
    15. }

    输入输出结果：

    1. 输入：
    2. [root@VM-55-24-centos ~]# nc -lk 1234
    3. 1
    4. 2
    5. 45
    6. 324321
    7. 234325235
    8. 12312
    9. 11412
    10.  
    11. 输出：
    12. 奇数流：:1> 1
    13. 偶数流：:2> 2
    14. 奇数流：:1> 45
    15. 偶数流：:2> 324321
    16. 奇数流：:1> 234325235
    17. 偶数流：:2> 12312
    18. 偶数流：:2> 11412

    用 Filter 实现虽然简单但不够高效，因为每次数据流都会经过两次 Filter 过滤 。

     3.5.2 使用侧输出流

    一条未被分类操作的流被称为“主流”，经过分流操作后，侧输出流可以理解为“主流”的“支流”。

    需求：id 为 s1 、s2 的数据被到另外两条侧流 ，非 s1 、s2不受影响，放在主流：

    1. public class SideOutputDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.         env.setParallelism(2);
    6.  
    7.         DataStreamSource source = env.socketTextStream("IP", 端口);
    8.  
    9.         SingleOutputStreamOperator sensorDs = source.map(new MapFunction() {
    10.             @Override
    11.             public WaterSensor map(String value) throws Exception {
    12.                 String[] data = value.split(",");
    13.                 return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
    14.             }
    15.         });
    16.  
    17.         // 侧输出流的标记
    18.         OutputTag s1Tag = new OutputTag<>("s1", Types.POJO(WaterSensor.class));
    19.         OutputTag s2Tag = new OutputTag<>("s2", Types.POJO(WaterSensor.class));
    20.  
    21.         SingleOutputStreamOperator process = sensorDs.process(new ProcessFunction() {
    22.             @Override
    23.             public void processElement(WaterSensor value, ProcessFunction.Context ctx, Collector out) throws Exception {
    24.                 if ("s1".equals(value.getId())) {
    25.                     ctx.output(s1Tag, value);
    26.                 } else if ("s2".equals(value.getId())) {
    27.                     ctx.output(s2Tag, value);
    28.                 } else {
    29.                     out.collect(value);
    30.                 }
    31.             }
    32.         });
    33.  
    34.         // process 默认只会返回主流数据
    35.         process.print("主流");
    36.  
    37.         // 根据输出标签(流的标签)找到 s1 这条支流斌输出
    38.         process.getSideOutput(s1Tag).printToErr("测输出流S1");
    39.  
    40.         // 根据输出标签(流的标签)找到 s2 这条支流斌输出
    41.         process.getSideOutput(s2Tag).printToErr("测输出流S2");
    42.  
    43.         env.execute();
    44.     }
    45. }

    输入与输出结果：

    1. 输入：
    2.  
    3. [root@VM-55-24-centos ~]# nc -lk 1234
    4. s1,1,1
    5. s3,3,3
    6. s2,2,2
    7. s9,9,9
    8. s1,33,1134
    9.  
    10. 输出：
    11.  
    12. 测输出流S1:2> WaterSensor{id='s1', ts=1, vc=1}
    13.  
    14. 主流:1> WaterSensor{id='s3', ts=3, vc=3}
    15.  
    16. 测输出流S2:2> WaterSensor{id='s2', ts=2, vc=2}
    17.  
    18. 主流:1> WaterSensor{id='s9', ts=9, vc=9}
    19.  
    20. 测输出流S1:2> WaterSensor{id='s1', ts=33, vc=1134}
    • Process 算子非常灵活，基础算子底层都是调用 Process 来实现的。
    • OutputTag 可以理解为侧输出流的名称以流的数据类型。
    • 将数据放入侧输出流中需要使用 ctx.output（）传入输出流标签和数据 ;
    • process 返回的流是主流，想获取侧输出流必须通过 process.getSideOutput（）传入输出流标签来获取。

    3.6 基本合流操作

    在实际应用中，我们经常会遇到来源不同的多条流，需要将它们的数据进行联合处理。所以Flink中合流的操作会更加普遍，对应的API也更加丰富。

    3.6.1 联合（Union）

    通过调用数据源的 Union() 就可以将一条或者多条流进行合并。联合操作要求必须流中的数据类型必须相同，合并之后的新流会包括所有流中的元素，数据类型不变。

    1. public class UnionDemo {
    2.  
    3.     /**
    4.      *  Union : 合并一条或多条相同数据类型的流
    5.      */
    6.     public static void main(String[] args) throws Exception {
    7.  
    8.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    9.         
    10.         env.setParallelism(1);
    11.  
    12.         DataStreamSource source1 = env.fromElements(1, 2, 3);
    13.         DataStreamSource source2 = env.fromElements(44, 55, 66);
    14.         DataStreamSource source3 = env.fromElements("777", "888", "999");
    15.  
    16.         // 写法1 ： 一次合并一个流
    17.         DataStream union = source1.union(source2).union(source3.map(Integer::parseInt));
    18.  
    19.         // 写法2 : 一次合并多个流
    20.         source1.union(source2,source3.map(Integer::parseInt));
    21.  
    22.         union.print();
    23.  
    24.         env.execute();
    25.     }
    26. }

    结果输出：

    1. 1
    2. 2
    3. 3
    4. 44
    5. 55
    6. 66
    7. 777
    8. 888
    9. 999

    3.6.2 连接（Connect）

    Union 虽然使用简单，但是受限于只能合并相同类型的流，不太灵活。Flink 提供了另一个更方便的河流操作：连接（Connect）。

    3.6.2.1 连接流（ConnectedStreams）

    通过 Connect 可以将两条不同类型的流进行连接，但是不再返回 DataStream ，而是返回 ConnectedStreams（连接流）。

    且两条流连接后只是形式上的“合并”，对这条流进行处理转换则需要对原本的两条流单独处理。

    1. public class ConnectDemo {
    2.  
    3.     /**
    4.      *  Connect : 连接(合并)两条流
    5.      *      返回的是 ConnectedStreams（连接流） 而不是 DataStream
    6.      *      只是名义上的统一，处理逻辑需要每条流单独处理
    7.      */
    8.     public static void main(String[] args) throws Exception {
    9.  
    10.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    11.         env.setParallelism(1);
    12.  
    13.         DataStreamSource source1 = env.fromElements(1, 2, 3);
    14.         DataStreamSource source2 = env.fromElements("777", "888", "999");
    15.  
    16.         ConnectedStreams connect = source1.connect(source2);
    17.  
    18.         // 需要对两条单独处理 CoMapFunction（第一条流的类型，第二条流的类型，输出的类型）
    19.         SingleOutputStreamOperator map = connect.map(new CoMapFunction() {
    20.             @Override
    21.             public String map1(Integer value) throws Exception {
    22.                 return value.toString();
    23.             }
    24.  
    25.             @Override
    26.             public String map2(String value) throws Exception {
    27.                 return value;
    28.             }
    29.         });
    30.  
    31.         map.print();
    32.  
    33.         env.execute();
    34.     }
    35. }

    结果输出：

    1. 1
    2. 777
    3. 2
    4. 888
    5. 3
    6. 999
    3.6.2.2 CoProcessFunction

    与CoMapFunction类似，如果是调用.map()就需要传入一个CoMapFunction，需要实现map1()、map2()两个方法；而调用.process()时，传入的则是一个CoProcessFunction。它也是“处理函数”家族中的一员，用法非常相似。它需要实现的就是processElement1()、processElement2()两个方法，在每个数据到来时，会根据来源的流调用其中的一个方法进行处理。

    例子：有两条数据类型不同的流，需要根据各自数据的第一个字段进行匹配。类似于 MySQL中的 Inner Join。

    1. public class ConnectKeyByDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.  
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(1);
    7.   
    8.         DataStreamSource> source1 = env.fromElements(
    9.                 Tuple2.of(1, "a1"),
    10.                 Tuple2.of(1, "a2"),
    11.                 Tuple2.of(2, "b"),
    12.                 Tuple2.of(3, "c")
    13.         );
    14.         DataStreamSource> source2 = env.fromElements(
    15.                 Tuple3.of(1, "aa1", 1),
    16.                 Tuple3.of(1, "aa2", 2),
    17.                 Tuple3.of(2, "bb", 1),
    18.                 Tuple3.of(3, "cc", 1)
    19.         );
    20.  
    21.         ConnectedStreams, Tuple3> connect = source1.connect(source2);
    22.  
    23.         SingleOutputStreamOperator process = connect.process(new CoProcessFunction, Tuple3, String>() {
    24.  
    25.             // 为各自两条流定义中间变量用于存储匹配时的数据
    26.             Map>> s1Cache = new HashMap<>();
    27.             Map>> s2Cache = new HashMap<>();
    28.  
    29.             @Override
    30.             public void processElement1(Tuple2 value, CoProcessFunction, Tuple3, String>.Context ctx, Collector out) throws Exception {
    31.                 Integer id = value.f0;
    32.                 // 第一次出现该 Key 则直接将数据put进s1的数据集合中
    33.                 if (!s1Cache.containsKey(id)) {
    34.                     ArrayList> s1Values = new ArrayList<>();
    35.                     s1Values.add(value);
    36.                     s1Cache.put(id, s1Values);
    37.                 } else {
    38.                     // 不是第一次出现该 Key ，直接添加进该 Key 的数组中
    39.                     s1Cache.get(id).add(value);
    40.                 }
    41.  
    42.                 // 去另外一条流的数据中寻找有没有 id 相匹配的，有则放入采集器
    43.                 if (s2Cache.containsKey(id)) {
    44.                     for (Tuple3 s2Element : s2Cache.get(id)) {
    45.                         out.collect("S1:" + value + "<---->" + "s2:" + s2Element);
    46.                     }
    47.                 }
    48.             }
    49.  
    50.             @Override
    51.             public void processElement2(Tuple3 value, CoProcessFunction, Tuple3, String>.Context ctx, Collector out) throws Exception {
    52.                 Integer id = value.f0;
    53.                 // 第一次出现该 Key 则直接将数据put进s2的数据集合中
    54.                 if (!s2Cache.containsKey(id)) {
    55.                     ArrayList> s2Values = new ArrayList<>();
    56.                     s2Values.add(value);
    57.                     s2Cache.put(id, s2Values);
    58.                 } else {
    59.                     // 不是第一次出现该 Key ，直接添加进该 Key 的数组中
    60.                     s2Cache.get(id).add(value);
    61.                 }
    62.  
    63.                 // 去另外一条流的数据中寻找有没有 id 相匹配的，有则放入采集器
    64.                 if (s1Cache.containsKey(id)) {
    65.                     for (Tuple2 s1Element : s1Cache.get(id)) {
    66.                         out.collect("S2:" + value + "<---->" + "s1:" + s1Element);
    67.                     }
    68.                 }
    69.             }
    70.         });
    71.  
    72.         process.print();
    73.  
    74.         env.execute();
    75.     }
    76. }

    结果：

    1. S2:(1,aa1,1)<---->s1:(1,a1)
    2. S1:(1,a2)<---->s2:(1,aa1,1)
    3. S2:(1,aa2,2)<---->s1:(1,a1)
    4. S2:(1,aa2,2)<---->s1:(1,a2)
    5. S2:(2,bb,1)<---->s1:(2,b)
    6. S2:(3,cc,1)<---->s1:(3,c)

    注意：在多并行度下，以上匹配会出错，因为多并行度下，数据会被发往 Process 不同的子任务中（Slot），而不同的子任务间数据无法共享，导致读取不到另一个子任务的数组，从而匹配错误。所以需要在连接流后对要匹配的字段进行 KeyBy 操作，确保同一个 Key 被分配到同一个子任务中。

    四.输出算子（Sink）

    Flink作为数据处理框架，最终还是要把计算处理的结果写入外部存储，为外部应用提供支持。

     4.1 连接到外部系统

    Flink 1.17 中的DataStream API专门提供了向外部写入数据的方法：sinkTo，对应着一个“Sink”算子，主要就是用来实现与外部系统连接、并将数据提交写入的；Flink程序中所有对外的输出操作，一般都是利用Sink算子完成的。

    stream.sinkTo(…)

    在大部分情况下，Sink 并不需要我们手动实现，Flink官方为我们提供了一部分的框架的Sink连接器。如下图所示，列出了Flink官方目前支持的第三方系统连接器：

    及第三方提供的连接器：

     地址：Overview | Apache Flink

    4.2 输出到文件 

    Flink专门提供了一个流式文件系统的连接器：FileSink，为批处理和流处理提供了一个统一的Sink，它可以将分区文件写入Flink支持的文件系统。

    FileSink支持行编码（Row-encoded）和批量编码（Bulk-encoded）格式。这两种不同的方式都有各自的构建器（builder），可以直接调用FileSink的静态方法：

    例子：使用数据生成器源源不断生成数据，并输出到文件夹的文本文件中，每隔一个小时生成一个新的文件夹，且每隔20秒或者文件大小达到 3KB 则新建一个文本文件。

    导入依赖：

    1. <dependency>
    2.     <groupId>org.apache.flinkgroupId>
    3.     <artifactId>flink-connector-filesartifactId>
    4.     <version>${flink.version}version>
    5. dependency>
    1. public class SinkFileDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.  
    5.         // 获取执行环境
    6.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    7.  
    8.         // 全局并行度设置为 2
    9.         env.setParallelism(2);
    10.  
    11.         // 开启checkpoint
    12.         env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
    13.  
    14.         /**
    15.          * 数据生成器:无限生成数字，一秒生成 1000 条
    16.          */
    17.         DataGeneratorSource dataGenSource = new DataGeneratorSource<>(
    18.                 (GeneratorFunction) num -> "Number:" + num,
    19.                 Long.MIN_VALUE,
    20.                 RateLimiterStrategy.perSecond(1000),
    21.                 Types.STRING
    22.         );
    23.  
    24.         DataStreamSource streamSource = env.fromSource(dataGenSource, WatermarkStrategy.noWatermarks(),"data-generator");
    25.  
    26.         /**
    27.          * 输出到文件系统
    28.          *      Sink 算子同样会受到 并行度 的影响：例如会同时有 并行度个 个文件被写入
    29.          */
    30.         FileSink fileSink = FileSink
    31.                 // 指定要输出的 文件目录 及 文件编码
    32.                 .forRowFormat(new Path("D:/tmp"), new SimpleStringEncoder<>("UTF-8"))
    33.  
    34.                 // 指定要生成文件的 前后缀
    35.                 .withOutputFileConfig(
    36.                         OutputFileConfig.builder() // 建造者模式
    37.                                 // 文件的前缀
    38.                                 .withPartPrefix("flink-file-test")
    39.                                 // 文件的后缀
    40.                                 .withPartSuffix(".txt")
    41.                                 .build()
    42.                 )
    43.  
    44.                 // 指定目录分桶：按照小时进行分桶（一小时生成一个新的目录），并设置时区为 Asia/Shanghai
    45.                 .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd--HH", ZoneId.of("Asia/Shanghai")))
    46.  
    47.                 // 文件滚动策略：每隔多少秒 或 文件超过多大 就生成新的文件
    48.                 .withRollingPolicy(
    49.                         DefaultRollingPolicy.builder() // 建造者模式
    50.                                 // 每隔 20S 生成一个新的文件
    51.                                 .withRolloverInterval(Duration.ofSeconds(20))
    52.                                 // 文件大小超过大于 3KB 则生成一个新的文件
    53.                                 .withMaxPartSize(new MemorySize(1024 * 3))
    54.                                 .build()
    55.                 ).build();
    56.  
    57.         // 输出
    58.         streamSource.sinkTo(fileSink);
    59.  
    60.         // 执行
    61.         env.execute();
    62.     }
    63.  
    64. }

     结果：

    FileSink

            .forRowFormat：指定要输出的文件目录及文件编码 

            .withOutputFileConfig：指定要生成文件的前后缀

            .withBucketAssigner：指定目录分桶

            .withRollingPolicy：文件滚动策略

    4.3 输出到RabbitMQ

    想要输出到 RabbitMQ，也需要调用对应的 Sink 算子--RMQSink 。 

    例子：从 Socket 读数据，写入到 RabbitMQ 中，作为一条消息。

    添加Kafka 连接器依赖:

    2. <dependency>
    3.     <groupId>org.apache.flinkgroupId>
    4.     <artifactId>flink-connector-rabbitmqartifactId>
    5.     <version>3.0.1-1.17version>
    6. dependency>
    1. public class SinkRabbitMqDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    5.  
    6.         env.setParallelism(1);
    7.  
    8.         env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
    9.  
    10.         DataStreamSource streamSource = env.socketTextStream("ip", 1234);
    11.  
    12.         // 配置 RabbitMQ 连接信息
    13.         RMQConnectionConfig mqConfig = new RMQConnectionConfig.Builder()
    14.                 .setHost("xxx.xxx.xxx.xxx") // RabbitMQ 服务地址
    15.                 .setPort(5379) // RabbitMQ 服务端口
    16.                 .setUserName("用户名") // 用户名
    17.                 .setPassword("密码") // 密码
    18.                 .setVirtualHost("/") // 虚拟主机名
    19.                 .build();
    20.  
    21.         // 创建一个RMQSink，用于将数据发送到RabbitMQ队列
    22.         // mq配置信息，队列名称，序列化器
    23.         RMQSink rmqSink = new RMQSink(mqConfig, "test_queue", new SimpleStringSchema());
    24.  
    25.         // 将数据流写入RabbitMQ队列 Flink 1.17 并不支持使用 sinkTo 对第三方系统进行输出
    26.         streamSource.addSink(rmqSink);
    27.  
    28.         env.execute("flink connectors rabbitmq");
    29.     }
    30. }

    输入：

    1. [root@VM-55-24-centos ~]# nc -lk 1234
    2. hello flink
    3. hello rabbitmq

    结果：

    4.4 输出到MySQL（JDBC） 

    同样的，要输出到 MySQL ，需要调用 JdbcSink.sink() 算子，且也只能使用 addSink 来添加输出。

    例子： 在 Socket 中写入数据，写入MySQL中。

    在 MySQL 中新建表：

    1. CREATE TABLE `ws` (
    2.   `id` varchar(100) NOT NULL,
    3.   `ts` bigint(20) DEFAULT NULL,
    4.   `vc` int(11) DEFAULT NULL,
    5.   PRIMARY KEY (`id`)
    6. ) ENGINE=InnoDB DEFAULT CHARSET=utf8

    导入 MySQL 驱动：

    2. <dependency>
    3.     <groupId>mysqlgroupId>
    4.     <artifactId>mysql-connector-javaartifactId>
    5.     <version>5.1.47version>
    6. dependency>

    导入 Flink - MySQL 连接器:

    2. <dependency>
    3.     <groupId>org.apache.flinkgroupId>
    4.     <artifactId>flink-connector-jdbcartifactId>
    5.     <version>3.1.1-1.17version>
    6. dependency>

    代码：

    1. public class SinkMySQLDemo {
    2.  
    3.     public static void main(String[] args) throws Exception {
    4.  
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.  
    7.         env.setParallelism(1);
    8.  
    9.         DataStreamSource streamSource = env.socketTextStream("xxx.xxx.xxx.xxx", 1234);
    10.  
    11.         // 将从 Socket 中读到的字符串转成实体类
    12.         SingleOutputStreamOperator map = streamSource.map((MapFunction) s -> {
    13.             String[] data = s.split(",");
    14.             return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
    15.         });
    16.  
    17.         /**
    18.          * jdbcSink 四大参数：
    19.          * 1、要执行的 SQL 语句
    20.          * 2、为占位符填充值
    21.          * 3、执行选项：重试次数，攒批
    22.          * 4、MySQL 连接信息
    23.          */
    24.         SinkFunction jdbcSink = JdbcSink.sink(
    25.                 "insert into ws values( ? , ? , ?)",
    26.                 new JdbcStatementBuilder() {
    27.                     @Override
    28.                     public void accept(PreparedStatement preparedStatement, WaterSensor waterSensor) throws SQLException {
    29.                         preparedStatement.setString(1, waterSensor.getId());
    30.                         preparedStatement.setLong(2, waterSensor.getTs());
    31.                         preparedStatement.setInt(3, waterSensor.getVc());
    32.                     }
    33.                 },
    34.                 JdbcExecutionOptions.builder()
    35.                         .withBatchIntervalMs(3000) // 批次的时间
    36.                         .withBatchSize(100) // 批次的大小：条数
    37.                         .withMaxRetries(3) // 重试次数
    38.                         .build(),
    39.                 new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
    40.                         .withUrl("jdbc:mysql://localhost:3306/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8")
    41.                         .withUsername("Username")
    42.                         .withPassword("Password")
    43.                         .withConnectionCheckTimeoutSeconds(60) // 重试的超时时间
    44.                         .build()
    45.         );
    46.  
    47.         map.addSink(jdbcSink);
    48.  
    49.         env.execute("flink connectors MySQL");
    50.     }
    51. }

    输入：

    1. [root@VM-55-24-centos ~]# nc -lk 1234
    2. hello,1,1
    3. flink,2,2
    4. mysql,3,3

    输出：

     4.5 自定义Sink输出

    Flink 为我们提供很多常用的连接器，一般不推荐自定义Sink，因为需要自行处理连接逻辑及错误逻辑。

    如果要自定义Sink，Flink 为我们提供了通用的SinkFunction接口和对应的RichSinkDunction抽象类，只要实现它，通过简单地调用DataStream的.addSink()方法就可以自定义写入任何外部存储。

    streamSource.addSink(new MySink());

    推荐继承RichSinkDunction，实现其中的三个方法open()、close()、invoke(String value, Context context)。

    1. public class  MySink extends RichSinkFunction{
    2.     @Override
    3.     public void open(Configuration parameters) throws Exception {
    4.         // 启动时会被调用一次
    5.         // 可以在这里创建连接
    6.     }
    7.  
    8.     @Override
    9.     public void close() throws Exception {
    10.         // 销毁时会被调用一次
    11.         // 可以在这里销毁连接
    12.     }
    13.  
    14.     // Sink 的核心逻辑
    15.     @Override
    16.     public void invoke(String value, Context context) throws Exception {
    17.         // 每条数据来都会调用一次
    18.         // 具体的写入逻辑...
    19.     }
    20. }
  • 相关阅读:
    数据库设计中如何选择主键 （自然键或代理键 ）| Part 1
    2022年电工杯数学建模B题5G网络环境下应急物资配送问题求解全过程论文及程序
    Android视角看鸿蒙第四课(module.json中的各字段含义之description&mainElement)修改程序入口
    Spring整合RabbitMQ——生产者（利用配置类）
    纯手码优质JAVA面试八股文
    给一个数组赋1到10的初始值（指针）
    AI芯片的性能评价
    C认证笔记 - Web基础 - 知识点1：语义化标签
    前端性能优化方法与实战08 诊断清单：如何实现监控预警并进行问题诊断
    数据结构入门-13-图
  • 原文地址：https://blog.csdn.net/weixin_53922163/article/details/133975274