大数据消息处理11.11促销活动

# Kafka Producer Example
from kafka import KafkaProducer
import json

producer = KafkaProducer(bootstrap_servers='localhost:9092')
data = {'user_id': 123, 'action': 'purchase', 'product_id': 456}
producer.send('user_actions', value=json.dumps(data).encode('utf-8'))
producer.flush()

# Flink Streaming Job Example
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.udf import udf

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

@udf(input_types=[DataTypes.STRING()], result_type=DataTypes.INT())
def count_purchases(json_str):
    data = json.loads(json_str)
    return 1 if data['action'] == 'purchase' else 0

t_env.register_function("count_purchases", count_purchases)

source_ddl = """
    CREATE TABLE user_actions (
        user_id INT,
        action STRING,
        product_id INT
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user_actions',
        'properties.bootstrap.servers' = 'localhost:9092',
        'format' = 'json'
    )
"""

sink_ddl = """
    CREATE TABLE purchase_counts (
        count INT
    ) WITH (
        'connector' = 'print'
    )
"""

t_env.execute_sql(source_ddl)
t_env.execute_sql(sink_ddl)

result_table = t_env.sql_query("""
    SELECT count_purchases(action) AS count
    FROM user_actions
""")

result_table.execute_insert("purchase_counts").wait()