如何首先从查询中获取所有数据，然后使用kafka连接器以增量方式只进行更改。
也许这对你有帮助。例如，我有一个表：

╔════╦═════════════╦═══════════╗
║ Id ║    Name     ║  Surname  ║
╠════╬═════════════╬═══════════╣
║  1 ║ Martin      ║ Scorsese  ║
║  2 ║ Steven      ║ Spielberg ║
║  3 ║ Christopher ║ Nolan     ║
╚════╩═════════════╩═══════════╝

在这种情况下，我将创建一个视图：

CREATE OR REPLACE VIEW EDGE_DIRECTORS AS
SELECT 0 AS EXID, ID, NAME, SURNAME
FROM DIRECTORS WHERE ID =< 2
UNION ALL
SELECT ID AS EXID, ID, NAME, SURNAME
FROM DIRECTORS WHERE ID > 2;

在kafka jdbc连接器的属性文件中，您可以使用：

connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
mode=incrementing
incrementing.column.name=EXID
topic.prefix=
tasks.max=1
name=gv-jdbc-source-connector
connection.url=
table.types=VIEW
table.whitelist=EDGE_DIRECTORS

因此，kafka jdbc连接器将采取以下步骤：
首先是exid=0的所有数据；
它将在connector.offsets文件中存储偏移值=0；
新行将插入到directors表中。
kafka jdbc连接器将执行： Select EXID, ID, NAME, SURNAME FROM EDGE_DIRECTORS 并且会注意到exid已经增加了。
数据将在Kafka流中更新。

很长一段时间后又回到这个主题。该方法能够解决这个问题，而且不必使用批量模式
停止连接器
擦除每个连接器jvm的偏移文件
（可选）如果您想执行完整的擦除和加载，您可能还想使用kafka/connect-utils/restapi删除主题（不要忘记state主题）
重新启动连接。