大数据中台与在线系统集成的技术实践

import json

from kafka import KafkaProducer

# 初始化Kafka生产者

producer = KafkaProducer(bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8'))

# 模拟用户行为日志

user_event = {

'user_id': '12345',

'action': 'click',

'timestamp': '2025-04-05T10:00:00Z'

}

# 发送消息到Kafka

producer.send('user_actions', user_event)

producer.flush()

producer.close()

from pyflink.datastream import StreamExecutionEnvironment

from pyflink.datastream.functions import MapFunction

from pyflink.common.serialization import SimpleStringEncoder

from pyflink.datastream.connectors import FlinkKafkaConsumer

from pyflink.common import WatermarkStrategy, Time

from pyflink.datastream import StreamExecutionEnvironment

from pyflink.datastream.functions import ProcessFunction

from pyflink.common import Types

from pyflink.datastream.checkpointing_mode import CheckpointingMode

from pyflink.datastream.window import TumblingEventTimeWindows

from pyflink.datastream.windows import Time

env = StreamExecutionEnvironment.get_execution_environment()

# 设置Kafka消费者

kafka_source = FlinkKafkaConsumer(

topics='user_actions',

deserialization_schema=SimpleStringEncoder(),

properties={'bootstrap.servers': 'localhost:9092', 'group.id': 'flink-group'}

)

# 添加数据源

ds = env.add_source(kafka_source)

# 解析JSON数据

class ParseEvent(MapFunction):

def map(self, value):

import json

event = json.loads(value)

return (event['user_id'], event['timestamp'])

ds.map(ParseEvent())

# 设置时间戳和水位线

ds = ds.assign_timestamps_and_watermarks(WatermarkStrategy.for_monotonic_timestamps().with_periodic_watermarks(Time.seconds(5)))

# 按用户ID分组并统计每分钟点击次数

result = ds.key_by(lambda x: x[0]) \

.window(TumblingEventTimeWindows.of(Time.minutes(1))) \

.process(ProcessFunction())

# 输出结果

result.print()

env.execute("User Action Count")

from kafka import KafkaConsumer

import json

# 初始化Kafka消费者

consumer = KafkaConsumer('user_actions', bootstrap_servers='localhost:9092', group_id='online-system')

# 消费并处理数据

for message in consumer:

event = json.loads(message.value)

print(f"Received event: {event}")

# 在这里执行在线系统的业务逻辑

from flask import Flask, jsonify

import requests

app = Flask(__name__)

# 模拟从大数据中台获取数据

def get_user_data(user_id):

response = requests.get(f'http://data-center/api/user/{user_id}')

if response.status_code == 200:

return response.json()

return {}

@app.route('/api/user/', methods=['GET'])

def get_user(user_id):

data = get_user_data(user_id)

return jsonify(data)

if __name__ == '__main__':

app.run(host='0.0.0.0', port=5000)