Flink1.11下如何将Kinesis接入的Unix时间戳设为PyFlink水位线

核心修改逻辑 #

• 先用BIGINT类型接收原始的13位毫秒级Unix时间戳event_time
• 通过Flink内置的时间函数将数值类型的时间戳转换为TIMESTAMP类型，再基于转换后的字段定义水位线

修改后的建表语句 #

CREATE TABLE event_input_table (
    event_time BIGINT,
    ticker VARCHAR,
    ts AS TO_TIMESTAMP_LTZ(event_time, 3),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kinesis',
    'stream' = 'inputstream1',
    'aws.region' = 'us-east-1',
    'scan.stream.initpos' = 'TRIM_HORIZON',
    'format' = 'json' ,
    'aws.credentials.provider' = 'ENV_VAR' 
)

这里TO_TIMESTAMP_LTZ的第二个参数3代表输入的时间戳精度为毫秒（10^3），如果你后续切换为秒级10位时间戳，把这个参数改为0即可。

完整Python Table API 实现示例 #

from pyflink.table import EnvironmentSettings, TableEnvironment

# 初始化流处理环境
env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
table_env = TableEnvironment.create(env_settings)

# 执行建表语句
table_env.execute_sql("""
CREATE TABLE event_input_table (
    event_time BIGINT,
    ticker VARCHAR,
    ts AS TO_TIMESTAMP_LTZ(event_time, 3),
    WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kinesis',
    'stream' = 'inputstream1',
    'aws.region' = 'us-east-1',
    'scan.stream.initpos' = 'TRIM_HORIZON',
    'format' = 'json' ,
    'aws.credentials.provider' = 'ENV_VAR' 
)
""")

# 后续可以基于event_input_table做窗口等需要水位线的计算
# 示例：打印表结构验证
table_env.execute_sql("DESCRIBE event_input_table").print()

注意事项 #

• Flink 1.11版本已原生支持TO_TIMESTAMP_LTZ函数，无需额外引入依赖
• 原本配置的5秒水位线延迟可根据业务场景的事件乱序程度自行调整
• AWS托管Flink运行时自带Kinesis连接器，本地测试时如果缺少依赖需要手动引入对应版本的flink-connector-kinesis包

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