11.11用户行为实时分析有活动吗

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_topic")
             .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")

# 实时分析逻辑
result_table = t_env.from_path("user_behavior_source") \
    .group_by("user_id") \
    .select("user_id, action.count as action_count")

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