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

news2024/9/20 5:39:17

目录

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.coloradmin.cn/o/1080996.html

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

相关文章

力扣第98题 验证二叉搜索树 c++ 与上一篇文章相似

力扣第98题 验证二叉搜索树 c++ 与上一篇文章相似

题目 98. 验证二叉搜索树 中等 相关标签 树 深度优先搜索 二叉搜索树 二叉树 给你一个二叉树的根节点 root &#xff0c;判断其是否是一个有效的二叉搜索树。 有效 二叉搜索树定义如下&#xff1a; 节点的左子树只包含 小于 当前节点的数。节点的右子树只包含 大于 当…
阅读更多...
淘宝天猫商品历史价格API接口

淘宝天猫商品历史价格API接口

获取淘宝商品历史价格接口的步骤如下&#xff1a; 注册淘宝开放平台&#xff1a;首先在淘宝开放平台上注册一个账号&#xff0c;并进行登录。创建应用&#xff1a;在淘宝开放平台上创建一个应用&#xff0c;并获取该应用的App Key和App Secret&#xff0c;用于后续的接口调用。…
阅读更多...
Python“梦寻”淘宝天猫店铺所有数据接口，淘宝店铺所有商品数据API接口，淘宝API接口申请指南(含代码示例）

Python“梦寻”淘宝天猫店铺所有数据接口，淘宝店铺所有商品数据API接口，淘宝API接口申请指南(含代码示例）

获取淘宝店铺所有商品数据的接口可以通过淘宝开放平台获取。 具体操作步骤如下&#xff1a; 在淘宝开放平台注册成为开发者&#xff0c;并创建一个应用&#xff0c;获取到所需的 App Key 和 App Secret 等信息。使用获取到的 App Key 和 App Secret&#xff0c;进行签名和认证…
阅读更多...
Android组件通信——Intent（二十三）

Android组件通信——Intent（二十三）

1. 认识Intent 1.1 知识点 &#xff08;1&#xff09;了解Intent的主要作用&#xff1b; &#xff08;2&#xff09;掌握Activity程序对Intent操作的支持&#xff1b; &#xff08;3&#xff09;可以使用Intent完成Activity程序间的跳转&#xff0c;也可以通过Intent接收返…
阅读更多...
QT基础 QChart绘制折线

QT基础 QChart绘制折线

目录 1.简单折线 2.数学折线 3.可滑动折线 1.简单折线 //![1] //! 折现段坐标QLineSeries *series new QLineSeries(); //![1]//![2] //! 添加点series->append(0, 6);series->append(2, 4);series->append(3, 8);series->append(7, 4);series->append(10, 5)…
阅读更多...
YOLOv7改进： CFP：即插即用的多尺度融合模块，EVC助力小目标检测| 顶刊TIP 2023

YOLOv7改进： CFP：即插即用的多尺度融合模块，EVC助力小目标检测| 顶刊TIP 2023

💡💡💡本文独家改进：即插即用的多尺度融合模块，EVC助力小目标检测 EVC | 亲测在多个数据集实现暴力涨点，强烈推荐，独家首发； 收录： YOLOv7高阶自研专栏介绍：http://t.csdnimg.cn/tYI0c ✨✨✨前沿最新计算机顶会复现 🚀🚀🚀YOLOv7自研创新结合，轻松搞…
阅读更多...
MS31804四通道低边驱动器可pin对pin兼容DRV8804

MS31804四通道低边驱动器可pin对pin兼容DRV8804

MS31804TE 是一个具有过流保护功能的四通道低边驱动器。MS31804TE 内置钳位二极管&#xff0c;用来钳制由电感负载续流产生的电压。MS31804TE 可以驱动单极步进电机、直流电机、继电器、螺线管或者其它负载。 散热良好的情况下&#xff0c;MS31804TE 可以提供每个通道最高 2A 的…
阅读更多...
整理笔记——二极管

整理笔记——二极管

一、什么是二极管 二极管是一种由半导体材料制成的一种具有单向导电性能的电子元器件&#xff0c;二极管的核心是PN结。 加在二极管两端的电压和通关的电流被成为&#xff0c;二极管的伏安特性曲线 ​​​ 二极管的正向特性&#xff1a;起初正向电压较小时&#xff0c;正向电流…
阅读更多...
[ACTF2020 新生赛]Exec1

[ACTF2020 新生赛]Exec1

拿到题目&#xff0c;不知道是sql注入还是命令执行漏洞 先ping一下主机 有回显&#xff0c;说明是命令执行漏洞 我们尝试去查看目录 127.0.0.1|ls&#xff0c;发现有回显&#xff0c;目录下面有个index.php的文件 我们之间访问index.php 输入127.0.0.1;cat index.php 发现又…
阅读更多...
基于Springboot实现汽车租赁平台管理系统项目【项目源码】

基于Springboot实现汽车租赁平台管理系统项目【项目源码】

基于Springboot实现汽车租赁平台管理系统演示 JAVA简介 Java主要采用CORBA技术和安全模型&#xff0c;可以在互联网应用的数据保护。它还提供了对EJB&#xff08;Enterprise JavaBeans&#xff09;的全面支持&#xff0c;java servlet API&#xff0c;JSP&#xff08;java ser…
阅读更多...
选择适合自身业务的HTTP代理有哪些因素决定？

选择适合自身业务的HTTP代理有哪些因素决定？

相信对很多爬虫工作者和数据采集的企业来说&#xff0c;如何选购适合自己业务的HTTP代理是一个特别特别困扰的选题&#xff0c;市面上那么多HTTP代理厂商&#xff0c;好像这家有这些缺点&#xff0c;转头又看到另外一家的缺点&#xff0c;要找一家心仪的仿佛大海捞针。今天我们…
阅读更多...
Table.Group系列_第4参数为全局的情况下，利用第五参数进行分组汇总

Table.Group系列_第4参数为全局的情况下，利用第五参数进行分组汇总

原始数据： 部门与职位存在于同一列中 实现功能： 根据筛选条件，可对部门或职位进行统计汇总第一列列名根据筛选自动变更，显示当前统计的维度 实现方式： 1. 构建筛选器内容 在任意空白单元格内输入需要筛选的内容 2. 插入"组合框"控件，并进行相应设置 从开发工具…
阅读更多...
flex布局在多层嵌套时，内层设置了justify-content: space-between；不生效问题

flex布局在多层嵌套时，内层设置了justify-content: space-between；不生效问题

内层的地址和时间这一行&#xff0c;设置了justify-content: space-between;但并不生效&#xff0c;原因是要在上一层.center 设置 flex:1;&#xff08; 重点&#xff09; 经常忘记&#xff0c;特在此记录一下&#xff0c;以下是代码 <view class"index-card" c…
阅读更多...
深度学习基础知识 BatchNorm、LayerNorm、GroupNorm的用法解析

深度学习基础知识 BatchNorm、LayerNorm、GroupNorm的用法解析

深度学习基础知识 BatchNorm、LayerNorm、GroupNorm的用法解析 1、BatchNorm2、LayerNorm3、GroupNorm用法&#xff1a; BatchNorm、LayerNorm 和 GroupNorm 都是深度学习中常用的归一化方式。 它们通过将输入归一化到均值为 0 和方差为 1 的分布中&#xff0c;来防止梯度消失和…
阅读更多...
学生用的台灯哪种比较好？分享专家推荐的学生台灯

学生用的台灯哪种比较好？分享专家推荐的学生台灯

对于学生来说&#xff0c;台灯是必不可少的一盏学习照明灯具&#xff0c;它能提供室内不足的光线、亮度&#xff0c;基本每个学生在宿舍、家里都会备着一台。不过台灯也并不是随便挑选一台使用就可以的&#xff0c;很多学生就是因为使用了一些价格低廉、质量安全没有保障的台灯…
阅读更多...
【大数据】hadoop安装部署（学习笔记）

【大数据】hadoop安装部署（学习笔记）

一、集群组成概述 Hadoop集群包括两个集群&#xff1a;HDFS集群、YARN集群 两个集群逻辑上分离、通常物理上在一起 两个集群都是标准的主从架构集群 HDFS集群&#xff08;分布式存储&#xff09;&#xff1a; 主角色&#xff1a;NameNode从角色&#xff1a;DataNode主角色…
阅读更多...
代码随想录算法训练营第五十八天 |583. 两个字符串的删除操作、72. 编辑距离、编辑距离总结篇

代码随想录算法训练营第五十八天 |583. 两个字符串的删除操作、72. 编辑距离、编辑距离总结篇

一、583. 两个字符串的删除操作 题目链接/文章讲解/视频讲解&#xff1a;代码随想录 思考&#xff1a; 1.确定dp数组&#xff08;dp table&#xff09;以及下标的含义 dp[i][j]&#xff1a;以i-1为结尾的字符串word1&#xff0c;和以j-1位结尾的字符串word2&#xff0c;想要达…
阅读更多...
postman 密码rsa加密登录-2加密密码

postman 密码rsa加密登录-2加密密码

上一篇讲了获取公钥&#xff0c;将环境准备好之后&#xff0c;在登录接口的Pre-request Scrip 里&#xff0c;使用公钥进行加密后在正常登录。本文采用的方案是使用第三方模块forge.js来实现加密。 1、环境准备好&#xff0c;系统git 和node都OK。下载forge.js git clone htt…
阅读更多...
Java-Atomic原子操作类详解及源码分析，Java原子操作类进阶，LongAdder源码分析

Java-Atomic原子操作类详解及源码分析，Java原子操作类进阶，LongAdder源码分析

文章目录 一、Java原子操作类概述1、什么是原子操作类2、为什么要用原子操作类3、CAS入门 二、基本类型原子类1、概述2、代码实例 三、数组类型原子类1、概述2、代码实例 四、引用类型原子类1、概述2、AtomicReference3、ABA问题与AtomicStampedReference4、一次性修改&#xf…
阅读更多...
SpringBoot (1)

SpringBoot (1)

目录 1 入门案例 1.1 环境准备 1.2 编写pom.xml 1.3 编写入口程序 1.4 编写接口 1.5 编写配置 1.6 快速部署 1.6.1 打jar包 1.6.2 部署 1.7 访问接口 2 全注解开发 2.1 常用注解 2.2 属性绑定注解 2.2.1 注册组件 2.2.2 ConfigurationProperties(prefix"te…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023