Kafka Connect JDBC Connector学习文档

前言:

最近需要调研Kafka Connect可连接哪些数据源,特此学习官网并翻译了下文档。
Confluent JDBC Connector官网地址(官网布局可能有些凌乱)
本文为纯理论学习,实践请看下文 Kafka Connect 实现MySQL增量同步
希望看过的小伙伴点个❤,谢谢。

JDBC Connector

JDBC connector允许您通过JDBC驱动程序将任何关系型数据库中的数据导入到Kafka的主题Topic中。通过使用JDBC,这个连接器可以支持各种数据库,不需要为每个数据库定制代码。
通过定期地执行SQL查询语句并为结果集中的每一行创建输出记录来加载数据。在默认情况下,在一个数据库中的所有表都会被复制,每个表都复制到自己的输出主题。数据库那些新的或删除的表被监视并自动适应调整。当从表中复制数据时,连接器可以仅仅加载新增修改的行通过指定哪些列应当被用来发现新增或修改的数据。

Quickstart

为了了解连接器的基本功能,我们将从本地SQLite数据库中复制单个表。在这个简单的示例中,我们假设表中的每个条目都被分配了一个唯一的ID,并且在创建之后不会对其进行修改。Confluent Platform内置包含了用于SQLitePostgreSQL的JDBC驱动程序,但是如果使用的是不同的数据库(如MySQL),则还需要确保JDBC驱动程序在Kafka Connect进程的CLASSPATH上可用。

快速开始

创建表中测试数据

sqlite> CREATE TABLE accounts(id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, name VARCHAR(255));
sqlite> INSERT INTO accounts(name) VALUES('alice');
sqlite> INSERT INTO accounts(name) VALUES('bob');

创建一个配置文件,用于从该数据库中加载数据。此文件包含在etc/kafka-connect-jdbc/quickstart-sqlite.properties中的连接器中,并包含以下设置:
(学习了解配置结构即可)

name=test-sqlite-jdbc-autoincrement
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:sqlite:test.db
mode=incrementing
incrementing.column.name=id
topic.prefix=test-sqlite-jdbc-

前几个设置是您将为所有连接器指定的常见设置。connection.url指定要连接的数据库,在本例中是本地SQLite数据库文件。mode指示我们想要如何查询数据。在本例中,我们有一个自增的唯一ID,因此我们选择incrementing递增模式并设置incrementing.column.name递增列的列名为id。在这种mode模式下,每次查询新的数据将只返回那些ID大于之前查询中最大ID的行。最后,我们可以通过设置topic.prefix,来控制每个表的输出发送到的Topic主题的名称。因为现在我们只有一个表,所以本例中唯一的输出主题将会是test-sqlite-jdbc-accounts。

在另一个终端的独立的Kafka Connect中运行(运行刚刚的配置文件)

$ ./bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/quickstart-sqlite.properties

可以看到进程启动并记录一些消息,然后它将开始执行查询并将结果发送到Topic

$ ./bin/kafka-avro-console-consumer --new-consumer --bootstrap-server localhost:9092 --topic test-sqlite-jdbc-accounts --from-beginning
{"id":1,"name":{"string":"alice"}}
{"id":2,"name":{"string":"bob"}}

默认的轮询间隔是5秒,因此可能需要几秒钟才能显示。根据您预期的速率更新或期望的延迟,可以使用较短的轮询间隔更快地显示。

Features 特性

JDBC connector支持使用各种JDBC数据类型来复制表,动态地从数据库中添加和删除表、whitelists白名单和blacklists黑名单、更改轮询间隔以及其他设置。然而,对于大多数用户来说,最重要的特性是控制如何从数据库增量地复制数据的设置。
Kafka Connect跟踪监测从每个表中检索到的最新记录,因此在下一次迭代循环中(或者在崩溃的情况下),它可以在正确的位置开始。JDBC connector使用此功能,仅在每次迭代中从表(或从自定义查询的输出)获取更新过updated的行。支持多种模式,每一种模式在修改行上的不同都能被发现查明。

Incremental Query Modes 增量查询模式

每一种增量查询模式跟踪每行的一组列,用途来跟踪哪些行已被处理以及哪些行是新增的或者已经被更新的。mode模式的设置控制它的行为,并支持以下选项:

  1. Incrementing Column 自增列
    每行包含唯一ID的列,它的较新的行保证具有较大的ID,即AUTOINCREMENT自增列。注意,此模式只能检测发现新增的行。对于存在的记录行的更新是无法检测到的,因此此模式应当仅用于那些不变的数据行。举个例子,是在数据仓库中流式传输事实表时,因为事实表通常是只插入的。
  2. Timestamp Column 时间戳列
    在这种模式下,一个列包含修改时间戳用来跟踪最近一次latest time处理的数据,并且仅查询从那时起已经修改的行。注意,由于时间戳不一定是唯一的,因此此模式不能保证所有更新的数据都将被传递:如果2行共有相同的时间戳,并且通过增量查询返回,在崩溃之前只处理了一行,那么当系统恢复时,将错过第二个修改更新。
  3. Timestamp and Incrementing Columns 时间戳和自增列
    这是将一个递增列和一个时间戳列组合在一起的最稳健和精确的模式。通过组合这两者,只要时间戳足够细,每个(id,timestamp)元组将唯一地标识对行的更新。即使部分完成后更新失败,当系统恢复时,仍然可以正确地检测和传递未处理的更新。
  4. Custom Query 定制化查询
    JDBC connector支持使用定制查询而不是复制整个表。通过自定义查询,可以使用其他自动更新模式之一,只要必要的WHERE子句可以正确地附加到查询中。或者,指定的查询可以处理对新的修改更新本身的过滤;然而,请注意,将不执行offset偏移量跟踪(不同于自动模式,其中为每个记录记录递增和/或时间戳列值),因此查询必须自己跟踪offset偏移量。
  5. Bulk 大块查询
    此模式未经过滤,因此根本不递增。它将在每次迭代中从表中加载所有行。如果希望周期性地消费转储整个表,其中条目最终会被删除,并且下游系统可以安全地处理副本,那么这非常有用。

Configuration 配置

JDBC Drivers

JDBC connector在可以从数据库导入数据以及如何导入数据的数据库中提供了相当大的灵活性。本节首先描述如何访问那些其驱动程序不包含在Confluent Platform中的数据库,然后给出几个覆盖常见场景的示例配置文件,然后提供可用配置选项的详尽描述。

JDBC connector在通用JDBC API上实现数据复制的功能,但是依赖于JDBC驱动来处理这些API的数据库特定实现。Confluent Platform附带了一些JDBC驱动程序,但是如果没有包括你用的数据库的驱动程序,则需要通过CLASSPATH使其可用。

一种选择是在连接器安装JDBC驱动程序jar。相对于安装目录,打包的连接器安装在share/java/kafka-connect-jdbc目录中。如果您已经从DebianRPM包安装,连接器将安装在/usr/share/java/kafka-connect-jdbc中。如果你是从zip或tar文件安装,连接器将安装在上面给出的路径中,该路径位于您解压缩Confluent Platform归档的目录下。
另一种选择,你也可以设置CLASSPATH变量在运行copycat-standalonecopycat-distributed

$ CLASSPATH=/usr/local/firebird/* ./bin/copycat-distributed ./config/copycat-distributed.properties
jdbc:firebirdsql:localhost/3050:/var/lib/firebird/example.db

Examples*

使用whitelist白名单来限制MySQL数据库表中子集的更改,对于所有白名单表上符合标准的,使用idmodified两列列来检测已修改的行。这种模式是最稳健的,因为它可以将唯一、不可变的行id与修改时间戳相结合,以确保即使过程在增量更新查询中死亡,也不会错过这些更新。

name=mysql-whitelist-timestamp-source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=10

connection.url=jdbc:mysql://mysql.example.com:3306/my_database?user=alice&password=secret
table.whitelist=users,products,transactions

mode=timestamp+incrementing
timestamp.column.name=modified
incrementing.column.name=id

topic.prefix=mysql-

下面这个例子使用了自定义查询query代替加载整个表,允许您从多个表中`join``连接数据。只要查询不包含其自身的过滤,您仍然可以使用内置模式进行增量查询(在本例中,使用时间戳列)。注意,这限制了您对每个连接器的单个输出,并且因为没有表名,所以本例中主题“prefix”实际上是完整的主题名。

name=mysql-whitelist-timestamp-source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=10

connection.url=jdbc:postgresql://postgres.example.com/test_db?user=bob&password=secret&ssl=true
query=SELECT users.id, users.name, transactions.timestamp, transactions.user_id, transactions.payment FROM users JOIN transactions ON (users.id = transactions.user_id)
mode=timestamp
timestamp.column.name=timestamp

topic.prefix=mysql-joined-data

Configuration Options 配置项

connection.url
要加载数据库的JDBC连接URL
-类型: String
-默认: ""
-重要度: 高

topic.prefix
众多表名之前的前缀,向其发布数据的KafkaTopic主题的名称,或者在自定义查询的情况下,生成要向其发布的Topic主题的全名。
-类型: String
-默认: ""
-重要度: 高

mode
incrementing、timestamp、timestamp+incrementing、bulk
-类型: String
-默认: ""
-重要度: 高

poll.interval.ms
以ms为单位轮询每个表中的新数据
-类型: int
-默认: "5000"
-重要度: 高

incrementing.column.name
用于检测新增行的递增列的名称。此列可能不为空。
-类型: String
-默认: ""
-重要度: 中

query
使用自定义查询代替加载表
-类型: String
-默认: ""
-重要度: 中

table.blacklist
不允许复制的表清单。如果指定table.blacklist,则不能设置table.whitelist
-类型: List
-默认: ""
-重要度: 中

table.whitelist
允许复制的表清单。如果指定table.whitelist,则不能设置table.blacklist
-类型: List
-默认: ""
-重要度: 中

timestamp.column.name
时间戳列的名字,用于检测新增或修改的行记录。此列可能不为空。
-类型: List
-默认: ""
-重要度: 中

batch.max.rows
在轮询新数据时,单个批处理中包括的最大行数。此设置可用于限制连接器内部缓冲的数据量。
-类型: int
-默认: 100
-重要度: 低

table.poll.interval.ms
以ms为单位轮询新表或已删除表的频率,这可能导致更新任务配置,以开始轮询已添加表中的数据或停止轮询已删除表中的数据。
-类型: long
-默认: 60000
-重要度: 低

谢谢阅读,有帮助的点个❤!

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 210,978评论 6 490
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 89,954评论 2 384
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 156,623评论 0 345
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,324评论 1 282
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,390评论 5 384
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,741评论 1 289
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,892评论 3 405
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,655评论 0 266
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,104评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,451评论 2 325
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,569评论 1 340
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,254评论 4 328
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,834评论 3 312
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,725评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,950评论 1 264
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,260评论 2 360
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,446评论 2 348

推荐阅读更多精彩内容