双十二Oceanus 选购

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 user_visits (
        user_id INT,
        visit_time TIMESTAMP(3)
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user_visits_topic',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
"""
t_env.execute_sql(source_ddl)

# 定义UDF
@udf(input_types=[DataTypes.INT(), DataTypes.TIMESTAMP(3)], result_type=DataTypes.INT())
def count_visits(user_id, visit_time):
    return 1

# 注册UDF
t_env.register_function("count_visits", count_visits)

# 数据处理逻辑
result_table = t_env.sql_query("""
    SELECT 
        TUMBLE_START(visit_time, INTERVAL '1' MINUTE) AS window_start,
        COUNT(count_visits(user_id, visit_time)) AS visit_count
    FROM user_visits
    GROUP BY TUMBLE(visit_time, INTERVAL '1' MINUTE)
""")

# 输出结果
sink_ddl = """
    CREATE TABLE visit_counts (
        window_start TIMESTAMP(3),
        visit_count INT
    ) WITH (
        'connector' = 'print'
    )
"""
t_env.execute_sql(sink_ddl)
result_table.execute_insert("visit_counts").wait()