基于Kafka连接器与Avro格式，在Flink SQL中定义Timestamp的方法

你有一个Kafka主题，消息值采用带Debezium类型的Avro格式，包含两个时间相关字段：

• updated：对应Debezium的io.debezium.time.ZonedTimestamp，底层存储为字符串
• etl_updated：对应Debezium的io.debezium.time.MicroTimestamp，底层存储为长整型

目前你在Flink SQL里将这两个字段定义为STRING和BIGINT，想要转换为Timestamp类型，有两种可行方案：

方案一：利用Flink对Debezium Avro类型的自动映射（推荐） #

Flink的avro-confluent格式原生支持识别Debezium通过connect.name定义的特殊类型，直接修改CREATE TABLE语句的字段类型即可：

• 对于updated（ZonedTimestamp）：映射为Flink的TIMESTAMP_TZ(3)（带时区的毫秒精度时间戳），如果不需要时区信息，也可以用TIMESTAMP_LTZ(3)（本地时区时间戳）
• 对于etl_updated（MicroTimestamp）：映射为Flink的TIMESTAMP(6)（微秒精度时间戳），匹配Debezium存储的微秒级长整型值

修改后的表定义：

CREATE TABLE mytable (
  ...,
  updated TIMESTAMP_TZ(3) NOT NULL,
  etl_updated TIMESTAMP(6) NOT NULL
) WITH (
  'connector' = 'kafka',
  'value.format' = 'avro-confluent',
  -- 若Schema Registry未全局配置，需在此指定地址
  'value.format.avro-confluent.schema-registry.url' = 'http://your-schema-registry:8081',
  ...
)

方案二：手动转换字段类型（自动映射失效时使用） #

如果因版本兼容等问题自动映射不生效，可以先按原类型定义源表，再通过SQL函数转换为Timestamp：

处理updated字段（带时区的字符串） #

用TO_TIMESTAMP_TZ函数将ZonedTimestamp格式的字符串转换为带时区的时间戳：

SELECT
  ...,
  TO_TIMESTAMP_TZ(updated, 'yyyy-MM-dd''T''HH:mm:ssXXX') AS updated_ts,
  ...
FROM mytable

处理etl_updated字段（微秒级长整型） #

将微秒数转换为Flink的微秒精度Timestamp，有两种写法：

SELECT
  ...,
  -- 方式1：拆分秒和微秒部分拼接
  TO_TIMESTAMP(etl_updated / 1000000) + INTERVAL '1' MICROSECOND * (etl_updated % 1000000) AS etl_updated_ts,
  -- 方式2：直接数值转换（更简洁）
  CAST(etl_updated / 1000000.0 AS TIMESTAMP(6)) AS etl_updated_ts,
  ...
FROM mytable

也可以创建视图封装转换逻辑，方便后续查询使用：

CREATE VIEW mytable_with_ts AS
SELECT
  ...,
  TO_TIMESTAMP_TZ(updated, 'yyyy-MM-dd''T''HH:mm:ssXXX') AS updated,
  CAST(etl_updated / 1000000.0 AS TIMESTAMP(6)) AS etl_updated,
  ...
FROM mytable;

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