• Flink（七）【输出算子（Sink）】

    前言

            今天是我写博客的第 200 篇，恍惚间两年过去了，现在已经是大三的学长了。仍然记得两年前第一次写博客的时候，当时学的应该是 Java 语言，菜的一批，写了就删，怕被人看到丢脸。当时就想着自己一年之后，两年之后能学到什么水平，什么是 JDBC、什么是 MVC、SSM，在当时都是特别好奇的东西，不过都在后来的学习中慢慢接触到，并且好多已经烂熟于心了。

            那，今天我在畅想一下，一年后的今天，我又学到了什么水平？能否达到三花聚顶、草木山石皆可为码的超凡入圣的境界？拿没拿到心仪的 offer？和那个心动过的女孩相处怎么样了？哈哈哈哈哈

    输出算子（Sink）

    学完了 Flink 在不同执行环境（本地测试环境和集群环境）下的多种读取（多种数据源）和转换操作（多种转换算子），最后就是输出操作了。

    1、连接到外部系统

    Flink 1.12 之前，Sink 算子是通过调用 DataStream 的 addSink 方法来实现的：

    stream.addSink(new SinkFunction(...));

    从 Flink 1.12 开始，Flink 重构了 Sink 架构：

    stream.sinkTo(...)

    查看 Flink 支持的连接器

    需要我们自己导入依赖，比如上面的 Kfaka 和 DataGen 我们之前使用的时候都导入过相关依赖，需要知道，有的是只支持source，有的只支持sink，有的全都支持。

    1.  <dependency>
    2.             <groupId>org.apache.flinkgroupId>
    3.             <artifactId>flink-connector-kafkaartifactId>
    4.             <version>${flink.version}version>
    5.         dependency>
    6.         <dependency>
    7.             <groupId>org.apache.flinkgroupId>
    8.             <artifactId>flink-connector-datagenartifactId>
    9.             <version>${flink.version}version>
    10.         dependency>

    2、输出到文件

            Flink 专门提供了一个流式文件系统的连接器：FileSink，为批处理和流处理提供了一个统一的 Sink，它可以将分区文件写入 Flink支持的文件系统。
            它的主要操作是将数据写入桶（buckets），每个桶中的数据都可以分割成一个个大小有限的分区文件，这样一来就实现真正意义上的分布式文件存储。我们可以通过各种配置来控制“分桶”的操作；默认的分桶方式是基于时间的，我们每小时写入一个新的桶。换句话说，每个桶内保存的文件，记录的都是 1 小时的输出数据。
            FileSink 支持行编码（Row-encoded）和批量编码（Bulk-encoded，比如 Parquet）格式。这两种不同的方式都有各自的构建器（builder），调用方法也非常简单，可以直接调用 FileSink 的静态方法：

    • 行编码：FileSink.forRowFormat（basePath，rowEncoder）。
    • 批量编码：FileSink.forBulkFormat（basePath，bulkWriterFactory）。

    在创建行或批量编码 Sink 时，我们需要传入两个参数，用来指定存储桶的基本路径（basePath）和数据的编码逻辑（rowEncoder 或 bulkWriterFactory）。

    1. package com.lyh.sink;
    2.  
    3. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    4. import org.apache.flink.api.common.serialization.SimpleStringEncoder;
    5. import org.apache.flink.api.common.typeinfo.TypeInformation;
    6. import org.apache.flink.api.common.typeinfo.Types;
    7. import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
    8. import org.apache.flink.configuration.MemorySize;
    9. import org.apache.flink.connector.datagen.source.DataGeneratorSource;
    10. import org.apache.flink.connector.datagen.source.GeneratorFunction;
    11. import org.apache.flink.connector.file.sink.FileSink;
    12. import org.apache.flink.core.fs.Path;
    13. import org.apache.flink.streaming.api.CheckpointingMode;
    14. import org.apache.flink.streaming.api.datastream.DataStreamSource;
    15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    16. import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
    17. import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
    18. import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
    19.  
    20. import java.time.Duration;
    21. import java.time.ZoneId;
    22.  
    23. /**
    24.  * @author 刘xx
    25.  * @version 1.0
    26.  * @date 2023-11-18 9:51
    27.  */
    28. public class SinkFile {
    29.     public static void main(String[] args) throws Exception {
    30.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    31.         env.setParallelism(2);
    32.         // 必须开启 检查点 不然一直都是 .inprogress
    33.         env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
    34.  
    35.         DataGeneratorSource dataGeneratorSource = new DataGeneratorSource(
    36.                 new GeneratorFunction() {
    37.             @Override
    38.             public String map(Long value) throws Exception {
    39.                 return "Number:"+value;
    40.             }
    41.         },
    42.             Long.MAX_VALUE,
    43.             RateLimiterStrategy.perSecond(10), // 每s 10条
    44.             Types.STRING
    45.         );
    46.  
    47.         DataStreamSource dataGen = env.fromSource(dataGeneratorSource, WatermarkStrategy.noWatermarks(), "data-generate");
    48.  
    49.         // todo 输出到文件系统
    50.         FileSink fileSink = FileSink.
    51.                 // 泛型方法 需要和输出结果的泛型保持一致
    52.                 forRowFormat(
    53.                 new Path("D:/Desktop"),    // 指定输出路径 可以是 hdfs:// 路径
    54.                 new SimpleStringEncoder<>("UTF-8")) // 指定编码
    55.                 .withOutputFileConfig(OutputFileConfig.builder()
    56.                         .withPartPrefix("lyh")
    57.                         .withPartSuffix(".log")
    58.                         .build())
    59.                 // 按照目录分桶 一个小时一个目录(这里的时间格式别改为分钟 会报错: flink Relative path in absolute URI:)
    60.                 .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd HH", ZoneId.systemDefault()))
    61.                 // 设置文件滚动策略-时间或者大小 10s 或 1KB 或 5min内没有新数据写入 滚动一次
    62.                 // 滚动的时候 文件就会更名为我们设定的格式(前缀)不再写入
    63.                 .withRollingPolicy(
    64.                         DefaultRollingPolicy.builder()
    65.                                 .withRolloverInterval(Duration.ofSeconds(10L))  // 10s
    66.                                 .withMaxPartSize(new MemorySize(1024)) // 1KB
    67.                                 .withInactivityInterval(Duration.ofMinutes(5))  // 5min
    68.                                 .build()
    69.                 )
    70.                 .build();
    71.  
    72.         dataGen.sinkTo(fileSink);
    73.  
    74.         env.execute();
    75.     }
    76. }

    这里我们创建了一个简单的文件 Sink，通过.withRollingPolicy()方法指定了一个“滚动策略”。“滚动”的概念在日志文件的写入中经常遇到：因为文件会有内容持续不断地写入，所以我们应该给一个标准，到什么时候就开启新的文件，将之前的内容归档保存。也就是说，上面的代码设置了在以下 3 种情况下，我们就会滚动分区文件：
    ⚫ 至少包含 10 秒的数据
    ⚫ 最近 5 分钟没有收到新的数据
    ⚫ 文件大小已达到 1 KB

    通过 withOutputFileConfig()方法指定了输出的文件名前缀和后缀。

    需要特别注意的就是一定要开启检查点，否则我们的数据一直都是正在写入的状态（具体原因后面学习到检查点的时候会详细说）。

    运行结果：

    3、输出到 Kafka

    1. 需要添加 Kafka 依赖（之前导入过了）
    2. 启动 Kafka
    3. 编写示例代码
    1. package com.lyh.sink;
    2.  
    3. import org.apache.flink.api.common.serialization.SimpleStringSchema;
    4. import org.apache.flink.connector.base.DeliveryGuarantee;
    5. import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
    6. import org.apache.flink.connector.kafka.sink.KafkaSink;
    7. import org.apache.flink.streaming.api.CheckpointingMode;
    8. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    9. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    10. import org.apache.kafka.clients.producer.ProducerConfig;
    11.  
    12. /**
    13.  * @author 刘xx
    14.  * @version 1.0
    15.  * @date 2023-11-18 11:20
    16.  */
    17. public class SinkKafka {
    18.     public static void main(String[] args) throws Exception {
    19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    20.         env.setParallelism(1);
    21.         // 如果是 精准一次 必须开启 checkpoint
    22.         env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
    23.  
    24.         SingleOutputStreamOperator sensorDS = env.socketTextStream("localhost", 9999);
    25.  
    26.         KafkaSink kafkaSink = KafkaSink.builder()
    27.                 // 指定 kafka 的地址和端口
    28.                 .setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092")
    29.                 // 指定序列化器 我们是发送方 所以我们是生产者
    30.                 .setRecordSerializer(
    31.                         KafkaRecordSerializationSchema.builder()
    32.                                 .setTopic("like")
    33.                                 .setValueSerializationSchema(new SimpleStringSchema())
    34.                                 .build()
    35.                 )
    36.                 // 写到 kafka 的一致性级别: 精准一次 / 至少一次
    37.                 // 如果是精准一次
    38.                 //  1.必须开启检查点 env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE)
    39.                 //  2.必须设置事务的前缀
    40.                 //  3.必须设置事务的超时时间: 大于 checkpoint间隔 小于 max 15分钟
    41.                 .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    42.                 .setTransactionalIdPrefix("lyh-")
    43.                 .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,10*60*1000+"")
    44.                 .build();
    45.  
    46.         sensorDS.sinkTo(kafkaSink);
    47.  
    48.         env.execute();
    49.     }
    50. }

    启动 kafka 并开启一个消费者：

    kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic like

    运行结果：

    需要特别注意的三点：

    如果是精准一次
 1.必须开启检查点 env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE)
 2.必须设置事务的前缀
 3.必须设置事务的超时时间: 大于 checkpoint间隔 小于 max 15分钟

    自定义序列化器

    我们上面用的自带的序列化器，但是如果我们有 key 的话，就需要自定义序列化器了，替换上面的代码：

    1. .setRecordSerializer(
    2.         /**
    3.          * 如果要指定写入 kafka 的key 就需要自定义序列化器
    4.          * 实现一个接口 重写序列化方法
    5.          * 指定key 转为 bytes[]
    6.          * 指定value 转为 bytes[]
    7.          * 返回一个 ProducerRecord(topic名,key,value)对象
    8.          */
    9.         new KafkaRecordSerializationSchema() {
    10.             @Nullable
    11.             @Override
    12.             // ProducerRecord 返回一个生产者消息,key,value 分别对应两个字节数组
    13.             public ProducerRecord<byte[], byte[]> serialize(String element, KafkaSinkContext context, Long timestamp) {
    14.                 String[] datas = element.split(",");
    15.                 byte[] key = datas[0].getBytes(StandardCharsets.UTF_8);
    16.                 byte[] value = element.getBytes(StandardCharsets.UTF_8);
    17.                 return new ProducerRecord<>("like",key,value);
    18.             }
    19.         }
    20. )

    运行结果： 

    4、输出到 MySQL

    添加依赖（1.17版本的依赖需要指定仓库才能找到，因为阿里云和默认的maven仓库是没有的）：

    1. <dependency>
    2.             <groupId>org.apache.flinkgroupId>
    3.             <artifactId>flink-connector-jdbcartifactId>
    4.             <version>1.17-SNAPSHOTversion>
    5.         dependency>
    6. <dependency>
    7.             <groupId>mysqlgroupId>
    8.             <artifactId>mysql-connector-javaartifactId>
    9.             <version>8.0.31version>
    10.         dependency>
    11.  
    12. ....
    13.  
    14.     <repositories>
    15.         <repository>
    16.             <id>apache-snapshotsid>
    17.             <name>apache snapshotsname>
    18.             <url>https://repository.apache.org/content/repositories/snapshots/url>
    19.         repository>
    20.     repositories>

    创建表格 

    编写代码，将输入的数据行分隔为对象参数，每行数据生成一个对象进行处理。 

    1. package com.lyh.sink;
    2.  
    3. import com.lyh.bean.WaterSensor;
    4. import function.WaterSensorFunction;
    5. import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
    6. import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
    7. import org.apache.flink.connector.jdbc.JdbcSink;
    8. import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
    9. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    11. import org.apache.flink.streaming.api.functions.sink.SinkFunction;
    12.  
    13. import java.sql.PreparedStatement;
    14. import java.sql.SQLException;
    15.  
    16. /**
    17.  * @author 刘xx
    18.  * @version 1.0
    19.  * @date 2023-11-18 12:32
    20.  */
    21. public class SinkMySQL {
    22.     public static void main(String[] args) throws Exception {
    23.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    24.         env.setParallelism(1);
    25.  
    26.         SingleOutputStreamOperator sensorDS = env.socketTextStream("localhost", 9999)
    27.                 .map(new WaterSensorFunction()); //输入进来的数据自动转为 WaterSensor类型
    28.  
    29.  
    30.         /**
    31.          * todo 写入 mysql
    32.          * 1.这里需要用旧的sink写法：addSink
    33.          * 2.JDBC的4个参数
    34.          *   (1) 执行的sql语句
    35.          *   (2) 对占位符进行填充
    36.          *   (3) 执行选项  -> 攒批,重试
    37.          *   (4) 连接选项 -> driver,username,password,url
    38.          */
    39.         SinkFunction jdbcSink = JdbcSink.sink("insert into flink.ws values(?,?,?)",
    40.                 // 指定 sql 中占位符的值
    41.                 new JdbcStatementBuilder() {
    42.                     @Override
    43.                     public void accept(PreparedStatement stmt, WaterSensor sensor) throws SQLException {
    44.                         // 占位符从 1 开始
    45.                         stmt.setString(1, sensor.getId());
    46.                         stmt.setLong(2, sensor.getTs());
    47.                         stmt.setInt(3, sensor.getVc());
    48.                     }
    49.                 }, JdbcExecutionOptions.builder()
    50.                         .withMaxRetries(3)  //最多重试3次(不包括第一次,共4次)
    51.                         .withBatchSize(100) //每收集100条记录进行一次写入
    52.                         .withBatchIntervalMs(3000)  // 批次3s(即使没有达到100条记录,只要过了3s JDBCSink也会进行记录的写入),这有助于确保数据及时写入，而不是无限期地等待批处理大小达到。
    53.                         .build()
    54.                 , new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
    55.                         .withUrl("jdbc:mysql://localhost:3306/flink?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8")
    56.                         .withDriverName("com.mysql.cj.jdbc.Driver")
    57.                         .withUsername("root")
    58.                         .withPassword("Yan1029.")
    59.                         // mysql 默认8小时不使用连接就主动断开连接
    60.                         .withConnectionCheckTimeoutSeconds(60) // 重试连接直接的间隔,上面我们设置最多重试3次,每次间隔60s
    61.                         .build()
    62.         );
    63.  
    64.         sensorDS.addSink(jdbcSink);
    65.  
    66.         env.execute();
    67.     }
    68. }

     查询结果：

    5、自定义 Sink 输出

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

    这里我们自定义实现一个向 HBase 中插入数据的 Sink。

    注意：这里只是做一个简单的 Demo，下面的代码不难发现，我们只是对 nosq:student 表下的 info:name 进行了两次的覆盖。如果要实现复杂的处理功能，需要对数据类型进行定义，因为 HBase 的数据是按列存储的，所以对于复杂的 Hbase 表，我们难以通过 Java bean 来插入数据。而且，一般经常用的连接器，Flink 大部分已经提供了，开发中我们一般也很少自定义 Sink 输出。

    1. package com.lyh.sink;
    2.  
    3. import com.lyh.utils.HBaseConnection;
    4. import org.apache.flink.configuration.Configuration;
    5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    6. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
    7. import org.apache.hadoop.hbase.TableName;
    8. import org.apache.hadoop.hbase.client.Connection;
    9. import org.apache.hadoop.hbase.client.Put;
    10. import org.apache.hadoop.hbase.client.Table;
    11.  
    12. import java.nio.charset.StandardCharsets;
    13.  
    14. /**
    15.  * @author 刘xx
    16.  * @version 1.0
    17.  * @date 2023-11-18 15:59
    18.  */
    19. public class SinkCustomHBase {
    20.     public static void main(String[] args) throws Exception {
    21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    22.         env.setParallelism(1);
    23.         env.fromElements("tom","bob").addSink(new RichSinkFunction() {
    24.             public Connection con;
    25.             @Override
    26.             public void open(Configuration parameters) throws Exception {
    27.                 super.open(parameters);
    28.                 con = HBaseConnection.getConnection("hadoop102:2181");
    29.             }
    30.  
    31.             @Override
    32.             public void invoke(String value, Context context) throws Exception {
    33.                 super.invoke(value, context);
    34.                 Table table = con.getTable(TableName.valueOf("nosql","student"));
    35.                 Put put = new Put("1001".getBytes(StandardCharsets.UTF_8));
    36.                 put.addColumn("info".getBytes(StandardCharsets.UTF_8)
    37.                 ,"name".getBytes(StandardCharsets.UTF_8),
    38.                         value.getBytes(StandardCharsets.UTF_8));
    39.                 table.put(put);
    40.                 table.close();
    41.             }
    42.  
    43.             @Override
    44.             public void close() throws Exception {
    45.                 super.close();
    46.                 HBaseConnection.close();
    47.             }
    48.  
    49.         });
    50.  
    51.         env.execute();
    52.     }
    53. }

    这里用到一个简单的连接 HBase 的工具类：
     

    1. package com.lyh.utils;
    2.  
    3. import org.apache.hadoop.conf.Configuration;
    4. import org.apache.hadoop.hbase.client.Connection;
    5. import org.apache.hadoop.hbase.client.ConnectionFactory;
    6.  
    7. import java.io.IOException;
    8.  
    9. /**
    10.  * @author 刘xx
    11.  * @version 1.0
    12.  * @date 2023-11-18 16:04
    13.  */
    14. public class HBaseConnection {
    15.  
    16.     private static Connection connection;
    17.  
    18.     public static Connection getConnection(String hosts) throws IOException {
    19.         Configuration conf = new Configuration();
    20.         conf.set("hbase.zookeeper.quorum", hosts);
    21.         conf.setInt("hbase.rpc.timeout", 10000); // 设置最大超时 10 s
    22.         connection = ConnectionFactory.createConnection(conf);
    23.         return connection;
    24.     }
    25.  
    26.     public static void close() throws IOException {
    27.         if (connection!=null)
    28.             connection.close();
    29.     }
    30. }

     

  • 相关阅读:
    【附源码】Python计算机毕业设计民宿网站管理系统
    CentOS7 安装 NVIDIA Container Toolkit
    Vue.js快速入门之七：系统权限管理
    IDEA Service窗口
    表情包APP小程序制作开发功能有哪些？
    OTA设计思路
    100天精通Python（可视化篇）——第107天：Pyecharts绘制多种炫酷旭日图参数说明+代码实战
    计算机网络协议------从入门到深化
    MySql 数据库【约束】
    服务器与Tomcat的区别
  • 原文地址：https://blog.csdn.net/m0_64261982/article/details/134474275