如何在Flink SQL中无需UDF转换TIMESTAMP至毫秒及用Table API表示Kafka事件时间？

1. 无需UDF转换TIMESTAMP(3)/TIMESTAMP_LTZ(3)为毫秒值 #

完全可以通过Flink SQL内置能力实现，无需自定义UDF，两种常用方式：

• 直接CAST为BIGINT：这是最高效的方案，因为Flink中TIMESTAMP(3)和TIMESTAMP_LTZ(3)类型内部就是以UTC 1970-01-01 00:00:00为起点的毫秒数存储的，直接转换就能拿到原始毫秒值：

  SELECT CAST(event_time AS BIGINT) AS event_time_ms FROM your_table;
  

• 使用UNIX_TIMESTAMP函数计算：UNIX_TIMESTAMP返回指定时间到UTC 1970-01-01 00:00:00的秒数，乘以1000即可得到毫秒值，适合需要先处理秒级时间的场景：

  SELECT UNIX_TIMESTAMP(event_time) * 1000 AS event_time_ms FROM your_table;
  

  注意：TIMESTAMP_LTZ类型会自动处理时区转换，最终计算的是UTC时间对应的毫秒数。

2. Table API以毫秒形式表示Kafka事件时间 #

可以实现，但需明确：Flink的事件时间属性要求为TIMESTAMP或TIMESTAMP_LTZ类型，但你可以随时将其转换为毫秒数值（BIGINT类型）使用或输出，分两种场景处理：

场景1：原始Kafka数据中的事件时间是毫秒数值 #

先将毫秒数值转换为带水印的事件时间属性，之后按需转回毫秒值：

// 定义Kafka源Schema，把毫秒字段转为事件时间属性
Schema schema = Schema.newBuilder()
    .column("event_time_ms", DataTypes.BIGINT())
    .column("payload", DataTypes.STRING())
    // 将毫秒值转为TIMESTAMP_LTZ类型作为事件时间
    .columnByExpression("event_time", "TO_TIMESTAMP_LTZ(event_time_ms, 3)")
    // 设置水印，处理乱序数据
    .watermark("event_time", "event_time - INTERVAL '5' SECOND")
    .build();

// 注册Kafka源
TableSource<?> kafkaSource = KafkaTableSource.builder()
    .setBootstrapServers("localhost:9092")
    .setTopic("your_topic")
    .setSchema(schema)
    .setFormat(new JsonFormat())
    .build();
tableEnv.registerTableSource("kafka_events", kafkaSource);

// 查询时直接用原始毫秒字段，或把事件时间转回毫秒
Table result = tableEnv.from("kafka_events")
    .select(
        $("event_time_ms"), // 直接使用原始毫秒值
        CAST($("event_time").as(DataTypes.BIGINT())).as("converted_ms"), // 事件时间转毫秒
        $("payload")
    );

场景2：事件时间是TIMESTAMP类型，需要输出为毫秒值 #

如果Kafka源的事件时间已经是TIMESTAMP/TIMESTAMP_LTZ类型，直接通过CAST转换为BIGINT即可得到毫秒值：

Table result = tableEnv.from("kafka_events")
    .select(
        CAST($("event_time").as(DataTypes.BIGINT())).as("event_time_ms"),
        $("payload")
    );

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