Confluent Kafka SMT：时间戳转换问题答案

【问题标题】：Confluent Kafka SMT: Timestamp conversion issueConfluent Kafka SMT：时间戳转换问题
【发布时间】：2021-05-29 20:47:47
【问题描述】：

我对 Kafka 完全陌生，并且一直在尝试让一些基本用例工作。

我想要完成什么？从 Oracle 中的表将数据加载到 Kafka 主题中。（使用基于查询的 CDC）

目标：当我从主题中读取数据时，获取正确的日期格式。

日期列的数据作为纪元出现。我在连接器配置中包含了时间戳转换步骤，以将其转换为 dd-MM-yyyy hh:mm:ss 格式，但无法使其正常工作。

以下是我在连接器配置中的内容。我尝试了几种格式组合，将目标类型更改为字符串，但我仍然看到纪元值。

            "incrementing.column.name": "MD_LOAD_NUMBER",
            "validate.non.null": "false",
            "key.converter": "io.confluent.connect.avro.AvroConverter",
            "key.converter.schema.registry.url": "http://localhost:8081",
            "value.converter": "io.confluent.connect.avro.AvroConverter",
            "value.converter.schema.registry.url": "http://localhost:8081",
            "value.converter.schemas.enable": "false",

            "transforms":"insrt_dt, updt_dt",
            "transforms.insrt_dt.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
            "transforms.insrt_dt.target.type": "Timestamp",
            "transforms.insrt_dt.field": "MD_INSERT_DATE",
            "transforms.insrt_dt.format": "yyyy-MM-dd HH:mm:ss",
            "transforms.updt_dt.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
            "transforms.updt_dt.target.type": "Timestamp",
            "transforms.updt_dt.field": "MD_UPDATE_DATE",
            "transforms.updt_dt.format": "yyyy-MM-dd HH:mm:ss"

ksql> 从头开始打印 biORGANIZATION_DIM；

密钥格式：SESSION(AVRO) or HOPPING(AVRO) or TUMBLING(AVRO) or AVRO or SESSION(PROTOBUF) or HOPPING(PROTOBUF) or TUMBLING(PROTOBUF) or PROTOBUF or SESSION(JSON) or HOPPING(JSON) or TUMBLING(JSON) 或 JSON 或 SESSION(JSON_SR) 或 HOPPING(JSON_SR) 或 TUMBLING(JSON_SR) 或 JSON_SR 或 SESSION(KAFKA_INT) 或 HOPPING(KAFKA_INT) 或 TUMBLING(KAFKA_INT) 或 KAFKA_INT 或 SESSION(KAFKA_BIGINT) 或 HOPPING(KAFKA_BIGINT)或 TUMBLING(KAFKA_BIGINT) 或 KAFKA_BIGINT 或 SESSION(KAFKA_DOUBLE) 或 HOPPING(KAFKA_DOUBLE) 或 TUMBLING(KAFKA_DOUBLE) 或 KAFKA_DOUBLE 或 SESSION(KAFKA_STRING) 或 HOPPING(KAFKA_STRING) 或 TUMBLING(KAFKA_STRING) 或 KAFKA_STRING

值格式：AVRO

行时间：太平洋夏令时间 6/4/20 上午 10:49:19，键：，值：{“ORGANIZATION_KEY”：-1，“ORGANIZATION_ID”：null，“ORGANIZATION_NAME”：“UNASSIGNED”，“SET_OF_BOOKS_ID”：null , "MD_INSERT_DATE": 1404321287000, "MD_UPDATE_DATE": 1404321287000, "MD_DELETED_FLAG": "N"}

我提到了https://www.confluent.de/blog/kafka-connect-deep-dive-jdbc-source-connector/和https://docs.confluent.io/current/connect/transforms/timestampconverter.html

如何将日期转换为易于阅读的格式？感谢您提供任何帮助，并感谢您阅读长篇文章！

版本：融合本地版本本地命令适用于单节点开发环境仅用于生产用途。 https://docs.confluent.io/current/cli/index.html

Confluent 社区软件：5.5.0

【问题讨论】：

标签： apache-kafka apache-kafka-connect confluent-platform

【解决方案1】：

更改为 transforms.insrt_dt.target.type :"string"。它应该可以工作。

【讨论】：

【解决方案2】：

根据Debezium Documentation可以添加属性：

"time.precision.mode":"connect"

这将使 oracle 日期类型被视为 Kafka Connect 内置日期类型。

我在此之前回答了question，这似乎与 MySQL 源连接器类似。

【讨论】：