debezium sqlserver 连接器输出数字/十进制字段的编码值答案

【问题标题】：debezium sqlserver connector outputs encoded values for numeric/decimal fieldsdebezium sqlserver 连接器输出数字/十进制字段的编码值
【发布时间】：2020-07-18 00:26:35
【问题描述】：

我在 SQL Server 中有这张表：

CREATE TABLE [dbo].[blast_info](
    [blast_id] [int] NOT NULL,
    [tnt_amount_kg] [decimal](18, 2) NOT NULL,
    [time_blasted] [datetime] NOT NULL,
    [hole_deep_ft] [numeric](9, 2) NULL,
    [hole_coord_n] [numeric](18, 6) NOT NULL,
    [hole_coord_e] [numeric](18, 6) NULL
) ON [PRIMARY]

debezium 配置为从 confluent 作为插件运行。数据被发布到 Kafka，但是当我通过 python 或控制台消费者读取它时，我看到了数字和十进制类型的编码值：

{"blast_id":17,"tnt_amount_kg":"eA==","time_blasted":1585803600000,"hole_deep_ft":"AOY=","hole_coord_n":"A/OOVvYA","hole_coord_e":"AKSQkBwA","__ts_ms":1586140437125}
{"blast_id":16,"tnt_amount_kg":"ANw=","time_blasted":1583125200000,"hole_deep_ft":"Aa4=","hole_coord_n":"A/OOVvYA","hole_coord_e":"AKSQkBwA","__ts_ms":1586140437125}
{"blast_id":17,"tnt_amount_kg":"eA==","time_blasted":1585803600000,"hole_deep_ft":"AOY=","hole_coord_n":"A/OOVvYA","hole_coord_e":"AKSQkBwA","__ts_ms":1586140437126}
Processed a total of 38 messages

为什么会这样？解决方法是什么？谢谢。

【问题讨论】：

您是否已将 Debezium 配置为精确处理 DECIMAL 值？
debezium.io/documentation/faq/…
@MitchWheat 我没有。文档说这是默认设置（尽管对于 Postgres 连接器）。会给它我拍。谢谢。
您能否粘贴您的连接器配置并关注 kafka 连接属性。 1.key.converter 2.value.converter 3.key.converter.schemas.enable 4.value.converter.schemas.enable @DmitryBuzolin
@RaviDesai 这里是："key.converter.schemas.enable": "false", "internal.key.converter": "org.apache.kafka.connect.json.JsonConverter", "internal.value.converter.schemas.enable": "false", "value.converter.schemas.enable": "false", "internal.value.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter": "org.apache.kafka.connect.json.JsonConverter",

标签： sql-server apache-kafka confluent-platform connector debezium

【解决方案1】：

可能的快速修复（完全不推荐）：

只需在 SQL Server 中使用 real 数据类型而不是 numeric 或 decimal，因为 Debezium 会将 real 存储为 float。

长期修复：

如Debezium SQL server Connector Documentation 中所述，它将decimal 和numeric 值存储为binary，由类org.apache.kafka.connect.data.Decimal 表示。

您可以从消息本身检索此信息，但为此您需要在消息中启用架构。您可以通过设置key.converter.schemas.enable=true（用于消息键）和value.converter.schemas.enable=true（用于消息值）来做到这一点。

更改上述属性后，您的消息将包含架构信息。参考这个例子：

表架构：

CREATE TABLE [dbo].[kafka_datatype](
    [id] [int] IDENTITY(1,1)  PRIMARY KEY,
    [col_value] [varchar](10) NULL,
    [create_date] [datetime] NULL,
    [col_decimal] [decimal](33, 18) NULL,
    [col_double] [real] NULL,
    [comments] [varchar](5000) NULL
)

卡夫卡消息：

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "int32",
        "optional": false,
        "field": "id"
      },
      {
        "type": "string",
        "optional": true,
        "field": "col_value"
      },
      {
        "type": "int64",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Timestamp",
        "version": 1,
        "field": "create_date"
      },
      {
        "type": "bytes",
        "optional": true,
        "name": "org.apache.kafka.connect.data.Decimal",
        "version": 1,
        "parameters": {
          "scale": "18",
          "connect.decimal.precision": "33"
        },
        "field": "col_decimal"
      },
      {
        "type": "float",
        "optional": true,
        "field": "col_double"
      },
      {
        "type": "string",
        "optional": true,
        "field": "comments"
      }
    ],
    "optional": true,
    "name": "TEST.dbo.kafka_datatype.Value"
  },
  "payload": {
    "id": 15,
    "col_value": "test",
    "create_date": 1586335960297,
    "col_decimal": "AKg/JYrONaAA",
    "col_double": 12.12345,
    "comments": null
  }
}

请阅读Debezium SQL server Connector Documentation 了解 Debezium 如何处理数据类型。

现在来到消费者部分，根据您的需要使用接收器连接器（例如JDBC Sink Connector）。如果你想使用 python 或控制台消费者，你需要编写自己的反序列化器。

附：

随着时间的推移可能会出现一个问题，主题大小会随着架构存储的每条消息而增加。为避免将模式持久化到消息中，您可以使用Avro Converter 和 Schema Registry 由 Confluent 提供。

希望这会有所帮助！

【讨论】：