• 2.1、如何在FlinkSQL中读取&写出到Kafka

    目录

    1、环境设置

    方式1：在Maven工程中添加pom依赖

    方式2：在 sql-client.sh 中添加 jar包依赖

    2、读取Kafka

    2.1 创建 kafka表

    2.2 读取 kafka消息体（Value）

    使用 'format' = 'json' 解析json格式的消息

    使用 'format' = 'csv' 解析csv格式的消息

    使用 'format' = 'raw' 解析kafka消息为单个字符串字段

    2.3 读取 kafka消息键（Key）

    2.4 读取 kafka元数据(Metadata)

    2.5 如何指定起始消费位点

    从指定的timestamp开始消费:

    从指定的timestamp开始消费:

    2.6 创建 kafka表时，添加水位线生成策略

    3、写入Kafka

    3.1 写入 kafka时，可以指定的元数据

    1、环境设置

            Kafka 连接器提供从 Kafka topic 中消费和写入数据的能力。

            官网链接：官网

    方式1：在Maven工程中添加pom依赖

    1. <dependency>
    2.   <groupId>org.apache.flinkgroupId>
    3.   <artifactId>flink-connector-kafkaartifactId>
    4.   <version>1.17.1version>
    5. dependency>

    方式2：在 sql-client.sh 中添加 jar包依赖

            将 flink-sql-connector-kafka-1.17.1.jar 上传到flink的lib目录下 (可以去官网下载jar包)

            或者 启动 sql-client.sh 时，指定jar依赖

    bin/sql-client.sh -j lib/flink-sql-connector-kafka-1.17.1.jar

    2、读取Kafka

    2.1 创建 kafka表

    1. CREATE TABLE SourceKafkaTable (
    2.    指定物理字段,
    3.    指定元数据字段,
    4.    指定水位线生成策略
    5. ) WITH (
    6.   'connector' = 'kafka',                             --【必选】指定 连接器类型，kafka用'kafka'
    7.   'properties.bootstrap.servers' = 'localhost:9092', --【必选】指定 Kafka broker列表,用逗号分隔
    8.   'topic' = 'user_behavior',                         --【必选】指定 topic列表,用逗号分隔
    9.   'topic-pattern' = '.*log_kafka.*',                 --【必选】指定 匹配读取 topic 名称的正则表达式, 和 topic 配置一个即可
    10.   'properties.group.id' = 'testGroup',               --【可选】指定 消费者组id，不指定时会自定生成 KafkaSource-{tableIdentifier}
    11.   'scan.startup.mode' = 'earliest-offset',           --【可选】指定起始消费位点，默认值 earliest-offset
    12.   'format' = 'csv'                                   --【必选】指定 消息的格式类型, 和 value.format 是等价的(配置一个即可)
    13. );

    2.2 读取 kafka消息体（Value）

    在FlinkSQL读取kafka时，可以根据kafka存储的消息的格式，通过 'value.format' = 'csv|raw|json...'

    来指定使用哪种格式来对kafka消息进行解析，并将解析的结果映射到表中的字段中去。

    使用 'format' = 'json' 解析json格式的消息

    当 kafka消息为json格式，可以使用  'format' = 'json' 在创建表时对json串进行解析，并将解析后的结果映射到表中的字段中去

    注意：这种方式只能解析单层级的json格式，多层级时无法解析

               如果为多层级json格式时，可以使用raw格式 + udf函数来对json进行解析

    导入Maven的pom依赖

    1. <dependency>
    2.   <groupId>org.apache.flinkgroupId>
    3.   <artifactId>flink-jsonartifactId>
    4.   <version>1.17.1version>
    5. dependency>

    创建FlinkTable

    1. -- TODO 创建用于读取kafka消息的flink表(消息格式为json)
    2. -- kafka消息示例：{"ID":0,"NAME":"大王0"}
    3. CREATE TABLE kafka_table_source_json (
    4.   `ID` STRING,
    5.   `NAME` STRING
    6. ) WITH (
    7.   'connector' = 'kafka',
    8.   'topic' = '20231009',
    9.   'properties.bootstrap.servers' = 'worker01:9092',
    10.   'properties.group.id' = 'FlinkConsumer',
    11.   'scan.startup.mode' = 'earliest-offset',
    12.   'format' = 'json',
    13.   'json.fail-on-missing-field' = 'false',
    14.   'json.ignore-parse-errors' = 'true'
    15. );
    16.  
    17. -- TODO 解析json串时，容错性设置
    18. 'json.fail-on-missing-field' = 'false' -- 当解析字段缺失时，是跳过当前字段或行，还是抛出错误失败（默认为 false，即抛出错误失败）
    19. 'json.ignore-parse-errors' = 'true'  -- 当解析异常时，是跳过当前字段或行，还是抛出错误失败（默认为 false，即抛出错误失败）。如果忽略字段的解析异常，则会将该字段值设置为null。
    20.  
    21. -- 触发读取kafka操作
    22. select * from kafka_table_source_json;

    运行结果：

    使用 'format' = 'csv' 解析csv格式的消息

    当 kafka消息为csv格式，可以使用  'format' = 'csv' 在创建表时对csv进行解析，并将解析后的结果映射到表中的字段中去

    导入Maven的pom依赖

    1. <dependency>
    2.   <groupId>org.apache.flinkgroupId>
    3.   <artifactId>flink-csvartifactId>
    4.   <version>1.17.1version>
    5. dependency>

    创建FlinkTable

    1. -- TODO 创建用于读取kafka消息的flink表(消息格式为csv)
    2. -- kafka消息示例：2,3.1
    3. CREATE TABLE kafka_table_source_csv (
    4.   `order_id` BIGINT,
    5.   `price` DOUBLE
    6. ) WITH (
    7.   'connector' = 'kafka',
    8.   'topic' = 'csv_format',
    9.   'properties.bootstrap.servers' = 'worker01:9092',
    10.   'properties.group.id' = 'FlinkConsumer',
    11.   'scan.startup.mode' = 'earliest-offset',
    12.   'value.format' = 'csv'
    13. );
    14.  
    15. -- 触发读取kafka操作
    16. select * from kafka_table_source_csv;

    运行结果：

    使用 'format' = 'raw' 解析kafka消息为单个字符串字段

    可以使用  'format' = 'raw' 将kafka消息以原始格式映射到flink表中的string类型的字段中

    创建FlinkTable

    1. -- TODO 创建用于读取kafka消息的flink表(消息格式为json)
    2. -- kafka消息示例：{"ID":0,"NAME":"大王0"}
    3. CREATE TABLE kafka_table_source_raw (
    4.   `log` STRING
    5. ) WITH (
    6.   'connector' = 'kafka',
    7.   'topic' = '20231009',
    8.   'properties.bootstrap.servers' = 'worker01:9092',
    9.   'properties.group.id' = 'FlinkConsumer',
    10.   'scan.startup.mode' = 'earliest-offset',
    11.   'format' = 'raw'
    12. );
    13.  
    14. -- 触发读取kafka操作
    15. select * from kafka_table_source_raw;

    运行结果：

    2.3 读取 kafka消息键（Key）

    kafka消息信息：

    1. {
    2.     "key":{
    3.         "ID_1":0,
    4.         "NAME_1":"大王0"
    5.     },
    6.     "value":{
    7.         "ID":0,
    8.         "NAME":"大王0"
    9.     },
    10.     "metadata":{
    11.         "offset":0,
    12.         "topic":"readKey",
    13.         "partition":0
    14.     }
    15. }

    创建FlinkTable

    1. -- 读取kafka消息中的key部分
    2. CREATE TABLE kafka_table_source_read_key (
    3.   `ID` STRING,
    4.   `NAME` STRING,
    5.   `ID_1` STRING,
    6.   `NAME_1` STRING
    7. ) WITH (
    8.   'connector' = 'kafka',
    9.   'topic' = 'readKey',
    10.   'properties.bootstrap.servers' = 'worker01:9092',
    11.   'properties.group.id' = 'FlinkConsumer',
    12.   'scan.startup.mode' = 'earliest-offset',
    13.   'key.format' = 'json',
    14.   'key.json.ignore-parse-errors' = 'true',
    15.   'key.fields' = 'ID_1;NAME_1',
    16.   'value.format' = 'json'
    17. );

    2.4 读取 kafka元数据(Metadata)

    创建FlinkTable

    1. -- TODO 创建读取kafka表时，同时读取kafka元数据字段
    2. CREATE TABLE kafka_table_source_read_metadata (
    3.   `log` STRING,
    4.   `topic` STRING METADATA VIRTUAL, -- 消息所属的 topic
    5.   `partition` BIGINT METADATA VIRTUAL, -- 消息所属的 partition ID
    6.   `offset` BIGINT METADATA VIRTUAL, -- 消息在partition中的 offset
    7.   `timestamp` TIMESTAMP(3) METADATA FROM 'timestamp' -- 消息的时间戳
    8. ) WITH (
    9.   'connector' = 'kafka',
    10.   'topic' = 'readKey',
    11.   'properties.bootstrap.servers' = 'worker01:9092',
    12.   'properties.group.id' = 'FlinkConsumer',
    13.   'scan.startup.mode' = 'earliest-offset',
    14.   'format' = 'raw'
    15. );
    16.  
    17. select * from kafka_table_source_read_metadata;

    2.5 如何指定起始消费位点

    scan.startup.mode 配置项决定了 Kafka consumer 的启动模式。有效值为：

    • group-offsets：从 Zookeeper/Kafka 中某个指定的消费组已提交的偏移量开始。
    • earliest-offset：从可能的最早偏移量开始。
    • latest-offset：从最末尾偏移量开始。
    • timestamp：从用户为每个 partition 指定的时间戳开始。
      • 如果使用了 timestamp，必须使用另外一个配置项              scan.startup.timestamp-millis=时间戳(毫秒值)
    • specific-offsets：从用户为每个 partition 指定的偏移量开始。
      • 如果使用了 specific-offsets，必须使用另外一个配置项 scan.startup.specific-offsets 来为每个 partition 指定起始偏移量， 例如，选项值 partition:0,offset:42;partition:1,offset:300 表示 partition 0 从偏移量 42 开始，partition 1 从偏移量 300 开始

    默认值 group-offsets 表示从 Zookeeper/Kafka 中最近一次已提交的偏移量开始消费。

    从指定的timestamp开始消费:

    1. // --------------------------------------------------------------------------------------------
    2. //  TODO 从指定的timestamp开始消费
    3. // --------------------------------------------------------------------------------------------
    4. drop table kafka_table_source_test_startup_timestamp;
    5. CREATE TABLE kafka_table_source_test_startup_timestamp (
    6.   `log` STRING,
    7.   `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
    8.   `offset` BIGINT METADATA VIRTUAL
    9. ) WITH (
    10.   'connector' = 'kafka',
    11.   'topic' = '20231009',
    12.   'properties.bootstrap.servers' = 'worker01:9092',
    13.   'properties.group.id' = 'FlinkConsumer',
    14.   'scan.startup.mode' = 'timestamp', -- 从用户为每个 partition 指定的时间戳开始
    15.   'scan.startup.timestamp-millis' = '1697008386973', -- 从 指定的timestamp开始(包括)消费
    16.    'value.format' = 'raw'
    17. );
    18.  
    19. select * 
    20. ,cast(UNIX_TIMESTAMP(cast(ts as string), 'yyyy-MM-dd HH:mm:ss.SSS') as string) || SPLIT_INDEX(cast(ts as string),'.',1) as timestamp_hmz
    21. from kafka_table_source_test_startup_timestamp;

    运行结果：

    从指定的timestamp开始消费:

    1. // --------------------------------------------------------------------------------------------
    2. //  TODO 从指定的offset开始消费
    3. // --------------------------------------------------------------------------------------------
    4. drop table kafka_table_source_test_startup_offsets;
    5. CREATE TABLE kafka_table_source_test_startup_offsets (
    6.   `log` STRING,
    7.   `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
    8.   `offset` BIGINT METADATA VIRTUAL
    9. ) WITH (
    10.   'connector' = 'kafka',
    11.   'topic' = '20231009',
    12.   'properties.bootstrap.servers' = 'worker01:9092',
    13.   'properties.group.id' = 'FlinkConsumer',
    14.   'scan.startup.mode' = 'specific-offsets', -- 从用户为每个 partition 指定的偏移量开始
    15.   'scan.startup.specific-offsets' = 'partition:0,offset:4', -- 为每个 partition 指定起始偏移量
    16.    'value.format' = 'raw'
    17. );
    18.  
    19. select * from kafka_table_source_test_startup_offsets;

    运行结果：

    2.6 创建 kafka表时，添加水位线生成策略

    1. // --------------------------------------------------------------------------------------------
    2. //  TODO 创建 kafka表时，添加水位线生成策略
    3. // --------------------------------------------------------------------------------------------
    4. drop table kafka_table_source_test_watermark;
    5. CREATE TABLE kafka_table_source_test_watermark (
    6.   `log` STRING,
    7.   `event_time` TIMESTAMP(3) METADATA FROM 'timestamp',
    8.   `offset` BIGINT METADATA VIRTUAL,
    9.   WATERMARK FOR event_time AS event_time -- 根据kafka的timestamp，生成水位线，使用 严格递增时间戳水位线生成策略
    10. ) WITH (
    11.   'connector' = 'kafka',
    12.   'topic' = '20231009',
    13.   'properties.bootstrap.servers' = 'worker01:9092',
    14.   'properties.group.id' = 'FlinkConsumer',
    15.   'scan.startup.mode' = 'specific-offsets', -- 从用户为每个 partition 指定的偏移量开始
    16.   'scan.startup.specific-offsets' = 'partition:0,offset:4', -- 为每个 partition 指定起始偏移量
    17.    'value.format' = 'raw'
    18. );
    19.  
    20. select * from kafka_table_source_test_watermark;

    3、写入Kafka

    3.1 写入 kafka时，可以指定的元数据

    1. // --------------------------------------------------------------------------------------------
    2. //  TODO 通过flinksql向kafka写入数据(写入时指定 timestamp)
    3. // --------------------------------------------------------------------------------------------
    4. drop table kafka_table_source_test_startup_mode;
    5. CREATE TABLE kafka_table_source_test_startup_mode (
    6.   `order_id` BIGINT,
    7.   `price` DOUBLE,
    8.   `ts` TIMESTAMP(3) METADATA FROM 'timestamp',
    9.   `offset` BIGINT METADATA VIRTUAL
    10. ) WITH (
    11.   'connector' = 'kafka',
    12.   'topic' = '20231011',
    13.   'properties.bootstrap.servers' = 'worker01:9092',
    14.   'properties.group.id' = 'FlinkConsumer',
    15.   'scan.startup.mode' = 'earliest-offset',
    16.   'value.format' = 'csv'
    17. );
    18.  
    19. insert into kafka_table_source_test_startup_mode(order_id, price,ts)
    20. SELECT * FROM (VALUES
    21.   (1, 2.0,TO_TIMESTAMP_LTZ(1000, 3))
    22. , (2, 4.0,TO_TIMESTAMP_LTZ(2000, 3))
    23. , (3, 6.0,TO_TIMESTAMP_LTZ(3000, 3))
    24. , (4, 7.0,TO_TIMESTAMP_LTZ(4000, 3))
    25. , (5, 8.0,TO_TIMESTAMP_LTZ(5000, 3))
    26. , (6, 10.0,TO_TIMESTAMP_LTZ(6000, 3))
    27. , (7, 12.0,TO_TIMESTAMP_LTZ(7000, 3))
    28. ) AS book (order_id, price,ts);
    29.  
    30. -- 触发读取kafka操作
    31. select * from kafka_table_source_test_startup_mode;

  • 相关阅读:
    ssm+vue租房维保系统-房屋维修系统
    Hive Lateral View explode列为空时导致数据异常丢失
    来实现一个DataStore的封装吧
    流畅的python
    计及新能源出力不确定性的电气设备综合能源系统协同优化（Matlab代码实现）
    【微服务部署】二、Jenkins一键打包部署SpringBoot应用步骤详解
    React-hooks有哪些 包括用法是什么？
    Since Maven 3.8.1 http repositories are blocked
    无监控，不运维！Prometheus 在线服务的监控实操指南
    Mybatis面经
  • 原文地址：https://blog.csdn.net/weixin_42845827/article/details/133708706