系统设计（七）：数据管道——批处理、流处理与 Lambda 架构

1
2
3
4
5
6

源系统 → 抽取 → 转换（暂存区） → 加载 → 数据仓库

示例：
  MySQL（订单） ─┐
  PostgreSQL（用户） ─┤→ 转换（清洗、关联、聚合） → 加载 → Snowflake
  MongoDB（日志） ─┘

1
2
3
4
5
6

源系统 → 抽取 → 加载（原始） → 转换（在仓库内） → 数仓分层/视图

示例：
  MySQL（订单） ─┐
  PostgreSQL（用户） ─┤→ 原始加载 → BigQuery → dbt模型 → 清洗后的表
  MongoDB（日志） ─┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 概念化MapReduce：统计各URL的页面浏览次数

# Map阶段（在多台机器上并行执行）
def map_function(log_line):
    """解析日志行，为每次页面浏览输出 (url, 1)"""
    url = parse_url(log_line)
    emit(url, 1)

# Shuffle阶段（框架自动按键分组）
# "/products/123" → [1, 1, 1, 1, 1]
# "/products/456" → [1, 1, 1]

# Reduce阶段（对每个Key的Values求和）
def reduce_function(url, counts):
    """对每个URL的所有计数求和"""
    emit(url, sum(counts))

# 输出：
# "/products/123" → 5
# "/products/456" → 3

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .appName("DailyOrderAnalytics") \
    .config("spark.sql.shuffle.partitions", 200) \
    .getOrCreate()

# 从数据湖读取原始订单数据
orders = spark.read.parquet("s3://data-lake/raw/orders/date=2025-07-24/")
users = spark.read.parquet("s3://data-lake/raw/users/")
products = spark.read.parquet("s3://data-lake/raw/products/")

# 转换：关联订单、用户与商品信息
enriched_orders = orders \
    .join(users, orders.user_id == users.id, "left") \
    .join(products, orders.product_id == products.id, "left") \
    .select(
        orders.order_id,
        orders.created_at,
        users.country,
        users.age_group,
        products.category,
        products.name.alias("product_name"),
        orders.quantity,
        orders.total_price,
    )

# 聚合：按国家与商品类目统计日销售额
daily_sales = enriched_orders \
    .groupBy("country", "category") \
    .agg(
        F.count("order_id").alias("order_count"),
        F.sum("total_price").alias("total_revenue"),
        F.avg("total_price").alias("avg_order_value"),
        F.countDistinct("user_id").alias("unique_buyers"),
    ) \
    .orderBy(F.desc("total_revenue"))

# 写入数据湖（分国家分区）
daily_sales.write \
    .mode("overwrite") \
    .partitionBy("country") \
    .parquet("s3://data-lake/curated/daily_sales/date=2025-07-24/")

# 同时写入报表数据库
daily_sales.write \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://reporting-db:5432/analytics") \
    .option("dbtable", "daily_sales") \
    .option("user", "etl_user") \
    .mode("append") \
    .save()

1
2
3

时间：    |----1分钟----|----1分钟----|----1分钟----|
事件：    [e1, e2, e3] [e4, e5]    [e6, e7, e8, e9]
窗口：    [  窗口 1   ] [ 窗口 2  ] [   窗口 3    ]

1
2
3
4
5
6

时间：    |----1分钟-----------|
         |     |----1分钟----|------|
事件：    [e1, e2, e3, e4, e5, e6, e7]
窗口：    [  窗口 1（0–60秒）      ]
              [  窗口 2（30–90秒）     ]
                   [  窗口 3（60–120秒）    ]

1
2
3

事件：    e1..e2..e3........e4..e5..e6............e7..e8
         |--- 会话 1 ---|--- 会话 2 ------|--会话 3--|
         （间隙 < 阈值）   （间隙 < 阈值）     （间隙 > 阈值）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

水位线 W=100 表示：“所有时间戳 ≤ 100 的事件均已到达。”

若当前水位线为 W=100：
  - 窗口 [0, 60] 已完成（所有事件均已到达）
  - 窗口 [60, 120] 仍可能收到新事件
  
迟到事件（时间戳 < 水位线）可被：
  - 丢弃（最简单）
  - 发送到旁路输出（side output）供后续修正
  - 用于更新先前结果（retractions）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

# 概念化Flink电商实时分析管道
# 使用PyFlink Table API

from pyflink.table import EnvironmentSettings, TableEnvironment
from pyflink.table.expressions import col, lit
from pyflink.table.window import Tumble

# 初始化环境
env_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(env_settings)

# 定义Kafka源表
t_env.execute_sql("""
    CREATE TABLE page_views (
        user_id STRING,
        page_url STRING,
        event_time TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'page-views',
        'properties.bootstrap.servers' = 'kafka:9092',
        'properties.group.id' = 'analytics-pipeline',
        'format' = 'json',
        'scan.startup.mode' = 'latest-offset'
    )
""")

# 定义输出目标表
t_env.execute_sql("""
    CREATE TABLE page_view_counts (
        window_start TIMESTAMP(3),
        window_end TIMESTAMP(3),
        page_url STRING,
        view_count BIGINT,
        unique_users BIGINT
    ) WITH (
        'connector' = 'jdbc',
        'url' = 'jdbc:postgresql://analytics-db:5432/metrics',
        'table-name' = 'page_view_counts',
        'driver' = 'org.postgresql.Driver',
        'username' = 'flink_user',
        'password' = '...'
    )
""")

# 滚动窗口聚合：统计每分钟各URL的浏览量与独立用户数
t_env.execute_sql("""
    INSERT INTO page_view_counts
    SELECT
        window_start,
        window_end,
        page_url,
        COUNT(*) AS view_count,
        COUNT(DISTINCT user_id) AS unique_users
    FROM TABLE(
        TUMBLE(TABLE page_views, DESCRIPTOR(event_time), INTERVAL '1' MINUTE)
    )
    GROUP BY window_start, window_end, page_url
""")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

                    ┌──────────────┐
全部数据 ──────────→│  批处理层    │ → 批处理视图 ─────┐
     │              │  （Spark）   │                     │
     │              └──────────────┘                     ▼
     │                                              ┌──────────┐
     │                                              │ 服务层    │ → 查询请求
     │                                              │           │
     │              ┌──────────────┐                └──────────┘
     └─────────────→│  速度层      │ → 实时视图 ─┘
                    │  （Flink）   │
                    └──────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

批处理层（每小时运行）：
  - 从数据湖读取全部页面浏览事件
  - 统计各URL的历史总浏览量
  - 写入 batch_view 表：{url, total_views, last_updated}

速度层（持续运行）：
  - 从Kafka读取自上次批处理以来的浏览事件
  - 实时统计各URL的浏览量（最近一小时）
  - 写入 realtime_view 表：{url, recent_views, last_updated}

服务层（查询时）：
  SELECT
    batch.url,
    batch.total_views + COALESCE(realtime.recent_views, 0) AS total_views
  FROM batch_view batch
  LEFT JOIN realtime_view realtime ON batch.url = realtime.url

1
2
3
4
5
6
7
8

全部数据 → Kafka（保留数月/数年） → 流处理器 → 服务层

历史数据重处理流程：
  1. 部署新版流处理器
  2. 从Kafka Topic起始位置（offset 0）开始消费
  3. 将全部历史事件经由新逻辑处理
  4. 切换服务层至新输出
  5. 下线旧版处理器

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

from pydantic import BaseModel, validator
from typing import Optional
from datetime import datetime

class OrderEvent(BaseModel):
    order_id: str
    user_id: str
    product_id: str
    quantity: int
    total_price: float
    currency: str
    timestamp: datetime

    @validator("quantity")
    def quantity_must_be_positive(cls, v):
        if v <= 0:
            raise ValueError("quantity must be positive")
        return v

    @validator("currency")
    def currency_must_be_valid(cls, v):
        valid = {"USD", "EUR", "GBP", "JPY", "CNY"}
        if v not in valid:
            raise ValueError(f"currency must be one of {valid}")
        return v

def validate_event(raw_event: dict) -> Optional[OrderEvent]:
    try:
        return OrderEvent(**raw_event)
    except Exception as e:
        # 发送至死信主题（DLQ）供人工排查
        send_to_dlq(raw_event, str(e))
        return None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 示例血缘元数据（随处理后数据一同存储）
lineage:
  source:
    system: "orders-service"
    topic: "orders"
    partition: 3
    offset: 145892
    timestamp: "2025-07-24T10:30:00Z"
  transformations:
    - step: "schema_validation"
      version: "1.2.0"
      timestamp: "2025-07-24T10:30:01Z"
    - step: "currency_conversion"
      version: "2.0.1"
      rates_source: "ecb_daily_2025-07-24"
      timestamp: "2025-07-24T10:30:01Z"
    - step: "user_enrichment"
      version: "1.0.0"
      source: "users-service-api"
      timestamp: "2025-07-24T10:30:02Z"
  destination:
    table: "enriched_orders"
    partition: "date=2025-07-24"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# PostgreSQL的Debezium连接器配置
{
  "name": "orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "orders-db",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "...",
    "database.dbname": "orders",
    "database.server.name": "orders",
    "table.include.list": "public.orders,public.order_items",
    "plugin.name": "pgoutput",
    "publication.name": "dbz_publication",
    "slot.name": "debezium_slot",
    "topic.prefix": "cdc",
    "transforms": "route",
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.route.regex": "cdc\\.public\\.(.*)",
    "transforms.route.replacement": "cdc.$1"
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

{
  "before": null,
  "after": {
    "order_id": "ord_789",
    "user_id": "user_123",
    "total": 59.98,
    "status": "created",
    "created_at": 1721822400000
  },
  "source": {
    "version": "2.5.0",
    "connector": "postgresql",
    "name": "orders",
    "ts_ms": 1721822400123,
    "db": "orders",
    "schema": "public",
    "table": "orders",
    "txId": 12345,
    "lsn": 98765432
  },
  "op": "c",
  "ts_ms": 1721822400200
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class IdempotentProcessor:
    """通过去重存储确保每个事件仅被处理一次"""

    def __init__(self, redis_client, ttl=86400):
        self.redis = redis_client
        self.ttl = ttl  # 记录已处理事件的过期时间（秒）

    def process(self, event: dict) -> bool:
        event_id = event["event_id"]
        dedup_key = f"processed:{event_id}"

        # 原子性“检查并设置”
        if self.redis.set(dedup_key, "1", ex=self.ttl, nx=True):
            # 首次见到该事件 —— 执行处理
            return True
        else:
            # 重复事件 —— 跳过
            return False

# 在管道中使用
processor = IdempotentProcessor(redis_client)

for event in stream:
    if processor.process(event):
        transform_and_store(event)
    else:
        metrics.increment("duplicate_events_skipped")

1
2
3

Web/Mobile App → 事件采集器（HTTP API） → Kafka → Flink → PostgreSQL → Grafana仪表盘
                                           ↓
                                      S3数据湖（原始事件，用于批处理重处理）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

峰值：50,000 事件/秒  
平均事件大小：500 字节  

Kafka吞吐量：50,000 × 500 = 25 MB/秒  
日数据量：25 MB/秒 × 86,400 = 2.16 TB/天  
月存储量（Kafka，30天保留）：64.8 TB  
S3存储量（原始，压缩后）：约15 TB/月（Parquet压缩率≈70%）

Flink输出：约1,000条聚合记录/分钟  
PostgreSQL存储：可忽略（聚合数据量极小）