Flink 1.16从数据流定义表时，事件时间与处理时间的配置方法

Flink 1.15起，StreamTableEnvironment.fromDataStream(DataStream, Expression...)方法被弃用，现在需要通过显式定义Schema来声明事件时间和处理时间，以下是具体实现方案：

一、事件时间（Rowtime）配置 #

事件时间的配置需要结合字段定义和水印策略，分两种场景：

1. 从DataStream已有字段提取事件时间 #

如果数据流中本身包含事件时间戳字段，直接在Schema中声明并绑定水印：

Table transactionTable = transactionDataStream.toTable(tableEnv,
    Schema.newBuilder()
        .column("field1", DataTypes.STRING())
        .column("field2", DataTypes.INT())
        .column("field3", DataTypes.DOUBLE())
        .column("transactionTime", DataTypes.TIMESTAMP(3))
        // 配置水印，允许5秒延迟
        .watermark("transactionTime", "transactionTime - INTERVAL '5' SECOND")
        .build()
);

2. 从数据源元数据获取事件时间 #

如果事件时间来自数据源的元数据（比如Kafka消息的时间戳），可以通过columnByMetadata直接读取：

Table transactionTable = transactionDataStream.toTable(tableEnv,
    Schema.newBuilder()
        .column("field1", DataTypes.STRING())
        .column("field2", DataTypes.INT())
        .column("field3", DataTypes.DOUBLE())
        // 从数据源的rowtime元数据中获取事件时间
        .columnByMetadata("transactionTime", DataTypes.TIMESTAMP(3), "rowtime")
        .watermark("transactionTime", "transactionTime - INTERVAL '5' SECOND")
        .build()
);

二、处理时间（Proctime）配置 #

处理时间不需要依赖数据流中的字段，直接通过PROCTIME()函数生成新字段即可：

Table transactionTable = transactionDataStream.toTable(tableEnv,
    Schema.newBuilder()
        .column("field1", DataTypes.STRING())
        .column("field2", DataTypes.INT())
        .column("field3", DataTypes.DOUBLE())
        // 添加处理时间字段ts
        .columnByExpression("ts", "PROCTIME()")
        .build()
);

替代调用方式：使用TableEnvironment.fromDataStream #

上述示例用的是DataStream.toTable()，你也可以直接调用TableEnvironment.fromDataStream的重载方法，用法完全一致：

// 事件时间示例
Table transactionTable = tableEnv.fromDataStream(transactionDataStream,
    Schema.newBuilder()
        .column("field1", DataTypes.STRING())
        .column("field2", DataTypes.INT())
        .column("field3", DataTypes.DOUBLE())
        .column("transactionTime", DataTypes.TIMESTAMP(3))
        .watermark("transactionTime", "transactionTime - INTERVAL '5' SECOND")
        .build()
);

// 处理时间示例
Table transactionTable = tableEnv.fromDataStream(transactionDataStream,
    Schema.newBuilder()
        .column("field1", DataTypes.STRING())
        .column("field2", DataTypes.INT())
        .column("field3", DataTypes.DOUBLE())
        .columnByExpression("ts", "PROCTIME()")
        .build()
);

注意事项 #

• 显式Schema定义更符合Flink的类型安全设计，避免旧API中可能出现的类型推断问题
• 水印是事件时间语义的核心，必须为事件时间字段配置合理的水印策略
• 处理时间字段是Flink运行时生成的，不需要从数据流中读取或转换

内容的提问来源于stack exchange，提问作者Mustafa İrfan Değerli