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依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>1.17.1</version>
</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表

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

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依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>1.17.1</version>
</dependency>

创建FlinkTable

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

运行结果：

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

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

导入Maven的pom依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-csv</artifactId><version>1.17.1</version>
</dependency>

创建FlinkTable

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

运行结果：

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

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

创建FlinkTable

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

运行结果：

2.3 读取 kafka消息键（Key）

kafka消息信息：

{"key":{"ID_1":0,"NAME_1":"大王0"},"value":{"ID":0,"NAME":"大王0"},"metadata":{"offset":0,"topic":"readKey","partition":0}
}

创建FlinkTable

-- 读取kafka消息中的key部分
CREATE TABLE kafka_table_source_read_key (`ID` STRING,`NAME` STRING,`ID_1` STRING,`NAME_1` STRING
) WITH ('connector' = 'kafka','topic' = 'readKey','properties.bootstrap.servers' = 'worker01:9092','properties.group.id' = 'FlinkConsumer','scan.startup.mode' = 'earliest-offset','key.format' = 'json','key.json.ignore-parse-errors' = 'true','key.fields' = 'ID_1;NAME_1','value.format' = 'json'
);

2.4 读取 kafka元数据(Metadata)

创建FlinkTable

-- TODO 创建读取kafka表时，同时读取kafka元数据字段
CREATE TABLE kafka_table_source_read_metadata (`log` STRING,`topic` STRING METADATA VIRTUAL, -- 消息所属的 topic`partition` BIGINT METADATA VIRTUAL, -- 消息所属的 partition ID`offset` BIGINT METADATA VIRTUAL, -- 消息在partition中的 offset`timestamp` TIMESTAMP(3) METADATA FROM 'timestamp' -- 消息的时间戳
) WITH ('connector' = 'kafka','topic' = 'readKey','properties.bootstrap.servers' = 'worker01:9092','properties.group.id' = 'FlinkConsumer','scan.startup.mode' = 'earliest-offset','format' = 'raw'
);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开始消费:

// --------------------------------------------------------------------------------------------
//  TODO 从指定的timestamp开始消费
// --------------------------------------------------------------------------------------------
drop table kafka_table_source_test_startup_timestamp;
CREATE TABLE kafka_table_source_test_startup_timestamp (`log` STRING,`ts` TIMESTAMP(3) METADATA FROM 'timestamp',`offset` BIGINT METADATA VIRTUAL
) WITH ('connector' = 'kafka','topic' = '20231009','properties.bootstrap.servers' = 'worker01:9092','properties.group.id' = 'FlinkConsumer','scan.startup.mode' = 'timestamp', -- 从用户为每个 partition 指定的时间戳开始'scan.startup.timestamp-millis' = '1697008386973', -- 从 指定的timestamp开始(包括)消费'value.format' = 'raw'
);select * 
,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
from kafka_table_source_test_startup_timestamp;

运行结果：

从指定的timestamp开始消费:

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

运行结果：

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

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

3、写入Kafka

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

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

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/103497.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

快速学习微服务保护框架--Sentinel

快速学习微服务保护框架--Sentinel

学习一个框架最好的方式就是查看官方地址,sentinel是国内阿里巴巴公司的,官网更方便官网 官网 微服务保护框架 Sentinel 1.初识Sentinel 1.1.雪崩问题及解决方案 1.1.1.雪崩问题 微服务中&#xff0c;服务间调用关系错综复杂&#xff0c;一个微服务往往依赖于多个其它微…
阅读更多...
全力以赴，火山引擎边缘云代表团出战亚运会

全力以赴，火山引擎边缘云代表团出战亚运会

END 未来&#xff0c;火山引擎边缘云赛事阵容将继续全力以赴&#xff0c;通过领先、可信赖的云和智能技术&#xff0c;助力游戏行业呈现更加精彩的竞技赛事。
阅读更多...
光耦合器继电器与传统继电器：哪种最适合您的项目？

光耦合器继电器与传统继电器：哪种最适合您的项目？

在电子和电气工程领域&#xff0c;继电器的选择可以显着影响项目的性能和安全性。两种常见类型的继电器是光耦合器继电器和传统机电继电器。每个都有其优点和缺点&#xff0c;因此选择过程对于项目的成功结果至关重要。 光耦合器继电器&#xff1a;基础知识 光耦合器继电器&…
阅读更多...
linux环境下使用lighthouse与selenium

linux环境下使用lighthouse与selenium

一、安装谷歌浏览器、谷歌浏览器驱动、lighthouse shell脚本 apt update && apt -y upgrade apt install -y curl curl -fsSL https://deb.nodesource.com/setup_18.x | bash apt install -y nodejs apt install -y npm npm install -g lighthouse apt-get install -y …
阅读更多...
AutoRunner自动化测试工具

AutoRunner自动化测试工具

AutoRunner自动化测试工具(简称AR&#xff09;是泽众软件自主研发的自动化测试工具&#xff0c;也是一个自动测试框架&#xff0c;加载不同的测试组件&#xff0c;能够实现面向不同应用的测试。通过录制和编写测试脚本&#xff0c;实现功能测试、回归测试的自动化&#xff0c;自…
阅读更多...
记录一个@Transaction注解引发的bug

记录一个@Transaction注解引发的bug

记录一个Transactional(readOnly true)注解引发的bug 一、问题代码和报错 1-1 问题代码模拟 引发这个问题的三大要素分别是&#xff1a; 事务注解任意数据库操作数据库操作后执行耗时业务&#xff08;耗时超过数据库配置的超时时间&#xff09; //1.这里是问题的核心之一…
阅读更多...
“过度炒作”的大模型巨亏，Copilot每月收10刀,倒赔20刀

“过度炒作”的大模型巨亏，Copilot每月收10刀,倒赔20刀

大模型无论是训练还是使用&#xff0c;都比较“烧钱”&#xff0c;只是其背后的成本究竟高到何处&#xff1f;已经推出大模型商用产品的公司到底有没有赚到钱&#xff1f;事实上&#xff0c;即使微软、亚马逊、Adobe 这些大厂&#xff0c;距离盈利之路还有很远&#xff01;同时…
阅读更多...
微信自动批量添加好友的方法

微信自动批量添加好友的方法

在现在的营销中微信已成为一种重要的沟通方式。微信目前是没有自动批量添加好友的功能&#xff0c;需要运营者一个一个手动去添加&#xff0c;这样太过于浪费时间&#xff0c;并且加频繁了还容易被封号&#xff0c;今天给大家介绍几种手动批量加好友的方式以及怎么借助第三方软…
阅读更多...
国窖1573持续演绎共生魅力，携手马岩松个展感知建筑艺术之美

国窖1573持续演绎共生魅力，携手马岩松个展感知建筑艺术之美

执笔 | 洪大大 编辑 | 萧 萧 从艺术到文化、从需求到场景、从体验到消费&#xff0c;国窖1573正通过一次次尝试与探索实现与多元文化的共创与共生。 10月12日&#xff0c;国窖1573品牌挚友马岩松举办的“流动的大地”展览在深圳当代艺术与城市规划馆正式开幕。泸州老窖股份…
阅读更多...
什么是MTU（Maximum Transmission Unit）？

什么是MTU（Maximum Transmission Unit）？

最大传输单元MTU&#xff08;Maximum Transmission Unit&#xff0c;MTU&#xff09;&#xff0c;是指网络能够传输的最大数据包大小&#xff0c;以字节为单位。MTU的大小决定了发送端一次能够发送报文的最大字节数。如果MTU超过了接收端所能够承受的最大值&#xff0c;或者是超…
阅读更多...
2023年09月 C/C++（五级）真题解析#中国电子学会#全国青少年软件编程等级考试

2023年09月 C/C++（五级）真题解析#中国电子学会#全国青少年软件编程等级考试

C/C++编程（1~8级）全部真题・点这里 Python编程（1~6级）全部真题・点这里 第1题：红与黑 有一间长方形的房子，地上铺了红色、黑色两种颜色的正方形瓷砖。你站在其中一块黑色的瓷砖上，只能向相邻的黑色瓷砖移动。请写一个程序，计算你总共能够到达多少块黑色的瓷砖。 时间限…
阅读更多...
[42000][923] ORA-00923: 未找到要求的 FROM 关键字

[42000][923] ORA-00923: 未找到要求的 FROM 关键字

在oracle数据库写分页查询&#xff0c;使用 rownum时候出错&#xff0c; 代码&#xff1a; SELECT *FROM (SELECT *, ROWNUM AS rnumFROM test t ) WHERE rnum BETWEEN 1 AND 5; 报错&#xff1a; [42000][923] ORA-00923: 未找到要求的 FROM 关键字 Position: 31 问题原因…
阅读更多...
vue3前端开发系列 - electron开发桌面程序(2023-10月最新版)

vue3前端开发系列 - electron开发桌面程序(2023-10月最新版)

文章目录 1. 说明2. 创建项目3. 创建文件夹electron3.1 编写脚本electron.js3.2 编写脚本proload.js 4. 修改package.json4.1 删除type4.2 修改scripts4.3 完整的配置如下 5. 修改App.vue6. 修改vite.config.ts7. 启动8. 打包安装9. 项目公开地址 1. 说明 本次安装使用的环境版…
阅读更多...
kantts docker化

kantts docker化

kan-tts docker本地化 环境安装 下载docker镜像(python3.8的) registry.cn-hangzhou.aliyuncs.com/modelscope-repo/modelscope:ubuntu20.04-cuda11.8.0-py38-torch2.0.1-tf2.13.0-1.9.2 安装基础模型 pip install modelscope 安装语音模型 pip install "modelscope…
阅读更多...
RunnerGo测试平台，无代码玩转UI自动化测试

RunnerGo测试平台，无代码玩转UI自动化测试

首先需要进入官网&#xff0c;RunnerGo支持开源&#xff0c;可以自行下载安装&#xff0c;也可以点击右上角体验企业版按钮快速体验 点击体验企业版进入工作台后可以点击页面上方的UI自动化 进入到测试页面 创建元素 我们可以在元素管理中创建我们测试时需要的元素 这里我们以…
阅读更多...
Chrome | ERR_MANDATORY_PROXY_CONFIGURATION_FAILED

Chrome | ERR_MANDATORY_PROXY_CONFIGURATION_FAILED

重置设置 / 创建新账户
阅读更多...
Xcode升级到15.0 解决DT_TOOLCHAIN_DIR问题

Xcode升级到15.0 解决DT_TOOLCHAIN_DIR问题

根据个人开发遇到的问题做的总结&#xff0c;公司要求Xcode 14.2 &#xff0c;Swift 5.7开发&#xff0c;由于升级了Mac 14.0系统后&#xff0c;Xcode 14.2不能使用&#xff0c;解决方案目前有2个 一、在原来Xcode 14.2 的显示包内容&#xff0c;如图 二、升级到Xcode的15.0后…
阅读更多...
CentOS系统/root根目录扩容（扩展逻辑卷）

CentOS系统/root根目录扩容（扩展逻辑卷）

具体操作步骤 1、查看本机磁盘环境挂载情况 2、添加磁盘分区 3、开始扩容 4、同步到文件系统 1、查看本机磁盘环境挂载情况 [rooticon ~]# df -lh 可以看到/dev/mapper/centos-root 路径下容量为50G&#xff0c;我们要给这个路径下的容量扩容&#xff1a;[rooticon ~]# lsblk…
阅读更多...
【Jenkins】调用API构建并钉钉通知

【Jenkins】调用API构建并钉钉通知

文章目录 Jenkins API介绍提交作业带参数的作业API 令牌 Shell调用代码 Jenkins API介绍 Jenkins 提供了远程访问 API。目前它有三种格式&#xff1a; XML JSON Python 远程访问 API 形式为"…/api/" 例如&#xff0c; Jenkins 安装位于https://ci.jenkins.io&a…
阅读更多...
协同育人|电巢携手武汉软件工程职业学院项目实训顺利开班！

协同育人|电巢携手武汉软件工程职业学院项目实训顺利开班！

为深化校企合作&#xff0c;产教融合助力新工科建设&#xff0c;提升学生工程实践能力&#xff0c;电巢工程能力实训班按照不同岗位类别&#xff0c;匹配对应的企业岗位任职能力要求对学生开展分级培养&#xff0c;以产业需求为导向&#xff0c;培养创新型、应用型人才。 10月…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023