第07章 流处理框架¶

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# 定义Pipeline
with beam.Pipeline(options=PipelineOptions()) as p:
    # 读取数据
    sales = p | 'ReadFromText' >> beam.io.ReadFromText('sales.csv')

    # Beam管道语法：| 连接处理步骤，>> 命名步骤；每个lambda接收上一步输出（x依次为字符串→列表→字典→元组）
    # 解析数据
    parsed = sales | 'ParseCsv' >> beam.Map(lambda x: x.split(','))  # lambda匿名函数：简洁的单行函数

    # 转换数据
    transformed = parsed | 'Transform' >> beam.Map(lambda x: {
        'product_id': int(x[0]),
        'quantity': int(x[1]),
        'amount': float(x[2])
    })

    # 过滤数据
    filtered = transformed | 'Filter' >> beam.Filter(lambda x: x['amount'] > 100)

    # 聚合数据
    aggregated = (filtered
        | 'PairWithKey' >> beam.Map(lambda x: (x['product_id'], x['amount']))
        | 'Aggregate' >> beam.CombinePerKey(sum)
    )

    # 写入数据
    aggregated | 'WriteToText' >> beam.io.WriteToText('output.txt')

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings, DataTypes
from pyflink.table.schema import Schema
from pyflink.common.types import Row
from pyflink.datastream.functions import MapFunction

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

# 定义数据源
sales = env.from_collection([
    (1, 'product1', 10, 100.0),
    (2, 'product2', 20, 200.0),
    (3, 'product1', 15, 150.0)
])

# 定义转换函数
class SalesMapper(MapFunction):
    def map(self, value):
        product_id, product_name, quantity, amount = value
        return Row(product_id, product_name, quantity, amount, quantity * amount)

# 应用转换
mapped_sales = sales.map(SalesMapper())

# 转换为Table并注册
sales_table = t_env.from_data_stream(mapped_sales).alias(
    'product_id', 'product_name', 'quantity', 'amount', 'total_amount'
)
t_env.create_temporary_view('sales_view', sales_table)

# 执行查询
result = t_env.sql_query("""
    SELECT
        product_id,
        product_name,
        SUM(quantity) as total_quantity,
        SUM(amount) as total_amount,
        SUM(total_amount) as total_revenue
    FROM sales_view
    GROUP BY product_id, product_name
""")

# 输出结果
result.execute().print()