Kafka Connect 用数组字段展平 postgres 记录的转换 [英] Kafka Connect flatten transformation of a postgres record with array field

问题描述

我有一个使用 Kafka Connect 连接到 Kafka 的 postgres 数据库，以便将 CDC 事件放在一个主题上.我们使用扁平化转换作为共享配置的一部分:

I have a postgres DB connected to Kafka using Kafka Connect, in order to put CDC events on a topic. We use a flatten transformation as part of a shared config:

flattenKey:  "org.apache.kafka.connect.transforms.Flatten$Key"

表中的一列属于 ARRAY 类型，因此在尝试应用转换时出现异常:

One of the columns in the table is of type ARRAY and thus when trying to apply the transformation I get an exception:

Flatten transformation does not support ARRAY for record without schemas (for field after.role_ids).

Kafka 连接参考:https://github.com/a0x8o/kafka/blob/master/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Flatten.java#L246

Kafka connect reference: https://github.com/a0x8o/kafka/blob/master/connect/transforms/src/main/java/org/apache/kafka/connect/transforms/Flatten.java#L246

我知道数组不能展平，为什么，我的问题是 - 有什么办法可以保持记录的展平，但以某种方式转换/转换数组，以便我仍然可以使用它?

I understand that arrays can't be flattened and why, my question is - is there any way I can somehow keep the flatten of the record, but cast / transform the array in some manner so I can still use it?

该数组将包含整数，因此将其转换为一个字符串，其中所有元素都由逗号分隔，例如对我来说是个技巧.任何其他建议都会很棒.

the array will contain integers, so transforming it into a string where all the elements are separated by commas for example do the trick for me. Any other suggestion will be great.

我们使用 Debezium 进行配置.

We use Debezium for the configuration.

推荐答案

在 JDBC sink 连接器中，Array 原语通过票证 https://github.com/confluentinc/kafka-connect-jdbc/pull/805

In the JDBC sink connector, the Array primitives are handled via the ticket https://github.com/confluentinc/kafka-connect-jdbc/pull/805

这篇关于Kafka Connect 用数组字段展平 postgres 记录的转换的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！