如何在Kafka JDBC Sink Connector中处理嵌套结构体数组

针对你遇到的Flatten转换不支持数组的报错，以下两种方案均无需依赖KSQL或Kafka Streams，仅通过Kafka Connect即可实现：

方案1：将数组字段映射为PostgreSQL JSONB类型 #

PostgreSQL原生支持JSONB类型存储复杂嵌套结构，无需展开数组，直接将Kafka中的嵌套结构体数组序列化为JSON写入即可。

配置步骤： #

1. 在PostgreSQL中创建目标表时，将对应数组字段定义为JSONB类型：

   CREATE TABLE target_table (
       id INT PRIMARY KEY,
       normal_field VARCHAR(255),
       mylist JSONB
   );
   

2. 配置Kafka Connect JDBC Sink Connector：
   • 使用JSON转换器处理消息值，根据你的Kafka消息是否带Schema调整配置：

     value.converter=org.apache.kafka.connect.json.JsonConverter
     value.converter.schemas.enable=false  # 消息无Schema时设为false，有Schema则设为true
     

   • 若仅需处理特定数组字段，可搭配ExtractField转换提取目标字段，直接写入JSONB列。

方案2：用ExpandArray SMT拆分数组为单条记录 #

如果需要将数组中的每个结构体单独写入数据库（比如拆分到关联表），可使用Confluent的ExpandArray SMT（AWS MSK Connect支持安装该插件），将数组元素拆分为独立的Kafka记录后，再用Flatten处理结构体。

配置步骤： #

1. 在MSK Connect中安装Confluent Transformations插件（未预装时需手动添加）。
2. 在Sink Connector配置中添加转换规则：

   transforms=expandArray,flatten
   transforms.expandArray.type=io.confluent.connect.transforms.ExpandArray$Value
   transforms.expandArray.field=Name.x.mylist  # 指定要拆分的数组字段完整路径
   transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
   transforms.flatten.delimiter=_  # 结构体字段的分隔符，如将Name.x.field转为Name_x_field
   

3. 调整PostgreSQL表结构，适配拆分后的单条结构体字段（去掉数组维度，直接存储结构体的各个字段）。

注意事项 #

• 采用JSONB方案时，后续可通过PostgreSQL的JSON函数（如jsonb_array_elements）展开数组进行查询分析。
• 使用ExpandArray时，原记录的其他字段会被复制到每条拆分后的记录中，可通过ReplaceField过滤不必要的字段，避免冗余数据。

内容的提问来源于stack exchange，提问作者repcak