如何组合/连接两个 jdbc sink 连接器转换？答案

【问题标题】：How to combine/concat two jdbc sink connector Transforms?如何组合/连接两个 jdbc sink 连接器转换？
【发布时间】：2021-01-14 02:09:57
【问题描述】：

在我的 Avro 模式中，有两个字符串字段，field1 和 field2。目前是否可以在 JDBC Sink 连接器上使用内置转换将这两个 Avro 字段组合成一个数据库列（通过 concat，或带有分隔符的 concat 的奖励积分，例如 . 或 :）？我熟悉一些基本的变换，例如：

transforms=myTransform
transforms.myTransform.type:org.apache.kafka.connect.transforms.InsertField$Value
transforms.myTransform.static.field:field2
transforms.myTransform.topic.value:field2

我实际上是在尝试执行其中两个topic.value 查找，然后将它们连接到单个下游数据库列中。

【问题讨论】：

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

【解决方案1】：

不，据我所知，没有单一消息转换可以做到这一点。

您可以自己编写，也可以使用 ksqlDB 来实现：

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

CREATE STREAM concated WITH (KAFKA_TOPIC='target_topic' 
  AS SELECT field1 + '.' + field2 AS concat_field, * 
     FROM mysource

从那里您可以使用 JDBC Sink 连接器流式传输 target_topic（FWIW，如果需要，您可以在 ksqlDB 中使用 CREATE SINK CONNECTOR 进行操作）

【讨论】：