基于查询的JDBC源连接器Kafka

Question

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。在worker的日志中也没有错误日志，当我检查它时，在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...)之类的聪明技巧

Answer 1

所以几个小时后玩不同的配置。我回到官方文件，意识到这一点

使用自定义查询而不是加载表，允许您联接来自多个表的数据。只要查询不包括自己的筛选，您仍然可以对增量查询使用内置模式（在本例中，使用时间戳列）。请注意，这限制了每个连接器只能有一个输出，而且由于没有表名，因此主题“prefix”在本例中实际上是完整的主题名。

所以关键是“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,

    }
}