jdbc连接teradata仓库_kafka-connect-jdbc-source 简介

你的名字 2023-01-03 04:46 194阅读 0赞

![f9e9c5987cf2f3d2324aafee77281c05.png][]

> 简介

kafka-connect-jdbc-source 可以从任意 JDBC 驱动相关数据库同步数据到Kafka主题中。通过使用 JDBC ，连接器支持大范围多种数据库而且不需要自己编码。

数据加载是通过定期执行查询SQL，为结果的每一条数据创建输出记录。默认情况下，加载数据库所有表数据到对应主题。自动调整监控数据库中新增表或删除表。连接器可以通过监控指定字段完成新增/修改数据的加载。

> 特点

kafka-connect-jdbc-source最大的特点是用户可以定义加载数据的条件。

Kafka Connect 从每张表最开始记录，所以下一次可以从一个正确的位置循环。source 连接器通过这种方法获取表的最新数据。支持以下几种模式，每种模式检测新增条件都是不同的。

*  增长列模式

一行包含唯一 ID 列，新增行比上一行ID大，比如自增列。注意，这种模式只能监控新增数据，更新数据不能被检测到。所以这种模式只能用于不变的数据，即只增加。在数据仓库中流化事实表时，可能会使用此模式的一个示例，因为这些表通常是仅插入的。递增列必须是整数类型。

*  时间戳模式

在这种模式下，单列包含修改时间通常用来追踪最终修改的数据，然后查询从某个时间开始改变的数据，包括更新的。注意timestamp不要求必须唯一，这种模式不能保证所有更新数据会被发送，当查询返回两条相同时间的数据，只有一条能够成功发送。

*  时间戳和增长列模式

这是最可靠和正确的模式，结合时间戳和增长列，通过结合这两种模式，只要时间戳足够细粒度，每一条数据都有唯一标示去更新。

*  自定义查询SQL模式

source 连接器支持自定义查询而不是查询整张表。在自定义查询下，自动更新模式可以使用只要where后追加正确的查询条件。或者通过自动更新查询条件来过滤新数据。然而，这种模式下不会记录查询offset，所以需要SQL条件可以自动更新，以更新每次查询。

*  批量模式

这种模式没有过滤，因此不需要增长。它会加载表中所有数据在每次查询时。这种模式适用于每次全量导出整张表。

> 消息键

Kafka 消息是 KV 对。对于 JDBC Connect来说 ，value 是从表中抓取的数据，但是 JDBC Connect 默认不生成 key。

在分区策略中，消息键是非常有用的，Key 可以将消息直接发送到指定分区，供下游处理使用。如果没有 key ，那消息发送就采用 round-robin 分发。

对 JDBC Connect数据设置消息键，有两种transform可以使用。ValueToKey 和 ExtractField。如下片段，account 表中的 id 字段作为消息键。

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
           "name": "jdbc_source_mysql_01",
           "config": {
                   "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                   "connection.url": "jdbc:mysql://mysql:3306/test",
                   "connection.user": "connect_user",
                   "connection.password": "connect_password",
                   "topic.prefix": "mysql-01-",
                   "poll.interval.ms" : 3600000,
                   "table.whitelist" : "test.accounts",
                   "mode":"bulk",
                   "transforms":"createKey,extractInt",
                   "transforms.createKey.type":"org.apache.kafka.connect.transforms.ValueToKey",
                   "transforms.createKey.fields":"id",
                   "transforms.extractInt.type":"org.apache.kafka.connect.transforms.ExtractField$Key",
                   "transforms.extractInt.field":"id"
                   }
           }'

[f9e9c5987cf2f3d2324aafee77281c05.png]: /images/20221119/03a37abe79d1424a9a816499b0eb70db.png