流式计算 新购优惠

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

# 初始化环境
env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

# 定义数据源
source_ddl = """
    CREATE TABLE user_behavior (
        user_id BIGINT,
        action STRING,
        event_time TIMESTAMP(3)
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user_behavior_topic',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
"""
t_env.execute_sql(source_ddl)

# 定义UDF来检查是否满足优惠条件
@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.BOOLEAN())
def check_discount_condition(action):
    return action == "purchase"

t_env.register_function("check_discount_condition", check_discount_condition)

# 应用UDF并定义输出目标
query = """
    SELECT user_id, event_time
    FROM user_behavior
    WHERE check_discount_condition(action)
"""
result_table = t_env.sql_query(query)

# 输出结果到控制台
sink_ddl = """
    CREATE TABLE discount_events (
        user_id BIGINT,
        event_time TIMESTAMP(3)
    ) WITH (
        'connector' = 'print'
    )
"""
t_env.execute_sql(sink_ddl)
result_table.execute_insert("discount_events").wait()

# 执行作业
env.execute("Real-time Discount Promotion")