基于查询的 JDBC 源连接器 Kafka答案

【问题标题】：query-based JDBC Source connector Kafka基于查询的 JDBC 源连接器 Kafka
【发布时间】：2019-11-13 05:55:10
【问题描述】：

我有一个旧数据库，它的主键列是字符串（是的，我知道）。我想使用 JDBC kafka Source Connector 将 increment 从 postgres DB 转储到 kafka 主题中

下面是我尝试重现问题

create table test(
id varchar(20) primary key,
name varchar(10) 
);

INSERT INTO test(
    id, name)
VALUES ('1ab', 't'),
('2ab', 't'),
('3ab', 't')

我的配置

{"name" : "test_connector",
    "config" : {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.url": "jdbc:postgresql://12.34.5.6:5432/",
        "connection.user": "user",
        "connection.password": "password",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "topic.prefix": "incre_",
        "mode": "incrementing",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "query" :"SELECT cast(replace(id, 'ab','') as integer) as id , name from test ORDER BY id ASC",
        "incrementing.column.name":"id",
        "value.converter.schema.registry.url": "http://schema-registry_url.com",
        "key.converter.schema.registry.url": "http://schema-registry_url.com",
        "offset.flush.timeout.ms": 2000,

    }
}

在我发布配置后，当我执行 HTTP curl 时，状态为 RUNNING。当我检查它时，工人的日志中也没有错误日志当我尝试做一个控制台消费者时，kafka 主题中也没有数据我还尝试了其他几种组合，例如添加"table.whitelist": "test"。

我尝试的另一件事是关注这两个链接 https://rmoff.net/2018/05/21/kafka-connect-and-oracle-data-types/ https://www.confluent.io/blog/kafka-connect-deep-dive-jdbc-source-connector 但无济于事，即使是像 SELECT * from (SELECT id, name from test where ...) 这样建议的聪明技巧

【问题讨论】：

标签： jdbc apache-kafka apache-kafka-connect

【解决方案1】：

所以在玩了几个小时后，使用了不同的配置。我回到官方document 并意识到这一点

使用自定义查询而不是加载表，允许您连接来自多个表的数据。只要查询不包含自己的过滤，您仍然可以使用内置模式进行增量查询（在这种情况下，使用时间戳列）。请注意，这将您限制为每个连接器的单个输出，并且由于没有表名，因此在这种情况下，主题“前缀”实际上是完整的主题名称。

所以关键是"topic.prefix": "incre_test"

跟进之前的设置，正确的配置应该是

{"name" : "test_connector",
    "config" : {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.url": "jdbc:postgresql://12.34.5.6:5432/",
        "connection.user": "user",
        "connection.password": "password",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "topic.prefix": "incre_test",
        "mode": "incrementing",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "query" :"SELECT cast(replace(id, 'ab','') as integer) as id , name from test ORDER BY id ASC",
        "incrementing.column.name":"id",
        "value.converter.schema.registry.url": "http://schema-registry_url.com",
        "key.converter.schema.registry.url": "http://schema-registry_url.com",
        "offset.flush.timeout.ms": 2000,

    }
}

【讨论】：

我在文档中没有遇到过这个，但是在使用查询时要记住这一点

【解决方案2】：

恐怕您不能在incrementing 模式下使用您的varchar id，因为它不是递增的列/类型。根据Confluent Docs，

递增列：包含每行唯一 ID 的单个列，其中更新的行保证具有更大的 ID，即 AUTOINCREMENT 专栏。请注意，此模式只能检测新行。无法检测到对现有行的更新，因此这种模式应该只用于不可变数据。您可能会使用此模式的一个示例是在数据仓库中流式传输事实表时，因为这些是通常只插入。

【讨论】：