第06章 实时数据处理¶

from kafka import KafkaProducer, KafkaConsumer
import json

# 生产者
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')  # lambda匿名函数：简洁的单行函数
)

# 发送消息
producer.send('sales', value={'product_id': 1, 'quantity': 10, 'amount': 100.0})
producer.flush()

# 消费者
consumer = KafkaConsumer(
    'sales',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),  # json.loads将JSON字符串转为Python对象
    auto_offset_reset='earliest'
)

# 接收消息
for message in consumer:
    print(f"Received: {message.value}")
    # 处理消息
    process_message(message.value)

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, DataTypes
from pyflink.table.expressions import col

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

# 定义数据源
t_env.execute_sql("""
    CREATE TABLE sales (
        product_id INT,
        quantity INT,
        amount DECIMAL(10,2),
        sale_time TIMESTAMP(3),
        WATERMARK FOR sale_time AS sale_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'sales',
        'properties.bootstrap.servers' = 'localhost:9092',
        'properties.group.id' = 'sales-group',
        'format' = 'json'
    )
""")

# 实时聚合
result = t_env.sql_query("""
    SELECT
        product_id,
        COUNT(*) as order_count,
        SUM(quantity) as total_quantity,
        SUM(amount) as total_amount
    FROM sales
    GROUP BY
        product_id,
        TUMBLE(sale_time, INTERVAL '1' MINUTE)
""")

# 定义输出表
t_env.execute_sql("""
    CREATE TABLE sales_aggregate (
        product_id INT,
        order_count BIGINT,
        total_quantity BIGINT,
        total_amount DECIMAL(10,2)
    ) WITH (
        'connector' = 'print'
    )
""")

# 将结果写入输出表
result.execute_insert('sales_aggregate').wait()

from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col
from pyspark.sql.types import StructType, IntegerType, TimestampType, DecimalType

# 创建SparkSession
spark = SparkSession.builder \
    .appName("SparkStreamingExample") \
    .getOrCreate()

# 读取Kafka数据
df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "sales") \
    .load()

# 解析数据
from pyspark.sql.functions import from_json, to_json

schema = StructType().add("product_id", IntegerType()) \
    .add("quantity", IntegerType()) \
    .add("amount", DecimalType(10,2)) \
    .add("sale_time", TimestampType())

df = df.select(
    from_json(col("value").cast("string"), schema).alias("data")
).select(
    col("data.product_id"),
    col("data.quantity"),
    col("data.amount"),
    col("data.sale_time")
)

# 实时聚合
from pyspark.sql.functions import sum as _sum, count as _count

aggregated_df = df.groupBy(
    window(col("sale_time"), "1 minute")
).agg(
    _count("*").alias("order_count"),
    _sum("quantity").alias("total_quantity"),
    _sum("amount").alias("total_amount")
)

# 输出结果
query = aggregated_df \
    .writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

# 等待流结束
query.awaitTermination(timeout=300)