Flink1.17 DataStream API

2023-12-13 08:42:15

目录

一.执行环境(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类的对象(流处理,批处理已经标记为过时),创建执行环境一般有以下三种方式:

// 创建一个本地执行环境并返回,可传入并行度,默认是本地CPU核心数
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();


// 返回远程集群执行环境,需传入远程IobManager的主机名与端口、及在集群中需运行的Jar包
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("JobManager ip", "JobManager port","提交给JobManager的JAR包");


// (推荐)根据当前环境自动选择执行环境,无脑选这个即可
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

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

1.2 执行模式

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

通过代码指定:

// 流 执行模式
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
// 批 执行模式
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
// 自动模式,根据数据源是否有界自动选择执行模式
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

提交任务时命令行指定(推荐):

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

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

1.3?触发程序执行

// 程序执行
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<String> stream = env.fromSource(…)

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

2.1 从集合中读取数据

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

 public static void main(String[] args) throws Exception {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 从集合中读取数据
        DataStreamSource<Integer> source = env.fromCollection(Arrays.asList(1, 10, 99, 53));

        source.print();

        env.execute();

    }

输出结果:

6> 1
8> 99
1> 53
7> 10

2.2?从文件读取数据

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

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

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-files</artifactId>
    <version>${flink.version}</version>
</dependency>

?代码如下:

public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("input/words.txt")).build();

        // !使用官方推荐的新的 Source 架构  => env.fromSource(Source实现类,Watermark,资源名称)
        env.fromSource(fileSource,WatermarkStrategy.noWatermarks(),"file").print();

        env.execute();

    }

输出结果:

3> hello flink
3> hello world
3> hello java

2.3 从 RabbitMQ 中读取数据

导入相关依赖:

<!--RabbitMQ 连接器-->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-rabbitmq</artifactId>
    <version>3.0.1-1.17</version>
</dependency>
<!--amqp 客户端-->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.14.1</version>
</dependency>

相关代码:

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

/**
 * 从 RabbitMQ读取数据
 */
public class RabbitMQSourceDemo {

    public static void main(String[] args) throws Exception {

        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 RabbitMQ 连接信息
        RMQConnectionConfig config = new RMQConnectionConfig.Builder()
                .setHost("RabbitMQ服务器地址")
                .setPort(RabbitMQ端口)
                .setUserName(用户名)
                .setPassword(密码)
                .setVirtualHost(虚拟主机)
                .build();

        // 添加 RabbitMQ 数据源(Flink 1.17 并不支持使用 env.fromSource 在 RabbitMQ 读取数据!)
        RMQSource<String> source = new RMQSource<>(
                config,                     //  连接配置
                "test_queue",               // 队列名称
                new SimpleStringSchema());  // 反序列化器

        // 添加数据源
        DataStreamSource<String> rabbitMQStream = env.addSource(source);

        // 打印
        rabbitMQStream.print();
        
        // 执行
        env.execute("RabbitMQ job");
    }
}

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

输出结果:

2.4?从数据生成器读取数据?

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

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-datagen</artifactId>
    <version>${flink.version}</version>
</dependency>

?代码:

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Flink 数据生成器
 */
public class DataGeneratorDemo {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 数据生成器的四个参数:
         * 1、GeneratorFunction的map实现。重写返回值
         * 2、返回的个数 会从0开始依次返回(使用Long.MAX_VALUE可模拟出无界流)
         * 3、限速,每秒多少个数据
         * 4、返回值类型
         */
        DataGeneratorSource<String> dataGeneratorSource = new DataGeneratorSource<>(
                (GeneratorFunction<Long, String>) num -> "Number:" + num,
                30,
                RateLimiterStrategy.perSecond(3),
                Types.STRING
        );

        env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(),"datagenerator-source").print();

        // 执行
        env.execute();
    }
}

输出:

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

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还具有一个类型提取系统,可以分析函数的输入和返回类型,自动获取类型信息,从而获得对应的序列化器和反序列化器。例如:

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

可写作:

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

三.转换算子(Transformation)

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

3.1 基本转换算子(map/ filter/ flatMap

准备工作

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

字段名

数据类型

说明

id

String

水位传感器类型

ts

Long

传感器记录时间戳

vc

Integer

水位记录

代码如下:

public class WaterSensor {
    public String id;
    public Long ts;
    public Integer vc;

    // 省略getter、setter、构造器、toString
}

3.1.1 映射(map)

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

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

public class MapDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 构造数据
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("id_1", 1l, 1),
                new WaterSensor("id_2", 2l, 2),
                new WaterSensor("id_3", 3l, 3)
        );

        // 方法1:实现匿名内部类
        source.map(new MapFunction<WaterSensor, String>() {
            @Override
            public String map(WaterSensor waterSensor) throws Exception {
                return waterSensor.getId();
            }
        }).print();


        // 方法2:Lambda 表达式
        source.map(WaterSensor::getId).print();

        // 方法三:定义 MapFunction 实现类
        source.map(new MyMapFunction()).print();
        env.execute();
    }

    // 实现 MapFunction , 可以复用
    static class MyMapFunction implements MapFunction<WaterSensor , String>{
        @Override
        public String map(WaterSensor waterSensor) throws Exception {
            return waterSensor.getId();
        }
    }
}

结果输出:

2> id_3
8> id_1
1> id_2

3.1.2?过滤(filter)

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

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

public class FilterDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 构造数据
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("id_1", 1l, 1),
                new WaterSensor("id_1", 12l, 2),
                new WaterSensor("id_2", 2l, 2),
                new WaterSensor("id_3", 3l, 3)
        );

        // 过滤数据中 id 不为 id_1 的元素
        source.filter(new FilterFunction<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor waterSensor) throws Exception {
                return "id_1".equals(waterSensor.getId());
            }
        }).print();

        env.execute();
    }

}

结果输出:

3> WaterSensor{id='id_1', ts=1, vc=1}
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 属性。

public class FlatmapDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 构造数据
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("id_1", 1l, 1),
                new WaterSensor("id_1", 12l, 2),
                new WaterSensor("id_2", 2l, 2),
                new WaterSensor("id_3", 3l, 3)
        );

        /**
         * 如果 id 为 id_1 则输出 vc 属性
         * 如果 id 为 id_2 则输出 ts、vc 属性
         */
        source.flatMap(new FlatMapFunction<WaterSensor, String>() {
            @Override
            public void flatMap(WaterSensor waterSensor, Collector<String> collector) throws Exception {
                if("id_1".equals(waterSensor.getId())){
                    // 将 vc 放入采集器
                    collector.collect(waterSensor.getVc().toString());
                } else if ("id_2".equals(waterSensor.getId())) {
                    // 将 ts、vc 放入采集器
                    collector.collect(waterSensor.getVc().toString());
                    collector.collect(waterSensor.getTs().toString());
                }
            }
        }).print();

        env.execute();
    }
}

结果输出:

2> 1
3> 2
4> 2
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?进行分区:

public class KeyByDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 构造数据
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("id_1", 1l, 1),
                new WaterSensor("id_1", 12l, 2),
                new WaterSensor("id_2", 2l, 2),
                new WaterSensor("id_3", 3l, 3)
        );

       // 以 id 为 Key 进行分区
        source.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor waterSensor) throws Exception {
                return waterSensor.getId();
            }
        }).print();

        env.execute();
    }
    
}

?结果输出:

2> WaterSensor{id='id_2', ts=2, vc=2}
3> WaterSensor{id='id_1', ts=1, vc=1}
3> WaterSensor{id='id_1', ts=12, vc=2}
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 只会保留第一次的非比较字段。

例子:

public class AggrDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        // 构造数据
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("id_1", 1l, 1),
                new WaterSensor("id_1", 12l, 22),
                new WaterSensor("id_2", 2l, 2),
                new WaterSensor("id_3", 3l, 3)
        );

       // 以 id 为 Key 进行分区
        KeyedStream<WaterSensor, String> keyBySource = source.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor waterSensor) throws Exception {
                return waterSensor.getId();
            }
        });


        keyBySource.sum("vc").print();
        keyBySource.min("vc").print();
        
        keyBySource.max("vc").print();
        /**
         * max结果:
         * 1> WaterSensor{id='id_1', ts=1, vc=1}
         * 1> WaterSensor{id='id_1', ts=1, vc=22}
         * 1> WaterSensor{id='id_2', ts=2, vc=2}
         * 1> WaterSensor{id='id_3', ts=3, vc=3}
         * ts 还是 第一次的值
         */

        keyBySource.maxBy("vc").print();
        /**
         * max结果:
         * 1> WaterSensor{id='id_1', ts=1, vc=1}
         * 1> WaterSensor{id='id_1', ts=12, vc=22}
         * 1> WaterSensor{id='id_2', ts=2, vc=2}
         * 1> WaterSensor{id='id_3', ts=3, vc=3}
         * 取当前整列值 
         */

        env.execute();
    }

}

3.2.3?归约聚合(reduce)

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

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

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

public class ReduceDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 构造数据
        DataStreamSource<WaterSensor> source = env.fromElements(
                new WaterSensor("id_1", 1l, 1),
                new WaterSensor("id_1", 21l, 21),
                new WaterSensor("id_1", 31l, 31),
                new WaterSensor("id_2", 2l, 2),
                new WaterSensor("id_3", 3l, 3)
        );

        // 以 id 为 Key 进行分区
        KeyedStream<WaterSensor, String> sensorKs = source.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor waterSensor) throws Exception {
                return waterSensor.getId();
            }
        });

        /**
         * reduce:
         *  1.必须在KeyBy后调用
         *  2.输入类型 = 输出类型
         *  3.每个分区的第一条数据来的时候不会执行reduce,但是会存起来保存状态,直接输出,,“Flink有状态的体现”
         *  4.reduce( value1,  value2)
         *     a.value1 是上一次的计算结果
         *     b.value2 是当前进入的数据
         */
        SingleOutputStreamOperator<WaterSensor> reduce = sensorKs.reduce(new ReduceFunction<WaterSensor>() {
            @Override
            public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                if(value2.getVc() > value1.getVc()){
                    return new WaterSensor(value2.getId(), value2.getTs(), value2.getVc());
                }else {
                    return value1;
                }

            }
        });

        reduce.print();

        env.execute();
    }
}

结果输出:

WaterSensor{id='id_1', ts=1, vc=1} // 分组的第一条数据直接返回
WaterSensor{id='id_1', ts=21, vc=21}
WaterSensor{id='id_1', ts=31, vc=31}
WaterSensor{id='id_2', ts=2, vc=2} // 分组的第一条数据直接返回
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等。所以用户可以自定义一个函数类,实现对应的接口。

匿名内部类实现

 source.filter(new FilterFunction<WaterSensor>() {
            @Override
            public boolean filter(WaterSensor waterSensor) throws Exception {
                return "id_1".equals(waterSensor.getId());
            }
        }).print();

Lambda表达式实现:

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

实现 XxxFunction 接口

public class FilterFunctionImpl implements FilterFunction<WaterSensor> {

    public String id ;

    public FilterFunctionImpl(String id) {
        this.id = id;
    }

    @Override
    public boolean filter(WaterSensor waterSensor) throws Exception {
        return this.id.equals(waterSensor.getId());
    }
}

3.3.2?富函数类(Rich Function Classes)

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

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

典型的生命周期方法有:

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

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

public class RichMapFunctionDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4);

        SingleOutputStreamOperator<Integer> map = source.map(new RichMapFunction<Integer, Integer>() {

            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("open:子任务名称"+getRuntimeContext().getTaskNameWithSubtasks());
                System.out.println("open:子任务编号"+getRuntimeContext().getIndexOfThisSubtask());
                super.open(parameters);

            }

            @Override
            public void close() throws Exception {
                System.out.println("close:子任务名称"+getRuntimeContext().getTaskNameWithSubtasks());
                System.out.println("close:子任务编号"+getRuntimeContext().getIndexOfThisSubtask());
                super.close();

            }

            @Override
            public Integer map(Integer value) throws Exception {
                return value + 1;
            }
        });

        map.print();

        env.execute();
    }
}

输出结果:

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

3.4 物理分区算子(Physical Partitioning

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

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

3.4.1?随机分区(shuffle)

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

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

public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> source = env.socketTextStream("ip", 端口);

        // 随机分区 random.nextInt(下游算子并行度)
        source.shuffle().print();

        env.execute();
    }
输入:
[root@VM-55-24-centos ~]# nc -lk 1234
1
2
3
4
5

输出:
1> 1
2> 2
2> 3
1> 4
1> 5

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

3.4.2?轮询分区(Round-Robin)

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

 // 轮询重分区
 source.rebalance().print();
输入:
[root@VM-55-24-centos ~]# nc -lk 1234
1
2
3
4
5

输出:
1> 1
2> 2
1> 3
2> 4
1> 5

3.4.3?重缩放分区(rescale)

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

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

?3.4.4?广播(broadcast)

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

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

3.4.5?全局分区(global)

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

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

3.4.6?自定义分区(Custom)?

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

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

自定义分区器实现 Partitioner:

public class MyPartitioner implements Partitioner<String> {

    @Override
    public int partition(String key, int numPartitions) {
        // key 为当前数据,numPartitions 为下游并行度
        return Integer.parseInt(key) % numPartitions;
    }
}

使用自定义分区

public class PartitionCustomDemo {
    
    public static void main(String[] args) throws Exception {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.setParallelism(2);

        DataStreamSource<String> source = env.socketTextStream("IP", 端口);

        source.partitionCustom(new MyPartitioner(), v -> v).print();
        
        env.execute();
    }
}

结果输出:

输入:
[root@VM-55-24-centos ~]# nc -lk 1234
2
3
4
6
8
10

输出:
2> 1
1> 2
2> 3
1> 4
1> 6
1> 8
1> 10

3.5?分流

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

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

3.5.1 Filter 实现分流

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

public class SplitByFilterDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> source = env.socketTextStream("IP", 端口);

        source.filter(num -> Integer.parseInt(num) % 2 == 0).print("偶数流:");

        source.filter(num -> Integer.parseInt(num) % 2 == 1).print("奇数流:");

        env.execute();
    }
}

输入输出结果:

输入:
[root@VM-55-24-centos ~]# nc -lk 1234
1
2
45
324321
234325235
12312
11412

输出:
奇数流::1> 1
偶数流::2> 2
奇数流::1> 45
偶数流::2> 324321
奇数流::1> 234325235
偶数流::2> 12312
偶数流::2> 11412

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

?3.5.2 使用侧输出流

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

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

public class SideOutputDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> source = env.socketTextStream("IP", 端口);

        SingleOutputStreamOperator<WaterSensor> sensorDs = source.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] data = value.split(",");
                return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
            }
        });

        // 侧输出流的标记
        OutputTag<WaterSensor> s1Tag = new OutputTag<>("s1", Types.POJO(WaterSensor.class));
        OutputTag<WaterSensor> s2Tag = new OutputTag<>("s2", Types.POJO(WaterSensor.class));

        SingleOutputStreamOperator<WaterSensor> process = sensorDs.process(new ProcessFunction<WaterSensor, WaterSensor>() {
            @Override
            public void processElement(WaterSensor value, ProcessFunction<WaterSensor, WaterSensor>.Context ctx, Collector<WaterSensor> out) throws Exception {
                if ("s1".equals(value.getId())) {
                    ctx.output(s1Tag, value);
                } else if ("s2".equals(value.getId())) {
                    ctx.output(s2Tag, value);
                } else {
                    out.collect(value);
                }
            }
        });

        // process 默认只会返回主流数据
        process.print("主流");

        // 根据输出标签(流的标签)找到 s1 这条支流斌输出
        process.getSideOutput(s1Tag).printToErr("测输出流S1");

        // 根据输出标签(流的标签)找到 s2 这条支流斌输出
        process.getSideOutput(s2Tag).printToErr("测输出流S2");

        env.execute();
    }
}

输入与输出结果:

输入:

[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s3,3,3
s2,2,2
s9,9,9
s1,33,1134

输出:

测输出流S1:2> WaterSensor{id='s1', ts=1, vc=1}

主流:1> WaterSensor{id='s3', ts=3, vc=3}

测输出流S2:2> WaterSensor{id='s2', ts=2, vc=2}

主流:1> WaterSensor{id='s9', ts=9, vc=9}

测输出流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() 就可以将一条或者多条流进行合并。联合操作要求必须流中的数据类型必须相同,合并之后的新流会包括所有流中的元素,数据类型不变。

public class UnionDemo {

    /**
     *  Union : 合并一条或多条相同数据类型的流
     */
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.setParallelism(1);

        DataStreamSource<Integer> source1 = env.fromElements(1, 2, 3);
        DataStreamSource<Integer> source2 = env.fromElements(44, 55, 66);
        DataStreamSource<String> source3 = env.fromElements("777", "888", "999");

        // 写法1 : 一次合并一个流
        DataStream<Integer> union = source1.union(source2).union(source3.map(Integer::parseInt));

        // 写法2 : 一次合并多个流
        source1.union(source2,source3.map(Integer::parseInt));

        union.print();

        env.execute();
    }
}

结果输出:

1
2
3
44
55
66
777
888
999

3.6.2?连接(Connect)

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

3.6.2.1?连接流(ConnectedStreams

通过 Connect 可以将两条不同类型的流进行连接,但是不再返回?DataStream ,而是返回?ConnectedStreams(连接流)。

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

public class ConnectDemo {

    /**
     *  Connect : 连接(合并)两条流
     *      返回的是 ConnectedStreams(连接流) 而不是 DataStream
     *      只是名义上的统一,处理逻辑需要每条流单独处理
     */
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Integer> source1 = env.fromElements(1, 2, 3);
        DataStreamSource<String> source2 = env.fromElements("777", "888", "999");

        ConnectedStreams<Integer, String> connect = source1.connect(source2);

        // 需要对两条单独处理 CoMapFunction(第一条流的类型,第二条流的类型,输出的类型)
        SingleOutputStreamOperator<String> map = connect.map(new CoMapFunction<Integer, String, String>() {
            @Override
            public String map1(Integer value) throws Exception {
                return value.toString();
            }

            @Override
            public String map2(String value) throws Exception {
                return value;
            }
        });

        map.print();

        env.execute();
    }
}

结果输出:

1
777
2
888
3
999
3.6.2.2?CoProcessFunction

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

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

public class ConnectKeyByDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
  
        DataStreamSource<Tuple2<Integer, String>> source1 = env.fromElements(
                Tuple2.of(1, "a1"),
                Tuple2.of(1, "a2"),
                Tuple2.of(2, "b"),
                Tuple2.of(3, "c")
        );
        DataStreamSource<Tuple3<Integer, String, Integer>> source2 = env.fromElements(
                Tuple3.of(1, "aa1", 1),
                Tuple3.of(1, "aa2", 2),
                Tuple3.of(2, "bb", 1),
                Tuple3.of(3, "cc", 1)
        );

        ConnectedStreams<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>> connect = source1.connect(source2);

        SingleOutputStreamOperator<String> process = connect.process(new CoProcessFunction<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>() {

            // 为各自两条流定义中间变量用于存储匹配时的数据
            Map<Integer, List<Tuple2<Integer, String>>> s1Cache = new HashMap<>();
            Map<Integer, List<Tuple3<Integer, String, Integer>>> s2Cache = new HashMap<>();

            @Override
            public void processElement1(Tuple2<Integer, String> value, CoProcessFunction<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>.Context ctx, Collector<String> out) throws Exception {
                Integer id = value.f0;
                // 第一次出现该 Key 则直接将数据put进s1的数据集合中
                if (!s1Cache.containsKey(id)) {
                    ArrayList<Tuple2<Integer, String>> s1Values = new ArrayList<>();
                    s1Values.add(value);
                    s1Cache.put(id, s1Values);
                } else {
                    // 不是第一次出现该 Key ,直接添加进该 Key 的数组中
                    s1Cache.get(id).add(value);
                }

                // 去另外一条流的数据中寻找有没有 id 相匹配的,有则放入采集器
                if (s2Cache.containsKey(id)) {
                    for (Tuple3<Integer, String, Integer> s2Element : s2Cache.get(id)) {
                        out.collect("S1:" + value + "<---->" + "s2:" + s2Element);
                    }
                }
            }

            @Override
            public void processElement2(Tuple3<Integer, String, Integer> value, CoProcessFunction<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>.Context ctx, Collector<String> out) throws Exception {
                Integer id = value.f0;
                // 第一次出现该 Key 则直接将数据put进s2的数据集合中
                if (!s2Cache.containsKey(id)) {
                    ArrayList<Tuple3<Integer, String, Integer>> s2Values = new ArrayList<>();
                    s2Values.add(value);
                    s2Cache.put(id, s2Values);
                } else {
                    // 不是第一次出现该 Key ,直接添加进该 Key 的数组中
                    s2Cache.get(id).add(value);
                }

                // 去另外一条流的数据中寻找有没有 id 相匹配的,有则放入采集器
                if (s1Cache.containsKey(id)) {
                    for (Tuple2<Integer, String> s1Element : s1Cache.get(id)) {
                        out.collect("S2:" + value + "<---->" + "s1:" + s1Element);
                    }
                }
            }
        });

        process.print();

        env.execute();
    }
}

结果:

S2:(1,aa1,1)<---->s1:(1,a1)
S1:(1,a2)<---->s2:(1,aa1,1)
S2:(1,aa2,2)<---->s1:(1,a1)
S2:(1,aa2,2)<---->s1:(1,a2)
S2:(2,bb,1)<---->s1:(2,b)
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 则新建一个文本文件。

导入依赖:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-files</artifactId>
    <version>${flink.version}</version>
</dependency>
public class SinkFileDemo {

    public static void main(String[] args) throws Exception {

        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 全局并行度设置为 2
        env.setParallelism(2);

        // 开启checkpoint
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        /**
         * 数据生成器:无限生成数字,一秒生成 1000 条
         */
        DataGeneratorSource<String> dataGenSource = new DataGeneratorSource<>(
                (GeneratorFunction<Long, String>) num -> "Number:" + num,
                Long.MIN_VALUE,
                RateLimiterStrategy.perSecond(1000),
                Types.STRING
        );

        DataStreamSource<String> streamSource = env.fromSource(dataGenSource, WatermarkStrategy.noWatermarks(),"data-generator");

        /**
         * 输出到文件系统
         *      Sink 算子同样会受到 并行度 的影响:例如会同时有 并行度个 个文件被写入
         */
        FileSink<String> fileSink = FileSink
                // 指定要输出的 文件目录 及 文件编码
                .<String>forRowFormat(new Path("D:/tmp"), new SimpleStringEncoder<>("UTF-8"))

                // 指定要生成文件的 前后缀
                .withOutputFileConfig(
                        OutputFileConfig.builder() // 建造者模式
                                // 文件的前缀
                                .withPartPrefix("flink-file-test")
                                // 文件的后缀
                                .withPartSuffix(".txt")
                                .build()
                )

                // 指定目录分桶:按照小时进行分桶(一小时生成一个新的目录),并设置时区为 Asia/Shanghai
                .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd--HH", ZoneId.of("Asia/Shanghai")))

                // 文件滚动策略:每隔多少秒 或 文件超过多大 就生成新的文件
                .withRollingPolicy(
                        DefaultRollingPolicy.builder() // 建造者模式
                                // 每隔 20S 生成一个新的文件
                                .withRolloverInterval(Duration.ofSeconds(20))
                                // 文件大小超过大于 3KB 则生成一个新的文件
                                .withMaxPartSize(new MemorySize(1024 * 3))
                                .build()
                ).build();

        // 输出
        streamSource.sinkTo(fileSink);

        // 执行
        env.execute();
    }

}

?结果:

FileSink

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

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

? ? ? ? .withBucketAssigner:指定目录分桶

? ? ? ? .withRollingPolicy:文件滚动策略

4.3?输出到RabbitMQ

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

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

添加Kafka 连接器依赖:

<!--RabbitMQ 连接器-->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-rabbitmq</artifactId>
    <version>3.0.1-1.17</version>
</dependency>
public class SinkRabbitMqDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        DataStreamSource<String> streamSource = env.socketTextStream("ip", 1234);

        // 配置 RabbitMQ 连接信息
        RMQConnectionConfig mqConfig = new RMQConnectionConfig.Builder()
                .setHost("xxx.xxx.xxx.xxx") // RabbitMQ 服务地址
                .setPort(5379) // RabbitMQ 服务端口
                .setUserName("用户名") // 用户名
                .setPassword("密码") // 密码
                .setVirtualHost("/") // 虚拟主机名
                .build();

        // 创建一个RMQSink,用于将数据发送到RabbitMQ队列
        // mq配置信息,队列名称,序列化器
        RMQSink rmqSink = new RMQSink(mqConfig, "test_queue", new SimpleStringSchema());

        // 将数据流写入RabbitMQ队列 Flink 1.17 并不支持使用 sinkTo 对第三方系统进行输出
        streamSource.addSink(rmqSink);

        env.execute("flink connectors rabbitmq");
    }
}

输入:

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

结果:

4.4?输出到MySQL(JDBC)?

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

例子: 在?Socket 中写入数据,写入MySQL中。

在 MySQL 中新建表:

CREATE TABLE `ws` (
  `id` varchar(100) NOT NULL,
  `ts` bigint(20) DEFAULT NULL,
  `vc` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

导入 MySQL 驱动:

<!-- MySQL 驱动-->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.47</version>
</dependency>

导入 Flink - MySQL 连接器:

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc</artifactId>
    <version>3.1.1-1.17</version>
</dependency>

代码:

public class SinkMySQLDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> streamSource = env.socketTextStream("xxx.xxx.xxx.xxx", 1234);

        // 将从 Socket 中读到的字符串转成实体类
        SingleOutputStreamOperator<WaterSensor> map = streamSource.map((MapFunction<String, WaterSensor>) s -> {
            String[] data = s.split(",");
            return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
        });

        /**
         * jdbcSink 四大参数:
         * 1、要执行的 SQL 语句
         * 2、为占位符填充值
         * 3、执行选项:重试次数,攒批
         * 4、MySQL 连接信息
         */
        SinkFunction<WaterSensor> jdbcSink = JdbcSink.sink(
                "insert into ws values( ? , ? , ?)",
                new JdbcStatementBuilder<WaterSensor>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, WaterSensor waterSensor) throws SQLException {
                        preparedStatement.setString(1, waterSensor.getId());
                        preparedStatement.setLong(2, waterSensor.getTs());
                        preparedStatement.setInt(3, waterSensor.getVc());
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchIntervalMs(3000) // 批次的时间
                        .withBatchSize(100) // 批次的大小:条数
                        .withMaxRetries(3) // 重试次数
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost:3306/test?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8")
                        .withUsername("Username")
                        .withPassword("Password")
                        .withConnectionCheckTimeoutSeconds(60) // 重试的超时时间
                        .build()
        );

        map.addSink(jdbcSink);

        env.execute("flink connectors MySQL");
    }
}

输入:

[root@VM-55-24-centos ~]# nc -lk 1234
hello,1,1
flink,2,2
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)。

public class  MySink extends RichSinkFunction<String>{
    @Override
    public void open(Configuration parameters) throws Exception {
        // 启动时会被调用一次
        // 可以在这里创建连接
    }

    @Override
    public void close() throws Exception {
        // 销毁时会被调用一次
        // 可以在这里销毁连接
    }

    // Sink 的核心逻辑
    @Override
    public void invoke(String value, Context context) throws Exception {
        // 每条数据来都会调用一次
        // 具体的写入逻辑...
    }
}

文章来源:https://blog.csdn.net/weixin_53922163/article/details/133975274
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。