Kafka 连接 MSSQL 中的多对多表

Question

我目前正在研究 Kafka Connect，以将我们的一些数据库流式传输到数据湖。为了测试 Kafka Connect，我设置了一个包含我们项目数据库之一的数据库。到目前为止一切都很好。

下一步我使用以下属性配置了 Kafka Connect：

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "timestamp.column.name": "updated_at,created_at",
  "incrementing.column.name": "id",
  "dialect.name": "SqlServerDatabaseDialect",
  "validate.non.null": "false",
  "tasks.max": "1",
  "mode": "timestamp+incrementing",
  "topic.prefix": "mssql-jdbc-",
  "poll.interval.ms": "10000",
}

虽然这适用于我获得 ID 和 created_at / updated_at 字段的大多数表，但它不适用于我解决了多对多关系的表和一个复合表钥匙。请注意，我将通用 JDBC 配置与 Microsoft 的 JDBC 驱动程序一起使用。

有没有办法针对这些特殊情况配置 Kafka Connect？

Answer 1

您可能需要创建多个连接器，而不是一个连接器来拉取所有表。如果您想使用不同的方法来获取数据或不同的 ID/时间戳列，就会出现这种情况。 正如@cricket_007 所说，您可以使用query 选项来拉回查询的结果——这可能是一个SELECT，表示您的多表连接。即使从单个表对象中提取数据，JDBC 连接器本身也只是从给定表中发出SELECT *，并使用WHERE 谓词来限制基于递增的 ID/时间戳选择的行。

另一种方法是使用基于日志的变更数据捕获 (CDC)，并将所有变更直接从数据库流式传输到 Kafka。

无论您使用 JDBC 还是基于日志的 CDC，您都可以使用流处理来解析 Kafka 本身中的连接。这方面的一个例子是 Kafka Streams 或 KSQL。我写过关于后者的lot here。

您可能还会发现 this article 详细描述了您将数据库与 Kafka 集成的选项。

免责声明：我为开源 KSQL 项目背后的公司 Confluent 工作。