用户行为实时分析搭建

# Flask Web服务器部分
from flask import Flask, request
app = Flask(__name__)

@app.route('/track', methods=['POST'])
def track():
    data = request.json
    # 将数据发送到Kafka
    producer.send('user_behavior_topic', value=data)
    return 'Tracked!'

# Kafka生产者配置
producer = KafkaProducer(bootstrap_servers='localhost:9092')

# Flink实时计算部分
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

# 从Kafka读取数据
kafka_source = t_env.connect(Kafka()
                             .version("universal")
                             .topic("user_behavior_topic")
                             .start_from_earliest()
                             .property("zookeeper.connect", "localhost:2181")
                             .property("bootstrap.servers", "localhost:9092"))
                             .with_format("json").with_schema(schema).create_temporary_table("kafka_input")

# 数据处理逻辑
table = t_env.from_path("kafka_input")
result = table.group_by("user_id").select("user_id, count(*) as visit_count")

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