Kafka連接器深度解讀之JDBC源連接器

復(fù)制自：https://liyuj.gitee.io/confluent/#介紹

在現(xiàn)實(shí)業(yè)務(wù)中，Kafka經(jīng)常會遇到的一個集成場景就是，從數(shù)據(jù)庫獲取數(shù)據(jù)，因?yàn)殛P(guān)系數(shù)據(jù)庫是一個非常豐富的事件源。數(shù)據(jù)庫中的現(xiàn)有數(shù)據(jù)以及對該數(shù)據(jù)的任何更改都可以流式傳輸?shù)終afka主題中，在這里這些事件可用于驅(qū)動應(yīng)用，也可以流式傳輸?shù)狡渌鼣?shù)據(jù)存儲（比如搜索引擎或者緩存）用于分析等。

實(shí)現(xiàn)這個需求有很多種做法，但是在本文中，會聚焦其中的一個解決方案，即Kafka連接器中的JDBC連接器，講述如何進(jìn)行配置，以及一些問題排查的技巧，至于更多的細(xì)節(jié)，請參見Kafka的文檔。

介紹

Kafka連接器中的JDBC連接器包含在Confluent Platform中，也可以與Confluent Hub分開安裝。它可以作為源端從數(shù)據(jù)庫提取數(shù)據(jù)到Kafka，也可以作為接收端從一個Kafka主題中將數(shù)據(jù)推送到數(shù)據(jù)庫。幾乎所有關(guān)系數(shù)據(jù)庫都提供JDBC驅(qū)動，包括Oracle、Microsoft SQL Server、DB2、MySQL和Postgres。


image.png

下面將從最簡單的Kafka連接器配置開始，然后進(jìn)行構(gòu)建。本文中的示例是從MySQL數(shù)據(jù)庫中提取數(shù)據(jù)，該數(shù)據(jù)庫有兩個模式，每個模式都有幾張表：

mysql> SELECT table_schema, table_name FROM INFORMATION_SCHEMA.tables WHERE TABLE_SCHEMA != 'information_schema';
+--------------+--------------+
| TABLE_SCHEMA | TABLE_NAME   |
+--------------+--------------+
| demo         | accounts     |
| demo         | customers    |
| demo         | transactions |
| security     | firewall     |
| security     | log_events   |
+--------------+--------------+

JDBC驅(qū)動

在進(jìn)行配置之前，要確保Kafka連接器可以實(shí)際連接到數(shù)據(jù)庫，即確保JDBC驅(qū)動可用。如果使用的是SQLite或Postgres，那么驅(qū)動已經(jīng)包含在內(nèi)，就可以跳過此步驟。對于所有其它數(shù)據(jù)庫，需要將相關(guān)的JDBC驅(qū)動JAR文件放在和kafka-connect-jdbcJAR相同的文件夾中。此文件夾的標(biāo)準(zhǔn)位置為：

  1. Confluent CLI：下載的Confluent Platform文件夾中的share/java/kafka-connect-jdbc/；
  2. Docker，DEB / RPM安裝：/usr/share/java/kafka-connect-jdbc/，關(guān)于如何將JDBC驅(qū)動添加到Kafka連接器的Docker容器，請參閱此處；
  3. 如果kafka-connect-jdbcJAR位于其它位置，則可以使用plugin.path指向包含它的文件夾，并確保JDBC驅(qū)動位于同一文件夾中。

還可以在啟動Kafka連接器時指定CLASSPATH，設(shè)置為可以找到JDBC驅(qū)動的位置。一定要將其設(shè)置為JAR本身，而不僅僅是包含它的文件夾，例如：

CLASSPATH=/u01/jdbc-drivers/mysql-connector-java-8.0.13.JAR ./bin/connect-distributed ./etc/kafka/connect-distributed.properties

兩個事情要注意一下：

如果kafka-connect-jdbcJAR位于其它位置，則Kafka連接器的plugin.path選項(xiàng)將無法直接指向JDBC驅(qū)動JAR文件 。根據(jù)文檔，每個JDBC驅(qū)動JAR必須與kafka-connect-jdbcJAR位于同一目錄；
如果正在運(yùn)行多節(jié)點(diǎn)Kafka連接器集群，則需要在集群中的每個連接器工作節(jié)點(diǎn)上都正確安裝JDBC驅(qū)動JAR。
確認(rèn)是否已加載JDBC驅(qū)動

Kafka連接器會加載與kafka-connect-jdbcJAR文件在同一文件夾中的所有JDBC驅(qū)動，還有在CLASSPATH上找到的任何JDBC驅(qū)動。如果要驗(yàn)證一下，可以將連接器工作節(jié)點(diǎn)的日志級別調(diào)整為DEBUG，然后會看到如下信息：

1.DEBUG Loading plugin urls：包含kafka-connect-jdbc-5.1.0.jar（或者對應(yīng)當(dāng)前正在運(yùn)行的版本號）的一組JAR文件：

DEBUG Loading plugin urls: [file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/audience-annotations-0.5.0.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/common-utils-5.1.0.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/jline-0.9.94.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/jtds-1.3.1.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/kafka-connect-jdbc-5.1.0.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/mysql-connector-java-8.0.13.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/netty-3.10.6.Final.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/postgresql-9.4-1206-jdbc41.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/slf4j-api-1.7.25.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/sqlite-jdbc-3.25.2.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/zkclient-0.10.jar, file:/Users/Robin/cp/confluent-5.1.0/share/java/kafka-connect-jdbc/zookeeper-3.4.13.jar] (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)

在這個JAR列表中，應(yīng)該有JDBC驅(qū)動JAR。在上面的輸出中，可以看到MySQL、Postgres和SQLite的JAR。如果期望的JDBC驅(qū)動JAR不在，可以將驅(qū)動放入kafka-connect-jdbcJAR所在的文件夾中。

2.INFO Added plugin 'io.confluent.connect.jdbc.JdbcSourceConnector'：在此之后，在記錄任何其它插件之前，可以看到JDBC驅(qū)動已注冊：

INFO Added plugin 'io.confluent.connect.jdbc.JdbcSourceConnector' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
DEBUG Registered java.sql.Driver: jTDS 1.3.1 to java.sql.DriverManager (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
DEBUG Registered java.sql.Driver: com.mysql.cj.jdbc.Driver@7bbbb6a8 to java.sql.DriverManager (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
DEBUG Registered java.sql.Driver: org.postgresql.Driver@ea9e141 to java.sql.DriverManager (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)
DEBUG Registered java.sql.Driver: org.sqlite.JDBC@236134a1 to java.sql.DriverManager (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader)

確認(rèn)JDBC驅(qū)動包含在已注冊的列表中。如果沒有，那么就是安裝不正確。

注意，雖然可能會在日志的其它地方看到驅(qū)動的Registered java.sql.Driver信息，但如果要確認(rèn)其對于JDBC連接器可用，那么它必須直接出現(xiàn)在INFO Added plugin 'io.confluent.connect.jdbc消息的后面。

注意，雖然JDBC URL通常允許嵌入身份驗(yàn)證信息，但這些內(nèi)容將以明文形式記錄在Kafka連接器日志中。因此應(yīng)該使用單獨(dú)的connection.user和connection.password配置項(xiàng)，這樣在記錄時會被合理地處理。

指定要提取的表
JDBC驅(qū)動安裝完成之后，就可以配置Kafka連接器從數(shù)據(jù)庫中提取數(shù)據(jù)了。下面是最小的配置，不過它不一定是最有用的，因?yàn)樗菙?shù)據(jù)的批量導(dǎo)入，在本文后面會討論如何進(jìn)行增量加載。

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/demo",
"connection.user": "connect_user",
"connection.password": "asgard",
"topic.prefix": "mysql-01-",
"mode":"bulk"
}
}'

使用此配置，每個表（用戶有權(quán)訪問）將完全復(fù)制到Kafka，通過使用KSQL列出Kafka集群上的主題，我們可以看到：
ksql> LIST TOPICS;

Kafka Topic | Registered | Partitions | Partition Replicas | Consumers | ConsumerGroups

mysql-01-accounts | false | 1 | 1 | 0 | 0
mysql-01-customers | false | 1 | 1 | 0 | 0
mysql-01-firewall | false | 1 | 1 | 0 | 0
mysql-01-log_events | false | 1 | 1 | 0 | 0
mysql-01-transactions | false | 1 | 1 | 0 | 0

注意mysql-01前綴，表格內(nèi)容的完整副本將每五秒刷新一次，可以通過修改poll.interval.ms進(jìn)行調(diào)整，例如每小時刷新一次：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
"name": "jdbc_source_mysql_02",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:mysql://mysql:3306/demo",
"connection.user": "connect_user",
"connection.password": "asgard",
"topic.prefix": "mysql-02-",
"mode":"bulk",
"poll.interval.ms" : 3600000
}
}'

找個主題確認(rèn)一下，顯示完整的數(shù)據(jù)，看看是不是自己想要的：

ksql> PRINT 'mysql-02-accounts' FROM BEGINNING;
Format:AVRO
12/20/18 3:18:44 PM UTC, null, {"id": 1, "first_name": "Hamel", "last_name": "Bly", "username": "Hamel Bly", "company": "Erdman-Halvorson", "created_date": 17759}
12/20/18 3:18:44 PM UTC, null, {"id": 2, "first_name": "Scottie", "last_name": "Geerdts", "username": "Scottie Geerdts", "company": "Mante Group", "created_date": 17692}
12/20/18 3:18:44 PM UTC, null, {"id": 3, "first_name": "Giana", "last_name": "Bryce", "username": "Giana Bryce", "company": "Wiza Inc", "created_date": 17627}
12/20/18 3:18:44 PM UTC, null, {"id": 4, "first_name": "Allen", "last_name": "Rengger", "username": "Allen Rengger", "company": "Terry, Jacobson and Daugherty", "created_date": 17746}
12/20/18 3:18:44 PM UTC, null, {"id": 5, "first_name": "Reagen", "last_name": "Volkes", "username": "Reagen Volkes", "company": "Feeney and Sons", "created_date": 17798}
…

目前會展示所有可用的表，這可能不是實(shí)際的需求，可能只希望包含特定模式的表，這個可以使用catalog.pattern/schema.pattern（具體哪一個取決于數(shù)據(jù)庫）配置項(xiàng)進(jìn)行控制：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
"name": "jdbc_source_mysql_03",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:mysql://mysql:3306/demo",
"connection.user": "connect_user",
"connection.password": "asgard",
"topic.prefix": "mysql-03-",
"mode":"bulk",
"poll.interval.ms" : 3600000,
"catalog.pattern" : "demo"
}
}'

這樣就只會從demo模式中取得3張表：

ksql> LIST TOPICS;

Kafka Topic | Registered | Partitions | Partition Replicas | Consumers | ConsumerGroups

[…]
mysql-03-accounts | false | 1 | 1 | 0 | 0
mysql-03-customers | false | 1 | 1 | 0 | 0
mysql-03-transactions | false | 1 | 1 | 0 | 0
[…]

也可以使用table.whitelist（白名單）或table.blacklist（黑名單）來控制連接器提取的表，下面的示例顯式地列出了希望拉取到Kafka中的表清單：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
"name": "jdbc_source_mysql_04",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:mysql://mysql:3306/demo",
"connection.user": "connect_user",
"connection.password": "asgard",
"topic.prefix": "mysql-04-",
"mode":"bulk",
"poll.interval.ms" : 3600000,
"catalog.pattern" : "demo",
"table.whitelist" : "accounts"
}
}'

這時就只有一個表從數(shù)據(jù)庫流式傳輸?shù)終afka：

ksql> LIST TOPICS;

Kafka Topic | Registered | Partitions | Partition Replicas | Consumers | ConsumerGroups

mysql-04-accounts | false | 1 | 1 | 0 | 0

因?yàn)橹挥幸粋€表，下面的配置：

"catalog.pattern" : "demo",
"table.whitelist" : "accounts",

等同于：

"table.whitelist" : "demo.accounts",

也可以在一個模式中指定多個表，比如：

"catalog.pattern" : "demo",
"table.whitelist" : "accounts, customers",

或者也可以跨越多個模式：

"table.whitelist" : "demo.accounts, security.firewall",

還可以使用其它的表過濾選項(xiàng)，比如table.types可以選擇表之外的對象，例如視圖。

過濾表時要注意，因?yàn)槿绻罱K沒有對象匹配該模式（或者連接到數(shù)據(jù)庫的已認(rèn)證用戶沒有權(quán)限訪問），那么連接器將報錯：

INFO After filtering the tables are: (io.confluent.connect.jdbc.source.TableMonitorThread)
…
ERROR Failed to reconfigure connector's tasks, retrying after backoff: (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
java.lang.IllegalArgumentException: Number of groups must be positive

在通過table.whitelist/table.blacklist進(jìn)行過濾之前，可以將日志級別調(diào)整為DEBUG，查看用戶可以訪問的表清單：

DEBUG Got the following tables: ["demo"."accounts", "demo"."customers"] (io.confluent.connect.jdbc.source.TableMonitorThread)

然后，連接器會根據(jù)提供的白名單/黑名單過濾此列表，因此要確認(rèn)指定的列表位于連接器可用的列表中，還要注意連接用戶要有權(quán)限訪問這些表，因此還要檢查數(shù)據(jù)庫端的GRANT語句。

增量提取

到目前為止，已經(jīng)按計(jì)劃將整張表都拉取到Kafka，這雖然對于轉(zhuǎn)存數(shù)據(jù)非常有用，不過都是批量并且并不總是適合將源數(shù)據(jù)庫集成到Kafka流系統(tǒng)中。

JDBC連接器還有一個流式傳輸?shù)終afka的選項(xiàng)，它只會傳輸上次拉取后的數(shù)據(jù)變更，具體可以基于自增列（例如自增主鍵）和/或時間戳（例如最后更新時間戳）來執(zhí)行此操作。在模式設(shè)計(jì)中的常見做法是使用這些中的一個或兩個，例如，事務(wù)表ORDERS可能有：

  • ORDER_ID：一個唯一鍵（可能是主鍵），每個新訂單遞增；
  • UPDATE_TS：每次數(shù)據(jù)變更時更新的時間戳列。

可以使用mode參數(shù)配置該選項(xiàng)，比如使用timestamp：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_mysql_08",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:mysql://mysql:3306/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-08-",
                "mode":"timestamp",
                "table.whitelist" : "demo.accounts",
                "timestamp.column.name": "UPDATE_TS",
                "validate.non.null": false
                }
        }'

下面會獲取表的全部數(shù)據(jù)，外加源數(shù)據(jù)后續(xù)的更新和插入：

image

注意：

  • 可以結(jié)合使用這些方法中的（時間戳/自增）或兩者（時間戳+自增）；
  • 要使用的時間戳和/或自增列必須在連接器處理的所有表上。如果不同的表具有不同名稱的時間戳/自增列，則需要創(chuàng)建單獨(dú)的連接器配置；
  • 如果只使用自增列，則不會捕獲對數(shù)據(jù)的更新，除非每次更新時自增列也會增加（在主鍵的情況下幾乎不可能）；
  • 某些表可能沒有唯一的標(biāo)識，或者有多個組合的列表示行的唯一標(biāo)識（聯(lián)合主鍵），不過JDBC連接器只支持單個標(biāo)識列；
  • 時間戳+自增列選項(xiàng)為識別新行和更新行提供了最大的覆蓋范圍；
  • 許多RDBMS支持聲明更新時間戳列的DDL，該列會自動更新。例如：
    • MySQL：
CREATE TABLE foo (
        …
        UPDATE_TS TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);
  • Postgres：
CREATE TABLE foo (
        …
        UPDATE_TS TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Courtesy of https://techblog.covermymeds.com/databases/on-update-timestamps-mysql-vs-postgres/
CREATE FUNCTION update_updated_at_column() RETURNS trigger
    LANGUAGE plpgsql
    AS $$
  BEGIN
    NEW.update_ts = NOW();
    RETURN NEW;
  END;
$$;

CREATE TRIGGER t1_updated_at_modtime BEFORE UPDATE ON foo FOR EACH ROW EXECUTE PROCEDURE update_updated_at_column();
  • Oracle：
CREATE TABLE foo (
        …
        CREATE_TS TIMESTAMP DEFAULT CURRENT_TIMESTAMP ,
);

CREATE OR REPLACE TRIGGER TRG_foo_UPD
BEFORE INSERT OR UPDATE ON foo
REFERENCING NEW AS NEW_ROW
  FOR EACH ROW
BEGIN
  SELECT SYSDATE
        INTO :NEW_ROW.UPDATE_TS
        FROM DUAL;
END;
/

#基于查詢的提取

有時可能想從RDBMS中提取數(shù)據(jù)，但希望有比整個表更靈活的方式，原因可能包括：

  • 一個有許多列的寬表，但是只希望有部分列被傳輸?shù)終afka主題上；
  • 表中包含敏感信息，不希望這些信息傳輸?shù)終afka主題上（盡管也可以提取時在Kafka連接器中使用單消息轉(zhuǎn)換進(jìn)行處理）；
  • 多個表之間存在依賴關(guān)系，因此在傳輸?shù)終afka之前，可能希望將其解析為一個單一的一致性視圖。

這可以使用JDBC連接器的query模式。在了解如何實(shí)現(xiàn)之前，需要注意以下幾點(diǎn)：

  • 謹(jǐn)防管道的“過早優(yōu)化”，僅僅因?yàn)椴恍枰幢碇械哪承┝谢蛐?，而不是說在流式傳輸?shù)終afka時不應(yīng)包含它們；
  • 正如將在下面看到的，當(dāng)涉及增量攝取時，query模式可能不那么靈活，因此從源中簡單地刪除列的另一種方法（無論是簡單地減少數(shù)量，還是因?yàn)槊舾行畔ⅲ┒际窃谶B接器本身中使用ReplaceField單消息轉(zhuǎn)換；
  • 隨著查詢越來越復(fù)雜（例如解析關(guān)聯(lián)），潛在的壓力和對源數(shù)據(jù)庫的影響會增加；
  • 在RDBMS（作為源頭）中關(guān)聯(lián)數(shù)據(jù)是解決關(guān)聯(lián)的一種方法，另一種方法是將源表流式傳輸?shù)絾为?dú)的Kafka主題，然后使用KSQL或Kafka Streams根據(jù)需求進(jìn)行關(guān)聯(lián)（過濾和標(biāo)記數(shù)據(jù)也是如此），KSQL是在Kafka中對數(shù)據(jù)進(jìn)行后處理的絕佳方式，使管道盡可能簡單。

下面將展示如何將transactions表，再加上customers表中的數(shù)據(jù)流式傳輸?shù)終afka：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_mysql_09",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:mysql://mysql:3306/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-09",
                "mode":"bulk",
                "query":"SELECT t.txn_id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id;",
                "poll.interval.ms" : 3600000
                }
        }'

可能注意到已切換回bulk模式，可以使用主鍵或者時間戳其中一個增量選項(xiàng)，但要確保在SELECT子句中包含相應(yīng)的主鍵/時間戳列（例如txn_id）：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_mysql_10",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:mysql://mysql:3306/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-10",
                "mode":"incrementing",
                "query":"SELECT txn_id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id",
                "incrementing.column.name": "txn_id",
                "validate.non.null": false
                }
        }'

如果不包括該列（即使它存在于源表中），那么連接器會報錯并顯示org.apache.kafka.connect.errors.DataException異常（＃561）或java.lang.NullPointerException異常（＃560），這是因?yàn)檫B接器需要在返回的數(shù)據(jù)中獲取值，以便可以存儲相應(yīng)偏移量的最新值。

如果使用query選項(xiàng)，除非使用mode: bulk（＃566），否則無法指定自己的WHERE子句，也就是說，在查詢中使用自己的謂詞和使用Kafka進(jìn)行增量提取之間是互斥的。

一個還是多個連接器？

如果需要不同的參數(shù)設(shè)定，可以創(chuàng)建新的連接器，例如，可能希望有不同的參數(shù)：

  • 包含自增主鍵和/或時間戳的列的名稱；
  • 輪詢表的頻率；
  • 連接數(shù)據(jù)庫的用戶不同。

簡單來說，如果所有表參數(shù)都一樣，則可以使用單個連接器。

#為什么沒有數(shù)據(jù)？

創(chuàng)建連接器之后，可能在目標(biāo)Kafka主題中看不到任何數(shù)據(jù)。下面會一步步進(jìn)行診斷：

1.查詢/connectors端點(diǎn)，可確認(rèn)連接器是否創(chuàng)建成功：

$ curl -s“http：// localhost：8083 / connectors”
[ “jdbc_source_mysql_10”]

1
2

應(yīng)該看到連接器列表，如果沒有，則需要按照之前的步驟進(jìn)行創(chuàng)建，然后關(guān)注Kafka連接器返回的任何錯誤。

2.檢查連接器及其任務(wù)的狀態(tài)：

$ curl -s "http://localhost:8083/connectors/jdbc_source_mysql_10/status"|jq '.'
{
"name": "jdbc_source_mysql_10",
"connector": {
"state": "RUNNING",
"worker_id": "kafka-connect:8083"
},
"tasks": [
{
"state": "RUNNING",
"id": 0,
"worker_id": "kafka-connect:8083"
}
],
"type": "source"
}

正常應(yīng)該看到所有的連接器和任務(wù)的state都是RUNNING，不過RUNNING不總是意味著正常。

3.如果連接器或任務(wù)的狀態(tài)是FAILED，或者即使?fàn)顟B(tài)是RUNNING但是沒有按照預(yù)期行為運(yùn)行，那么可以轉(zhuǎn)到Kafka連接器工作節(jié)點(diǎn)的輸出（這里有相關(guān)的說明），這里會顯示是否存在任何實(shí)際的問題。以上面的連接器為例，其狀態(tài)為RUNNING，但是連接器工作節(jié)點(diǎn)日志中實(shí)際上全是重復(fù)的錯誤：

ERROR Failed to run query for table TimestampIncrementingTableQuerier{table=null, query='SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id;', topicPrefix='mysql-10', incrementingColumn='t.id', timestampColumns=[]}: {} (io.confluent.connect.jdbc.source.JdbcSourceTask)
java.sql.SQLSyntaxErrorException: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'WHERE `t.id` > -1 ORDER BY `t.id` ASC' at line 1
 at com.mysql.cj.jdbc.exceptions.SQLError.createSQLException(SQLError.java:120)

4.在這里，問題是什么并不明確，需要調(diào)出連接器的配置來檢查指定的查詢是否正確：

$ curl -s "http://localhost:8083/connectors/jdbc_source_mysql_10/config"|jq '.'
{
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "mode": "incrementing",
  "incrementing.column.name": "t.id",
  "topic.prefix": "mysql-10",
  "connection.password": "asgard",
  "validate.non.null": "false",
  "connection.user": "connect_user",
  "query": "SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id;",
  "name": "jdbc_source_mysql_10",
  "connection.url": "jdbc:mysql://mysql:3306/demo"
}

5.在MySQL中運(yùn)行此查詢發(fā)現(xiàn)能正常執(zhí)行：

mysql> SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id;
+------+-------------+--------+----------+----------------------+------------+-----------+----------------------------+--------+------------------------------------------------------+
| id   | customer_id | amount | currency | txn_timestamp        | first_name | last_name | email                      | gender | comments                                             |
+------+-------------+--------+----------+----------------------+------------+-----------+----------------------------+--------+------------------------------------------------------+
|    1 |           5 | -72.97 | RUB      | 2018-12-12T13:58:37Z | Modestia   | Coltart   | mcoltart4@scribd.com       | Female | Reverse-engineered non-volatile success

6.所以肯定是Kafka連接器在執(zhí)行時做了什么。鑒于錯誤消息引用t.id，這是在incrementing.column.name參數(shù)中指定的，可能問題與此有關(guān)。通過將Kafka連接器的日志級別調(diào)整為DEBUG，可以看到執(zhí)行的完整SQL語句：

DEBUG TimestampIncrementingTableQuerier{table=null, query='SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id;', topicPrefix='mysql-10', incrementingColumn='t.id', timestampColumns=[]} prepared SQL query: SELECT t.id, t.customer_id, t.amount, t.currency, t.txn_timestamp, c.first_name, c.last_name, c.email, c.gender, c.comments FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id; WHERE `t.id` > ? ORDER BY `t.id` ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier)

7.看一下該prepared SQL query部分，可能會發(fā)現(xiàn)：

[…] FROM demo.transactions t LEFT OUTER JOIN demo.customers c on t.customer_id = c.id; WHERE `t.id` > ? ORDER BY `t.id` ASC

8.注意在JOIN子句的c.id后面有語句終止符（;），后面有WHERE子句。該WHERE子句由Kafka連接器附加，用于實(shí)現(xiàn)所要求的incrementing模式，但創(chuàng)建了一個無效的SQL語句； 9.然后在GitHub中查找與看到的錯誤相關(guān)的問題，因?yàn)橛袝r它實(shí)際上是一個已知的問題，例如這個問題； 10.如果連接器存在并且是RUNNING，并且Kafka連接器工作節(jié)點(diǎn)日志中也沒有錯誤，還應(yīng)該檢查：

- 連接器的提取間隔是多少？也許它完全按照配置運(yùn)行，并且源表中的數(shù)據(jù)已經(jīng)更改，但就是沒有拉取到新數(shù)據(jù)。要檢查這一點(diǎn)，可以在Kafka連接器工作節(jié)點(diǎn)的輸出中查找`JdbcSourceTaskConfig`的值和`poll.interval.ms`的值；
- 如果正在使用的是增量攝取，Kafka連接器關(guān)于偏移量是如何存儲的？如果刪除并重建相同名稱的連接器，則將保留前一個實(shí)例的偏移量?？紤]這樣的場景，創(chuàng)建完連接器之后，成功地將所有數(shù)據(jù)提取到源表中的給定主鍵或時間戳值，然后刪除并重新創(chuàng)建了它，新版本的連接器將獲得之前版本的偏移量，因此僅提取比先前處理的數(shù)據(jù)更新的數(shù)據(jù)，具體可以通過查看保存在其中的`offset.storage.topic`值和相關(guān)表來驗(yàn)證這一點(diǎn)。

#重置JDBC源連接器讀取數(shù)據(jù)的點(diǎn)

當(dāng)Kafka連接器以分布式模式運(yùn)行時，它會在Kafka主題（通過offset.storage.topic配置）中存儲有關(guān)它在源系統(tǒng)中讀取的位置（稱為偏移量）的信息，當(dāng)連接器任務(wù)重啟時，它可以從之前的位置繼續(xù)進(jìn)行處理，具體可以在連接器工作節(jié)點(diǎn)日志中看到：

INFO Found offset {{protocol=1, table=demo.accounts}={timestamp_nanos=0, timestamp=1547030056000}, {table=accounts}=null} for partition {protocol=1, table=demo.accounts} (io.confluent.connect.jdbc.source.JdbcSourceTask)

每次連接器輪詢時，都會使用這個偏移量，它會使用預(yù)編譯的SQL語句，并且使用Kafka連接器任務(wù)傳遞的值替換?占位符：

DEBUG TimestampIncrementingTableQuerier{table="demo"."accounts", query='null', topicPrefix='mysql-08-', incrementingColumn='', timestampColumns=[UPDATE_TS]} prepared SQL query: SELECT * FROM `demo`.`accounts` WHERE `demo`.`accounts`.`UPDATE_TS` > ? AND `demo`.`accounts`.`UPDATE_TS` < ? ORDER BY `demo`.`accounts`.`UPDATE_TS` ASC (io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier)
DEBUG Executing prepared statement with timestamp value = 2019-01-09 10:34:16.000 end time = 2019-01-09 13:23:40.000 (io.confluent.connect.jdbc.source.TimestampIncrementingCriteria)

這里，第一個時間戳值就是存儲的偏移量，第二個時間戳值是當(dāng)前時間戳。

雖然沒有文檔記載，但可以手動更改連接器使用的偏移量，因?yàn)槭窃贘DBC源連接器的上下文中，所以可以跨多個源連接器類型，這意味著更改時間戳或主鍵，連接器會將后續(xù)記錄視為未處理的狀態(tài)。

首先要做的是確保Kafka連接器已經(jīng)刷新了周期性的偏移量，可以在工作節(jié)點(diǎn)日志中看到何時執(zhí)行此操作：

INFO WorkerSourceTask{id=jdbc_source_mysql_08-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask)

看下Kafka的主題，可以看到Kafka連接器創(chuàng)建的內(nèi)部主題，并且負(fù)責(zé)偏移量的主題也是其中之一，名字可能有所不同：

ksql> LIST TOPICS;

 Kafka Topic            | Registered | Partitions | Partition Replicas | Consumers | ConsumerGroups
----------------------------------------------------------------------------------------------------
 docker-connect-configs | false      | 1          | 1                  | 0         | 0
 docker-connect-offsets | false      | 1          | 1                  | 0         | 0
 docker-connect-status  | false      | 5          | 1                  | 0         | 0

ksql> PRINT 'docker-connect-offsets' FROM BEGINNING;
Format:JSON
{"ROWTIME":1547038346644,"ROWKEY":"[\"jdbc_source_mysql_08\",{\"protocol\":\"1\",\"table\":\"demo.customers\"}]","timestamp_nanos":0,"timestamp":1547030057000}

當(dāng)Kafka連接器任務(wù)啟動時，它會讀取此主題并使用適當(dāng)主鍵的最新值。要更改偏移量，只需插入一個新值即可。最簡單的方法是轉(zhuǎn)存當(dāng)前主題內(nèi)容，修改內(nèi)容并重新執(zhí)行，因?yàn)橐恢滦院秃唵?，可以考慮使用kafkacat：

  • 轉(zhuǎn)存當(dāng)前的內(nèi)容：
$ kafkacat -b kafka:29092 -t docker-connect-offsets -C -K# -o-1
% Reached end of topic docker-connect-offsets [0] at offset 0
["jdbc_source_mysql_08",{"protocol":"1","table":"demo.accounts"}]#{"timestamp_nanos":0,"timestamp":1547030056000}
);

如果是多個連接器，可能復(fù)雜些，但是這里只有一個，所以使用了-o-1標(biāo)志，它定義了返回的偏移量。

  • 根據(jù)需要修改偏移量。在這里使用了mode=timestamp來監(jiān)測表中的變化。時間戳值是1547030056000，使用相關(guān)的時間戳轉(zhuǎn)換之類的工具，可以很容易地轉(zhuǎn)換和操作，比如將其提前一小時（1547026456000）。接下來，使用更新后的timestamp值準(zhǔn)備新消息：
["jdbc_source_mysql_08",{"protocol":"1","table":"demo.accounts"}]#{"timestamp_nanos":0,"timestamp":1547026456000}
  • 將新消息發(fā)給主題：
echo '["jdbc_source_mysql_08",{"protocol":"1","table":"demo.accounts"}]#{"timestamp_nanos":0,"timestamp":1547026456000}' | \
kafkacat -b kafka:29092 -t docker-connect-offsets -P -Z -K#
  • 如果要從頭開始重啟連接器，可以發(fā)送NULL消息值：
echo'[“jdbc_source_mysql_08”，{“protocol”：“1”，“table”：“demo.accounts”}]＃'| \
kafkacat -b kafka：29092 -t docker-connect-offsets -P -Z -K＃
  • 重啟連接器任務(wù)：
curl -i -X POST -H "Accept:application/json" \
        -H "Content-Type:application/json" http://localhost:8083/connectors/jdbc_source_mysql_08/tasks/0/restart
  • 也可以只重啟Kafka連接器工作節(jié)點(diǎn)，重啟之后，數(shù)據(jù)源中所有比新設(shè)置的偏移量更新的記錄，都會被重新提取到Kafka主題中。

從指定的時間戳或者主鍵處開啟表的捕獲

當(dāng)使用時間戳或自增主鍵模式創(chuàng)建JDBC源連接器時，它會從主鍵為-1和/或時間戳為1970-01-01 00:00:00.00開始，這意味著會獲得表的全部內(nèi)容，然后在后續(xù)的輪詢中獲取任何插入/更新的數(shù)據(jù)。

但是如果不想要表的完整副本，只是希望連接器從現(xiàn)在開始，該怎么辦呢？這在目前的Kafka連接器中還不支持，但可以使用前述的方法。不需要獲取現(xiàn)有的偏移量消息并對其進(jìn)行定制，而是自己創(chuàng)建。消息的格式依賴于正在使用的連接器和表的名稱，一種做法是先創(chuàng)建連接器，確定格式，然后刪除連接器，另一種做法是使用具有相同源表名和結(jié)構(gòu)的環(huán)境，除非在該環(huán)境中沒有可供連接器提取的數(shù)據(jù)，否則同樣也能得到所需的消息格式。

在創(chuàng)建連接器之前，使用適當(dāng)?shù)闹蹬渲闷屏恐黝}。在這里，希望從demo.transactions表中提取自增主鍵大于42的所有行：

echo '["jdbc_source_mysql_20",{"protocol":"1","table":"demo.transactions"}]#{"incrementing":42}' | \
kafkacat -b kafka:29092 -t docker-connect-offsets -P -Z -K#

下面創(chuàng)建連接器：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_mysql_20",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:mysql://mysql:3306/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-20-",
                "mode":"incrementing",
                "table.whitelist" : "demo.transactions",
                "incrementing.column.name": "txn_id",
                "validate.non.null": false
                }
        }'

在生成的Kafka連接器工作日志中，可以看到：

INFO Found offset {{protocol=1, table=demo.transactions}={incrementing=42}, {table=transactions}=null} for partition {protocol=1, table=demo.transactions} (io.confluent.connect.jdbc.source.JdbcSourceTask)
…
DEBUG Executing prepared statement with incrementing value = 42 (io.confluent.connect.jdbc.source.TimestampIncrementingCriteria)

和預(yù)期一樣，Kafka主題中只注入了txn_id大于42的行：

ksql> PRINT 'mysql-20x-transactions' FROM BEGINNING;
Format:AVRO
1/9/19 1:44:07 PM UTC, null, {"txn_id": 43, "customer_id": 3, "amount": {"bytes": "ús"}, "currency": "CNY", "txn_timestamp": "2018-12-15T08:23:24Z"}
1/9/19 1:44:07 PM UTC, null, {"txn_id": 44, "customer_id": 5, "amount": {"bytes": "\f!"}, "currency": "CZK", "txn_timestamp": "2018-10-04T13:10:17Z"}
1/9/19 1:44:07 PM UTC, null, {"txn_id": 45, "customer_id": 3, "amount": {"bytes": "?ò"}, "currency": "USD", "txn_timestamp": "2018-04-03T03:40:49Z"}

#配置Kafka消息鍵

Kafka消息是鍵/值對，其中值是有效內(nèi)容。在JDBC連接器的上下文中，值是要被提取的表行的內(nèi)容。Kafka消息中的鍵對于分區(qū)和下游處理非常重要，其中任何關(guān)聯(lián)（比如KSQL）都將在數(shù)據(jù)中完成。

JDBC連接器默認(rèn)不設(shè)置消息鍵，但是使用Kafka連接器的單消息轉(zhuǎn)換（SMT）機(jī)制可以輕松實(shí)現(xiàn)。假設(shè)想要提取accounts表并將其ID列用作消息鍵。只需簡單地將其添加到下面的配置中即可：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_mysql_06",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:mysql://mysql:3306/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-06-",
                "poll.interval.ms" : 3600000,
                "table.whitelist" : "demo.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"
                }
        }'

這時如果使用諸如kafka-avro-console-consumer之類的工具檢查數(shù)據(jù)，就會看到鍵（JSON內(nèi)容之前的最左列）與id值匹配：

kafka-avro-console-consumer \
      --bootstrap-server kafka:29092 \
      --property schema.registry.url=http://schema-registry:8081 \
      --topic mysql-06-accounts --from-beginning --property print.key=true

1       {"id":{"int":1},"first_name":{"string":"Hamel"},"last_name":{"string":"Bly"},"username":{"string":"Hamel Bly"},"company":{"string":"Erdman-Halvorson"},"created_date":{"int":17759}}
2       {"id":{"int":2},"first_name":{"string":"Scottie"},"last_name":{"string":"Geerdts"},"username":{"string":"Scottie Geerdts"},"company":{"string":"Mante Group"},"created_date":{"int":17692}}

如果要在數(shù)據(jù)中設(shè)置鍵以便與KSQL一起使用，則需要將其創(chuàng)建為字符串類型，因?yàn)镵SQL目前不支持其它鍵類型，具體可以在連接器配置中添加如下內(nèi)容：

"key.converter": "org.apache.kafka.connect.storage.StringConverter"

然后就可以在KSQL中使用了：

ksql> CREATE STREAM ACCOUNTS WITH (KAFKA_TOPIC='mysql-06X-accounts', VALUE_FORMAT='AVRO');
ksql> SELECT ROWKEY, ID, FIRST_NAME + ' ' + LAST_NAME FROM ACCOUNTS;
1 | 1 | Hamel Bly
2 | 2 | Scottie Geerdts
3 | 3 | Giana Bryce

#更改主題名稱

JDBC連接器要求指定topic.prefix，但如果不想要，或者想將主題名更改為其它模式，SMT可以實(shí)現(xiàn)。

假設(shè)要刪除mysql-07-前綴，那么需要一點(diǎn)正則表達(dá)式的技巧：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_mysql_07",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:mysql://mysql:3306/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-07-",
                "poll.interval.ms" : 3600000,
                "catalog.pattern" : "demo",
                "table.whitelist" : "accounts",
                "mode":"bulk",
                "transforms":"dropTopicPrefix",
                "transforms.dropTopicPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
                "transforms.dropTopicPrefix.regex":"mysql-07-(.*)",
                "transforms.dropTopicPrefix.replacement":"$1"
                }
        }'

這樣主題名就和表名一致了：

ksql> LIST TOPICS;

 Kafka Topic            | Registered | Partitions | Partition Replicas | Consumers | ConsumerGroups
----------------------------------------------------------------------------------------------------
 accounts               | false      | 1          | 1                  | 0         | 0

#Bytes, Decimals, Numerics和自定義類型

這是話題比較深入。

  • numeric.mapping: best_fit如果源中包含NUMERIC/NUMBER類型的數(shù)據(jù)，則可能需要這個配置項(xiàng)；
  • 如果需要，可以在JDBC連接器中使用query選項(xiàng)，用于對源表中的數(shù)據(jù)進(jìn)行轉(zhuǎn)換；
  • 如果字段以JDBCDECIMAL類型暴露，則numeric.mapping無法處理：
    • MySQL將所有數(shù)值存儲為DECIMAL；
    • SQL Server將DECIMAL和NUMERIC原生存儲，因此必須將DECIMAL字段轉(zhuǎn)換為NUMERIC；
  • 在Oracle中，要在NUMBER字段中指定長度和標(biāo)度，例如NUMBER(5,0)，不能是NUMBER；
  • NUMERIC和DECIMAL都被視為NUMBER，INT也是；

完成之后，下面會做一個解釋：

Kafka連接器是一個可以將數(shù)據(jù)注入Kafka、與特定源技術(shù)無關(guān)的框架。無論是來自SQL Server、DB2、MQTT、文本文件、REST還是Kafka連接器支持的任何其它數(shù)十種來源，它發(fā)送給Kafka的數(shù)據(jù)格式都為Avro或JSON，這通常是一個透明的過程，只是在處理數(shù)值數(shù)據(jù)類型時有些特別，比如DECIMAL，NUMBER等等，以下面的MySQL查詢?yōu)槔?/p> 

mysql> SELECT * FROM transactions LIMIT 1;
+--------+-------------+--------+----------+----------------------+
| txn_id | customer_id | amount | currency | txn_timestamp        |
+--------+-------------+--------+----------+----------------------+
|      1 |           5 | -72.97 | RUB      | 2018-12-12T13:58:37Z |
+--------+-------------+--------+----------+----------------------+

挺正常是吧？其實(shí)，amount列是DECIMAL(5,2)：

mysql> describe transactions;
+---------------+--------------+------+-----+---------+-------+
| Field         | Type         | Null | Key | Default | Extra |
+---------------+--------------+------+-----+---------+-------+
| txn_id        | int(11)      | YES  |     | NULL    |       |
| customer_id   | int(11)      | YES  |     | NULL    |       |
| amount        | decimal(5,2) | YES  |     | NULL    |       |
| currency      | varchar(50)  | YES  |     | NULL    |       |
| txn_timestamp | varchar(50)  | YES  |     | NULL    |       |
+---------------+--------------+------+-----+---------+-------+
5 rows in set (0.00 sec)

但是當(dāng)使用JDBC連接器的默認(rèn)設(shè)置提取到Kafka中時，最終會是這樣：

ksql> PRINT 'mysql-02-transactions' FROM BEGINNING;
Format:AVRO
1/4/19 5:38:45 PM UTC, null, {"txn_id": 1, "customer_id": 5, "amount": {"bytes": "?\u007F"}, "currency": "RUB", "txn_timestamp": "2018-12-12T13:58:37Z"}

DECIMAL變成了一個看似亂碼的bytes值，連接器默認(rèn)會使用自己的DECIMAL邏輯類型，該類型在Avro中被序列化為字節(jié)，可以通過查看Confluent模式注冊表中的相關(guān)條目來看到這一點(diǎn)：

$ curl -s "http://localhost:8081/subjects/mysql-02-transactions-value/versions/1"|jq '.schema|fromjson.fields[] | select (.name == "amount")'
{
  "name": "amount",
  "type": [
    "null",
    {
      "type": "bytes",
      "scale": 2,
      "precision": 64,
      "connect.version": 1,
      "connect.parameters": {
        "scale": "2"
      },
      "connect.name": "org.apache.kafka.connect.data.Decimal",
      "logicalType": "decimal"
    }
  ],
  "default": null
}

當(dāng)連接器使用AvroConverter消費(fèi)時，這會正常處理并保存為DECIMAL（并且在Java中也可以反序列化為BigDecimal），但對于反序列化Avro的其它消費(fèi)者，它們只會得到字節(jié)。在使用啟用了模式的JSON時，也會看到這一點(diǎn)，amount值會是Base64編碼的字節(jié)字符串：

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "bytes",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Decimal",
        "version": 1,
        "parameters": {
          "scale": "2"
        },
        "field": "amount"
      },
  },
  "payload": {
    "txn_id": 1000,
    "customer_id": 5,
    "amount": "Cv8="
  }
}

因此，不管使用的是JSON還是Avro，這都是numeric.mapping配置項(xiàng)的來源。它默認(rèn)設(shè)置為none（即使用連接器的DECIMAL類型），但通常希望連接器將類型實(shí)際轉(zhuǎn)換為更兼容的類型，以適合數(shù)字的精度，更具體的說明，可以參見相關(guān)的文檔。

此選項(xiàng)目前不支持DECIMAL類型，因此這里是在Postgres中具有NUMERIC類型的相同原理的示例：

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
              "name": "jdbc_source_postgres_12",
              "config": {
                      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                      "connection.url": "jdbc:postgresql://postgres:5432/postgres",
                      "connection.user": "connect_user",
                      "connection.password": "asgard",
                      "topic.prefix": "postgres-12-",
                      "numeric.mapping": "best_fit",
                      "table.whitelist" : "demo.transactions",
                      "mode":"bulk",
                      "poll.interval.ms" : 3600000
                      }
              }'

結(jié)果如下所示：

ksql> PRINT 'postgres-12-transactions' FROM BEGINNING;
Format:AVRO
1/7/19 6:27:16 PM UTC, null, {"txn_id": 1, "customer_id": 5, "amount": -72.97, "currency": "RUB", "txn_timestamp": "2018-12-12T13:58:37Z"}

可以在這里看到有關(guān)此內(nèi)容的更多詳細(xì)信息，以及Postgres、Oracle和MS SQL Server中的示例。

#處理多個表

如果需要從多個表中提取數(shù)據(jù)，則可以通過并行處理來減少總提取時間，這在Kafka的JDBC連接器有兩種方法：

  1. 定義多個連接器，每個連接器都處理單獨(dú)的表；
  2. 定義單個連接器，但增加任務(wù)數(shù)。每個Kafka連接器的工作由一個或多個任務(wù)來執(zhí)行，每個連接器默認(rèn)只有一個任務(wù)，這意味著從數(shù)據(jù)庫中提取數(shù)據(jù)是單進(jìn)程處理的。

前者具有更高的管理開銷，但確實(shí)提供了每個表自定義設(shè)置的靈活性。如果可以使用相同的連接器配置提取所有表，則增加單個連接器中的任務(wù)數(shù)是一種好方法。

當(dāng)增加從數(shù)據(jù)庫中提取數(shù)據(jù)的并發(fā)性時，要從整體上考慮。因?yàn)檫\(yùn)行一百個并發(fā)任務(wù)雖然可能會更快，但數(shù)百個與數(shù)據(jù)庫的連接可能會對數(shù)據(jù)庫產(chǎn)生負(fù)面影響。

以下是同一連接器的兩個示例。兩者都將從數(shù)據(jù)庫中提取所有表，總共6個。在第一個連接器中，未指定最大任務(wù)數(shù)，因此為默認(rèn)值1。在第二個中，指定了最多運(yùn)行三個任務(wù)（"tasks.max":3）：

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/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-01-",
                "mode":"bulk"
                }
        }'


curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
        "name": "jdbc_source_mysql_11",
        "config": {
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "connection.url": "jdbc:mysql://mysql:3306/demo",
                "connection.user": "connect_user",
                "connection.password": "asgard",
                "topic.prefix": "mysql-11-",
                "mode":"bulk",
                "tasks.max":3
                }
        }'

當(dāng)查詢連接器的Kafka連接器RESTAPI時，可以看到每個連接器正在運(yùn)行的任務(wù)數(shù)以及它們已分配的表。第一個連接器有一個任務(wù)負(fù)責(zé)所有6張表：

$ curl -s "http://localhost:8083/connectors/jdbc_source_mysql_01/tasks"|jq '.'
[
  {
    "id": {
      "connector": "jdbc_source_mysql_01",
      "task": 0
    },
    "config": {
      "tables": "`demo`.`NUM_TEST`,`demo`.`accounts`,`demo`.`customers`,`demo`.`transactions`,`security`.`firewall`,`security`.`log_events`",
      …
    }
  }
]

第二個連接器有3個任務(wù)，每個任務(wù)分配2張表：

$ curl -s“http：// localhost：8083 / connectors / jdbc_source_mysql_11 / tasks”| jq'。'
[
  {
    “ID”： {
      “connector”：“jdbc_source_mysql_11”，“任務(wù)”：0
    }，
    “config”：{
      “tables”：“`demo` .NUM_TEST`，`demo` .accounts`”，
      ...
    }
  }，
  {
    “ID”： {
      “connector”：“jdbc_source_mysql_11”，“任務(wù)”：1
    }，
    “config”：{
      “tables”：“`demo``customers`，`demo` .transactions`”，
      ...
    }
  }，
  {
    “ID”： {
      “connector”：“jdbc_source_mysql_11”，“任務(wù)”：2
    }，
    “config”：{
      “tables”：“`security``firewall`，`security``log_events`”，
      ...
    }
  }
]
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容