实时计算 新购活动

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import TableEnvironment, DataTypes
from pyflink.table.udf import udf

# 创建执行环境
env = StreamExecutionEnvironment.get_execution_environment()
t_env = TableEnvironment.create(env)

# 定义数据源（模拟新购活动数据流）
source_ddl = """
    CREATE TABLE purchase_events (
        user_id INT,
        product_id INT,
        purchase_time TIMESTAMP(3),
        WATERMARK FOR purchase_time AS purchase_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'purchase_topic',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
"""
t_env.execute_sql(source_ddl)

# 定义实时计算逻辑（例如：计算每分钟的购买次数）
@udf(input_types=[DataTypes.INT(), DataTypes.TIMESTAMP(3)], result_type=DataTypes.INT())
def count_purchases_per_minute(user_id, purchase_time):
    # 这里可以编写具体的计算逻辑
    return 1  # 示例返回固定值

t_env.register_function("count_purchases_per_minute", count_purchases_per_minute)

# 应用实时计算逻辑
result_table = t_env.from_path("purchase_events") \
    .group_by("user_id, TUMBLE(purchase_time, INTERVAL '1' MINUTE)") \
    .select("user_id, TUMBLE_START(purchase_time, INTERVAL '1' MINUTE) as window_start, count_purchases_per_minute(user_id, purchase_time) as purchase_count")

# 输出结果到指定存储或展示平台
sink_ddl = """
    CREATE TABLE result_table (
        user_id INT,
        window_start TIMESTAMP(3),
        purchase_count INT
    ) WITH (
        'connector' = 'print'
    )
"""
t_env.execute_sql(sink_ddl)
t_env.insert_into("result_table", result_table)

# 执行作业
t_env.execute("Real-time Purchase Analysis")