如何使用 Kafka JDBC 连接器跟踪具有特定列值的行(按 id)?

Question

我有一个包含大量记录的表。有一列定义记录的类型。我想收集该列中具有特定值的记录。有点：

Select * FROM myVeryOwnTable WHERE type = "VERY_IMPORTANT_TYPE" 

我注意到当我选择增量(+时间戳)模式时，我不能在自定义查询中使用WHERE子句，否则如果自己过滤，我需要小心。 我想实现的背景是我使用Logstac将某种类型的数据从MySQL传输到ES。通过使用可以包含 where 子句的查询，这很容易实现。但是，使用 Kafka，我可以在数据库中插入新行后更快地(几乎立即)传输数据。

感谢您的任何提示或建议。

---

多亏了@wardziniak我才能设置它。

query=select * from (select * from myVeryOwnTable p where type = 'VERY_IMPORTANT_TYPE') p
topic.prefix=test-mysql-jdbc-
incrementing.column.name=id

但是，我期待一个主题test-mysql-jdbc-myVeryOwnTable所以我已经注册了我的消费者。但是，使用上面显示的查询表名称被跳过，因此我的主题完全按照上面定义的前缀命名。所以我刚刚更新了我的属性topic.prefix=test-mysql-jdbc-myVeryOwnTable它似乎工作得很好。

Answer 1

可以在Jdbc 源连接器query属性中使用子查询。

JDBC 源连接器配置示例：

{
...
"query": "select * from (select * from myVeryOwnTable p where type = 'VERY_IMPORTANT_TYPE') p",
"incrementing.column.name": "id",
...
}