Kafka-Elasticsearch Sink用AVRO Schema时Decimal/Date值转换异常求助

问题本质 #

你遇到的Decimal字段输出为AJiwXlCA是因为：AvroConverter将Decimal逻辑类型编码为字节数组后，Elasticsearch Sink默认未解析Kafka Connect的逻辑类型，直接把字节数组转成Base64字符串存入ES；Date类型同理，未被转换为ES兼容的时间格式。

一、修复Decimal类型问题 #

方法1：使用Kafka Connect Transform转换Decimal为数值字符串

在连接器配置中添加以下Transform规则，将Decimal字节数组直接转换为带精度的数值字符串：

# 配置Decimal转换Transform
transforms=convertDecimal
transforms.convertDecimal.type=org.apache.kafka.connect.transforms.Decimal$Value
transforms.convertDecimal.field=salesRevenue
transforms.convertDecimal.scale=4  # 与Schema中定义的scale一致

配置后，salesRevenue会被转换为144800000.0000（或对应精度的字符串），Elasticsearch会自动识别为数值类型。

方法2：配置AvroConverter直接解析Decimal逻辑类型

修改AvroConverter的配置，让它将Decimal逻辑类型解析为原生数值格式，而非字节数组：

value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://你的SchemaRegistry地址:8081
# 开启Decimal逻辑类型解析为JSON兼容格式
value.converter.avro.logical.type.decimal.format=plain

此配置会让AvroConverter直接输出Decimal的数值字符串，Elasticsearch Sink无需额外转换即可正确存储。

二、修复Date类型问题 #

使用TimestampConverter Transform将Date类型转换为Elasticsearch兼容的格式（ISO字符串或时间戳）：

# 配置Date转换Transform
transforms=convertDate
transforms.convertDate.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
transforms.convertDate.field=你的日期字段名  # 替换为实际Date字段名
# 可选：转换为ISO格式字符串
transforms.convertDate.target.type=string
transforms.convertDate.format=yyyy-MM-dd'T'HH:mm:ss.SSSZ
# 或转换为long类型时间戳
# transforms.convertDate.target.type=long

三、额外注意事项 #

• 确保Elasticsearch Sink配置中schema.ignore=false，让Sink根据Schema自动映射字段类型；
• 如果使用多个Transform，可用逗号分隔（如transforms=convertDecimal,convertDate），并分别配置每个Transform的参数；
• 验证Schema注册表中的Decimal字段scale和precision与Transform配置一致，避免精度丢失。

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