TiDB系列之:使用Flink TiDB CDC Connector采集数据
TiDB系列之:使用Flink TiDB CDC Connector采集数据
- 一、依赖项
- 二、Maven依赖
- 三、SQL Client JAR
- 四、如何创建 TiDB CDC 表
- 五、连接器选项
- 六、可用元数据
- 七、特征
- 一次性处理
- 启动阅读位置
- 多线程读取
- DataStream Source
- 八、数据类型映射
TiDB CDC 连接器允许从 TiDB 数据库读取快照数据和增量数据。本文档介绍如何设置 TiDB CDC 连接器以对 TiDB 数据库运行 SQL 查询。
- TiDB系列之:使用TiCDC增量同步TiDB数据库数据
- TiDB系列之:TiCDC同步数据到Kafka集群使用Debezium数据格式
- TiDB系列之:TiCDC同步TiDB数据库数据到Kafka集群Topic
一、依赖项
为了设置 TiDB CDC 连接器,下表提供了使用构建自动化工具(例如 Maven 或 SBT)的项目和带有 SQL JAR 包的 SQL Client 的依赖信息。
二、Maven依赖
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-tidb-cdc</artifactId><version>3.0.1</version>
</dependency>
三、SQL Client JAR
下载链接仅适用于稳定版本。
下载 flink-sql-connector-tidb-cdc-3.0.1.jar 并将其放在 <FLINK_HOME>/lib/ 下。
四、如何创建 TiDB CDC 表
TiDB CDC 表可以定义如下:
-- checkpoint every 3000 milliseconds
Flink SQL> SET 'execution.checkpointing.interval' = '3s'; -- register a TiDB table 'orders' in Flink SQL
Flink SQL> CREATE TABLE orders (order_id INT,order_date TIMESTAMP(3),customer_name STRING,price DECIMAL(10, 5),product_id INT,order_status BOOLEAN,PRIMARY KEY(order_id) NOT ENFORCED) WITH ('connector' = 'tidb-cdc','tikv.grpc.timeout_in_ms' = '20000', 'pd-addresses' = 'localhost:2379','database-name' = 'mydb','table-name' = 'orders'
);-- read snapshot and binlogs from orders table
Flink SQL> SELECT * FROM orders;
五、连接器选项
参数 | 是否必须 | 默认值 | 类型 | 描述 |
---|---|---|---|---|
connector | required | (none) | String | 指定使用什么连接器,这里应该是“tidb-cdc”。 |
database-name | required | (none) | String | 要监控的 TiDB 服务器的数据库名称。 |
table-name | required | (none) | String | 要监控的 TiDB 数据库的表名。 |
scan.startup.mode | optional | initial | String | TiDB CDC Consumer 可选的启动模式,有效枚举为“initial”和“latest-offset”。 |
pd-addresses | required | (none) | String | TiKV 集群的 PD 地址。 |
tikv.grpc.timeout_in_ms | optional | (none) | Long | TiKV GRPC 超时(以毫秒为单位)。 |
tikv.grpc.scan_timeout_in_ms | optional | (none) | Long | TiKV GRPC 扫描超时(以毫秒为单位)。 |
tikv.batch_get_concurrency | optional | 20 | Integer | TiKV GRPC 批量获取并发。 |
tikv.* | optional | (none) | String | 传递 TiDB 客户端的属性。 |
六、可用元数据
以下格式元数据可以在表定义中公开为只读(虚拟)列。
key | DataType | 描述 |
---|---|---|
table_name | STRING NOT NULL | 包含该行的表的名称。 |
database_name | STRING NOT NULL | 包含该行的数据库的名称。 |
op_ts | TIMESTAMP_LTZ(3) NOT NULL | 它指示在数据库中进行更改的时间。 |
如果记录是从表的快照而不是binlog中读取的,则该值始终为0。 |
扩展的 CREATE TABLE 示例演示了公开这些元数据字段的语法:
CREATE TABLE products (db_name STRING METADATA FROM 'database_name' VIRTUAL,table_name STRING METADATA FROM 'table_name' VIRTUAL,operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,order_id INT,order_date TIMESTAMP(0),customer_name STRING,price DECIMAL(10, 5),product_id INT,order_status BOOLEAN,PRIMARY KEY(order_id) NOT ENFORCED
) WITH ('connector' = 'tidb-cdc','tikv.grpc.timeout_in_ms' = '20000','pd-addresses' = 'localhost:2379','database-name' = 'mydb','table-name' = 'orders'
);
七、特征
一次性处理
TiDB CDC 连接器是一个 Flink Source 连接器,它会先读取数据库快照,然后继续读取更改事件,即使发生故障也只处理一次。
启动阅读位置
配置选项 scan.startup.mode 指定 TiDB CDC Consumer 的启动模式。有效的枚举是:
- initial(默认):拍摄捕获表的结构和数据的快照;如果您想从捕获的表中获取数据的完整表示,则很有用。
- latest-offset:仅对捕获的表的结构进行快照;如果只需要获取从现在开始发生的更改,则很有用。
多线程读取
TiDB CDC 源可以并行读取工作,因为有多个任务可以接收更改事件。
DataStream Source
TiDB CDC 连接器也可以是 DataStream 源。您可以创建一个 SourceFunction,如下所示:
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;import org.apache.flink.cdc.connectors.tidb.TDBSourceOptions;
import org.apache.flink.cdc.connectors.tidb.TiDBSource;
import org.apache.flink.cdc.connectors.tidb.TiKVChangeEventDeserializationSchema;
import org.apache.flink.cdc.connectors.tidb.TiKVSnapshotEventDeserializationSchema;
import org.tikv.kvproto.Cdcpb;
import org.tikv.kvproto.Kvrpcpb;import java.util.HashMap;public class TiDBSourceExample {public static void main(String[] args) throws Exception {SourceFunction<String> tidbSource =TiDBSource.<String>builder().database("mydb") // set captured database.tableName("products") // set captured table.tiConf(TDBSourceOptions.getTiConfiguration("localhost:2399", new HashMap<>())).snapshotEventDeserializer(new TiKVSnapshotEventDeserializationSchema<String>() {@Overridepublic void deserialize(Kvrpcpb.KvPair record, Collector<String> out)throws Exception {out.collect(record.toString());}@Overridepublic TypeInformation<String> getProducedType() {return BasicTypeInfo.STRING_TYPE_INFO;}}).changeEventDeserializer(new TiKVChangeEventDeserializationSchema<String>() {@Overridepublic void deserialize(Cdcpb.Event.Row record, Collector<String> out)throws Exception {out.collect(record.toString());}@Overridepublic TypeInformation<String> getProducedType() {return BasicTypeInfo.STRING_TYPE_INFO;}}).build();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// enable checkpointenv.enableCheckpointing(3000);env.addSource(tidbSource).print().setParallelism(1);env.execute("Print TiDB Snapshot + Binlog");}
}
八、数据类型映射
TiDB type | Flink SQL type | NOTE |
---|---|---|
TINYINT | TINYINT | |
SMALLINT、TINYINT UNSIGNED | SMALLINT | |
INT、MEDIUMINT、SMALLINT UNSIGNED | INT | |
BIGINT、INT UNSIGNED | BIGINT | |
BIGINT UNSIGNED | DECIMAL(20, 0) | |
FLOAT | FLOAT | |
REAL、DOUBLE | DOUBLE | |
NUMERIC(p, s) DECIMAL(p, s) where p <= 38 | DECIMAL(p, s) | |
NUMERIC(p, s) DECIMAL(p, s) where 38 < p <= 65 | STRING | 在 TiDB 中 DECIMAL 数据类型的精度最高为 65,但在 Flink 中 DECIMAL 的精度限制为 38。因此,如果定义精度大于 38 的十进制列,则应将其映射到 STRING 以避免精度损失。 |
BOOLEAN、TINYINT(1)、BIT(1) | BOOLEAN | |
DATE | DATE | |
TIME [§] | TIME [§] | |
TIMESTAMP [§] | TIMESTAMP_LTZ [§] | |
DATETIME [§] | TIMESTAMP [§] | |
CHAR(n) | CHAR(n) | |
VARCHAR(n) | VARCHAR(n) | |
BIT(n) | BINARY(⌈n/8⌉) | |
BINARY(n) | BINARY(n) | |
TINYTEXT、TEXT、MEDIUMTEXT、LONGTEXT | STRING | |
TINYBLOB、BLOB、MEDIUMBLOB、LONGBLOB | BYTES | 目前,TiDB 中的 BLOB 数据类型仅支持长度不大于 2,147,483,647(2 ** 31 - 1) 的 Blob。 |
YEAR | INT | |
ENUM | STRING | |
JSON | STRING | JSON 数据类型在 Flink 中会被转换为 JSON 格式的 STRING。 |
SET | ARRAY | 由于 TiDB 中的 SET 数据类型是一个字符串对象,可以有零个或多个值,因此它应该始终映射到字符串数组 |
相关文章:

TiDB系列之:使用Flink TiDB CDC Connector采集数据
TiDB系列之:使用Flink TiDB CDC Connector采集数据 一、依赖项二、Maven依赖三、SQL Client JAR四、如何创建 TiDB CDC 表五、连接器选项六、可用元数据七、特征一次性处理启动阅读位置多线程读取DataStream Source 八、数据类型映射 TiDB CDC 连接器允许从 TiDB 数…...

每日一道算法题 最接近的三数之和
题目 16. 最接近的三数之和 - 力扣(LeetCode) Python class Solution:def threeSumClosest(self, nums: List[int], target: int) -> int:nums.sort()nlen(nums)ans0min_diffinf # infinite 无穷for i in range(n-2):tmpnums[i]li1rn-1while l<…...

搭建自己的金融数据源和量化分析平台(六):下载并存储沪深两市上市公司财报
基于不依赖wind、某花顺等第三方平台数据的考虑,尝试直接从财报中解析三大报表进而计算ROE等财务指标,因此需要下载沪深两市的上市公司财报数据,便于后续从pdf中解析三大报表。 深市爬虫好做,先放深市爬虫: 根据时间段…...

C语言-常见关键字详解
一、const 关键字const用于声明常量,赋值后,其值不能再被修改。 示例: const int MAX_COUNT 100; 二、static static关键字在不同情境下有不同作用: 1.函数中的静态变量:保留变量状态,仅初始化一次&a…...

异步编程之std::future(一): 使用
目录 1.概述 2.std::future的基本用法 3.使用 std::shared_future 4.std::future的使用场景 5.总结 1.概述 在编程实践中,我们常常需要使用异步调用。通过异步调用,我们可以将一些耗时、阻塞的任务交给其他线程来执行,从而保证当前线程的…...

Vue3 + JS项目配置ESLint Pretter
前言 如果在开发大型项目 同时为多人协作开发 那么 ESLint 在项目中极为重要 在使用 ESLint 的同时 也需要使用 Pretter插件 统一对代码进行格式化 二者相辅相成 缺一不可 1. 安装 VsCode 插件 在 VsCode 插件市场搜索安装 ESLint 和 Pretter 2. 安装依赖 这里直接在 pac…...

JavaScript (十四)——JavaScript typeof和类型转换
目录 JavaScript typeof, null, 和 undefined typeof 操作符 null undefined undefined 和 null 的区别 JavaScript 类型转换 JavaScript 数据类型 JavaScript 类型转换 将数字转换为字符串 将布尔值转换为字符串 将日期转换为字符串 将字符串转换为数字 一元运算符…...

CTF-web 基础
网络协议 OSI七层参考模型:一个标准的参考模型 物理层 网线,网线接口等。 数据链路层 可以处理物理层传入的信息。 网络层 比如IP地址 传输层 控制传输的内容的传输,在传输的过程中将要传输的信息分块传输完成之后再进行合并。 应用…...

CP AUTOSAR标准之ChineseV2XNetwork(AUTOSAR_SWS_ChineseV2XNetwork)(更新中……)
1 简介和功能概述 本文档指定了AUTOSAR基础软件模块中国车辆对接网络(CnV2xNet)的功能、API和配置。 中国车联网网络(CnV2xNet)与中国车联网消息(CnV2xMsg)、中国车联网管理(CnV2xMgt)、中国车联网安全(CnV2xSec)以及AUTOSAR BSW模块以太网接口(EthIf)共同构成了AUTOSAR架构…...

【hloc】 项目流程
hloc 项目流程 1. 数据集准备2. 特征提取3. 匹配特征4. 三维重建5. 定位6. 结果评估7. 示例脚本 这个项目涉及到了视觉定位和三维重建的一系列步骤,从特征提取、匹配、三维重建到定位和结果评估。通过提供的脚本文件,用户可以方便地运行整个流程。 1. 数…...

鸿蒙系统开发【应用接续】基本功能
应用接续 介绍 基于ArkTS扩展的声明式开发范式编程语言编写的一个分布式视频播放器,主要包括一个直播视频播放界面,实现视频播放时可以从一台设备迁移到另一台设备继续运行,来选择更合适的设备继续执行播放功能以及PAD视频播放时协同调用手…...

nextTick方法的作用是什么?什么时候会用到
nextTick 方法在 Vue.js 中扮演着重要的角色,它用于在下次 DOM 更新循环结束之后执行延迟回调。这主要用于确保在 Vue 完成 DOM 更新后执行依赖于 DOM 的操作。 作用 确保 DOM 更新完成:Vue 的 DOM 更新是异步的,当你修改了数据后࿰…...

多 NodeJS 环境管理
前言 对于某个项目依赖特定版本的 NodeJS,或几个项目的 NodeJS 版本冲突时,需要在系统中安装多个版本的 NodeJS,这时可以使用一些工具来进行多个 NodeJS 的管理。 有很多类似的 NodeJS 管理工具,如 nvm, nvs, n 等,接…...

解决网站被植入跳转木马病毒
概述 网站被植入跳转木马病毒是一种常见的安全威胁,它可能导致网站用户被重定向到恶意站点。本文将指导您如何检测、清除这类木马病毒以及采取预防措施。 步骤1:确认感染 首先,需要确认您的网站确实受到了跳转木马的影响。 示例ÿ…...

Node.js(6)——npm软件包管理
npm npm是Node.js标准的软件包管理器。 使用: 初始化清单文件:npm init-y(得到package.json文件,有则略过此命令)下载软件包:npm i 软件包名称使用软件包 示例: 初始状态下npm文件夹下只有server.js,下载软件包前看…...

区块链核心概念与技术架构简介
引言 区块链,一种分布式账本技术,不仅为数字货币提供了基础设施,更在金融、供应链、物联网等多个领域展现出广泛的应用前景。区块链技术被认为是继蒸汽机、电力、互联网之后,下一代颠覆性的核心技术。 如果说蒸汽机释放了人们的…...

≌图概念凸显包含射线V的直线W是比V长的线
黄小宁 x轴中:各非负数点xh≥0都变回自己即都作恒等变换,其余点x-h都变号为xh就使x轴失去负数点而变为射线V{xh≥0}。这x轴变为射线V⊂x轴是不保距变换即不是x轴的刚体运动使x轴不≌V⊂x轴(小学生都知道x轴不≌射线V)。据≌图概念…...

子路由的配置方法?
子路由的配置方法主要涉及到在Vue-router中定义嵌套路由,即一个路由内部包含多个子路由。以下是配置子路由的基本步骤: 1. 定义父路由 首先,在Vue Router中定义父路由。父路由可以像其他普通路由一样定义,但通常会有一个组件与之…...

【大模型从入门到精通2】openAI api的入门介绍2
互动对话界面的搭建 让我们来看看如何建立一个互动对话界面,用户可以在此输入查询,系统实时处理并显示响应。 import panel as pn # 用于构建图形用户界面# 初始化对话历史记录和GUI组件 conversation_history [] input_widget pn.widgets.TextInpu…...

【前端编程小白】的HTML从零入门到实战
之前有高中毕业生读了博客,想让我帮他找一些前端入门的内容,他们报的计算机专业,想利用开学前夕学习一下,我给他推荐了一些菜鸟教程呀什么的。后来想,看来还是很多人需要一些更加入门的可成的,而且很多教程…...

easyexcel读文件入批量入es
1. 封装实体类,并对应excel表中的列 Data public class User {private String md5;private String id; ExcelProperty(value "age")private String age;ExcelProperty(value "username")private String name;} 2. 批量入库 private void in…...

JS+H5打字练习器
实现功能 1.导入.TXT文件到打字练习内容(部分浏览器可能出于安全问题限制了这一步操作) 2.输入文本到打字练习内(弹出输入框,将要练习的内容输入至输入框) 3. 开始练习,并根据正误在打字练习内容文本上修…...

windows系统关闭开机自检硬盘
效果: 注册表关闭开机硬盘自检,你可以按照以下步骤操作: 打开注册表编辑器: 按 Win R 键打开“运行”对话框。输入 regedit 并按回车,打开注册表编辑器。 定位到自检相关的键: 依次展开以下路径&#x…...

【多线程开发 5】实践使用Lock和Condition
Lock和Condition Lock 线程之间同步或者竞争都需要锁这类结构,一般我们都会用Object的wait和signal搭配synchronized关键字进行多线程开发,但是很多时候会造成死锁的现象,这是因为synchroniezd无法破坏死锁的产生条件,但是Lock接…...

2.4-结构化并发:协程的结构化异常管理
文章目录 协程结构化异常流程协程结构化异常流程和取消流程的区别子协程异常为什么要连带取消父协程? CoroutineExceptionHandler异常协程异常的最后一道拦截:CoroutineExceptionHandlerCoroutineExceptionHandler 为什么只能设置给最外层协程才有效&…...

Android 12.0 debug版本打开OEM解锁开关功能实现
通常为了方便push在debug版本会采用如下命令 adb root adb disable-verity 提示: Device is locked. Please unlock the device first. 查找日志可以发现system/core/set-verity-state/set-verity-state.cpp文件中is_avb_device_locked方法里 这个获取ro.boot…...

linux用户组练习
准备工作 [rootlocalhost ~]# watch -n 1 tail -n 5 /etc/group使用watch 动态监控 1.建立用户组 shengcan,其id 为2000 2.建立用户组 caiwu,其id 为 2001 3.足建立用户组 jishu,其id 为 2002 4.建立用户lee,指定其主组id为sh…...

[Docker][Docker Container]详细讲解
目录 1.什么是容器?2.容器命令1.docker creatre2.docker run3.docker ps4.docker logs5.docker attach6.docker exec7.docker start8.docker stop9.docker restart10.docker kill11.docker top12.docker stats13.docker container inspect14.docker port15.docker c…...

塑造美好心灵,激发创造活力|第三届瓷艺中华“陶溪川杯”儿童青少年陶瓷作品展开展
第三届瓷艺中华“陶溪川杯”儿童青少年陶瓷作品展 展览现场 由中央美术学院、景德镇陶瓷大学、景德镇陶文旅控股集团共同主办,由中国非物质文化遗产保护协会陶瓷分会、中国文化艺术发展促进会陶瓷专业委员会、中央美术学院陶瓷艺术研究院、中央美术学院少儿美术教…...

鸿蒙开发刷新单个item会闪一下处理
鸿蒙开发刷新单个item会闪一下 首先我用的是懒加载方式,改变某位数据后我调listener.onDataChange(index),发现item的改动是变了,但是item也闪了一下。 先分析为什么item会闪一下 其他是因为item上有图片,加载的网络图。你onDataChange(index)时,它会重新加载这一item,…...