KafkaJDBCsink不支持十进制/数字反序列化？

可以使用Kafka Connect Decimal插件来解决此问题。此插件为Kafka Connect提供了用于处理十进制数据类型的支持。

1.首先，在Kafka Connect工作流中安装Kafka Connect Decimal插件。以下是使用Confluent Hub CLI进行安装的示例代码。

$ confluent-hub install --no-prompt confluentinc/kafka-connect-datagen:latest

2.接下来，在Kafka Connect的配置文件中启用Decimal插件。以下是示例配置文件中的示例代码。

... "transforms": "unwrap,insertDecimal", "transforms.unwrap.type": "io.confluent.connect.transforms.UnwrapFromEnvelope", "transforms.insertDecimal.type": "io.confluent.connect.transforms.InsertField$Value", "transforms.insertDecimal.timestamp.field": "insert_timestamp", "transforms.insertDecimal.target.type": "decimal", "transforms.insertDecimal.precision": "38", "transforms.insertDecimal.scale": "8" ...

3.最后，在Java应用程序中使用序列化/反序列化程序来处理十进制数据类型。以下是Java代码示例。

... ObjectMapper mapper = new ObjectMapper(); SimpleModule module = new SimpleModule(); module.addSerializer(BigDecimal.class, new BigDecimalSerializer()); module.addDeserializer(BigDecimal.class, new BigDecimalDeserializer()); mapper.registerModule(module); ... public class BigDecimalSerializer extends JsonSerializer<BigDecimal> { @Override public void serialize(BigDecimal value, JsonGenerator gen, SerializerProvider serializers) throws IOException { gen.writeString(value.toString()); } } ... public class BigDecimalDeserializer extends JsonDeserializer<BigDecimal> { @Override public BigDecimal deserialize(JsonParser p, DeserializationContext ctxt) throws IOException { String value = p.getValueAsString(); try { return new BigDecimal(value); } catch (NumberFormatException e) { throw new IOException("Invalid BigDecimal value: " + value, e); } } } ...