流计算双11活动

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,
        item_id BIGINT,
        category_id INT,
        behavior STRING,
        ts TIMESTAMP(3)
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user_behavior',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
"""
t_env.execute_sql(source_ddl)

# 定义UDF
@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.STRING())
def process_behavior(behavior):
    # 处理逻辑
    return behavior.upper()

# 应用UDF
t_env.register_function("process_behavior", process_behavior)
result_table = t_env.sql_query("""
    SELECT user_id, item_id, category_id, process_behavior(behavior) AS processed_behavior, ts
    FROM user_behavior
""")

# 输出结果
sink_ddl = """
    CREATE TABLE result (
        user_id BIGINT,
        item_id BIGINT,
        category_id INT,
        processed_behavior STRING,
        ts TIMESTAMP(3)
    ) WITH (
        'connector' = 'print'
    )
"""
t_env.execute_sql(sink_ddl)
result_table.execute_insert("result").wait()