如何根据连接器名称获取 Kafka 源连接器架构

Question

我正在使用 Confluent JDBC Kafka 连接器将消息发布到主题中。源连接器将在每次轮询时将数据连同架构一起发送到主题。我想检索此模式。

可能吗？如何？谁能推荐我

我的意图是创建一个 KSQL 流或 table 基于 Kafka 连接器在轮询时构建的模式。

Answer 1

执行此操作的最佳方法是使用 Avro，其中模式单独存储并由 Kafka Connect 和 KSQL 自动使用。

您可以通过将 Kafka Connect 配置为使用 AvroConverter 来使用 Avro。在您的 Kafka Connect worker 配置集中：

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081

(将 schema-registry 更新为您的模式注册表所在的主机名 运行)

从那里开始，在 KSQL 中您只需使用

CREATE STREAM my_stream WITH (KAFKA_TOPIC='source_topic', VALUE_FORMAT='AVRO');

您不需要在此处指定架构本身，因为 KSQL 从架构注册表中获取它。

您可以阅读有关转换器和串行器的更多信息 here。

免责声明：我为 Confluent 工作，并撰写了参考博客 post。