用户行为实时分析限时特惠

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.descriptors import Kafka, Schema

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

# 配置Kafka连接
t_env.connect(Kafka()
              .version("universal")
              .topic("user_behavior")
              .start_from_latest()
              .property("zookeeper.connect", "localhost:2181")
              .property("bootstrap.servers", "localhost:9092"))
    .with_format("json")
    .with_schema(Schema()
                 .field("user_id", DataTypes.STRING())
                 .field("action", DataTypes.STRING())
                 .field("timestamp", DataTypes.TIMESTAMP()))
    .create_temporary_table("user_behavior_source")

# 实时分析用户行为
table = t_env.from_path("user_behavior_source")
result = table.group_by("user_id").select("user_id, action.count as action_count")

# 输出结果到控制台
result.execute().print()