Kafka + Flink 管道里的 Schema 瘟疫：用判别字段把几十张表压成两张

-- 每种事件一张表，UNION ALL 拼接
CREATE TABLE order_created (
  event_id   STRING,
  user_id    STRING,
  amount     DECIMAL(10, 2),
  proctime   AS PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic'     = 'order-created',
  'format'    = 'avro',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'flink-order-created'
);

CREATE TABLE order_cancelled (
  event_id   STRING,
  user_id    STRING,
  reason     STRING,
  proctime   AS PROCTIME()
) WITH (
  'connector' = 'kafka',
  'topic'     = 'order-cancelled',
  'format'    = 'avro',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'flink-order-cancelled'
);

-- 再来 order_refunded, order_shipped ... 一路加下去

-- 查询：所有订单相关事件
SELECT 'CREATED' AS event_type, event_id, user_id, CAST(amount AS STRING) AS detail
FROM order_created
UNION ALL
SELECT 'CANCELLED' AS event_type, event_id, user_id, reason AS detail
FROM order_cancelled
-- UNION ALL ... 继续 13 路
;

-- 核心表：单一 Kafka 主题，所有事件共用
CREATE TABLE order_events_core (
  event_type  STRING,        -- 判别字段：CREATED / CANCELLED / REFUNDED ...
  event_id    STRING,
  user_id     STRING,
  ts          TIMESTAMP(3),
  payload     STRING,        -- 扩展字段 JSON payload
  proctime    AS PROCTIME(),
  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'topic'     = 'order-events-all',  -- 合并后的单一主题
  'format'    = 'avro',
  'avro.schema.literal' = '
    {
      "type": "record",
      "name": "OrderEvent",
      "fields": [
        {"name": "event_type", "type": "string"},
        {"name": "event_id",   "type": "string"},
        {"name": "user_id",    "type": "string"},
        {"name": "ts",         "type": {"type": "long", "logicalType": "timestamp-millis"}},
        {"name": "payload",    "type": "string"}
      ]
    }
  ',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'flink-order-events'
);

-- 查询：单表 + WHERE，不再需要 UNION ALL
SELECT event_id, user_id, event_type, payload
FROM order_events_core
WHERE event_type IN ('CREATED', 'CANCELLED', 'REFUNDED')
;

-- 只关心创建事件的消费者——过滤条件一加就行，新事件类型对它完全透明
SELECT event_id, user_id,
       CAST(JSON_VALUE(payload, '$.amount') AS DECIMAL(10, 2)) AS amount
FROM order_events_core
WHERE event_type = 'CREATED'
;

# merge_topics.py — 用 confluent-kafka 把多主题数据写入合并主题
# 依赖：pip install confluent-kafka

from confluent_kafka import Consumer, Producer
import json

BOOTSTRAP = "kafka:9092"
SOURCE_TOPICS = ["order-created", "order-cancelled", "order-refunded"]
TARGET_TOPIC  = "order-events-all"

# 事件类型与主题的映射
TOPIC_TO_EVENT_TYPE = {
    "order-created":   "CREATED",
    "order-cancelled": "CANCELLED",
    "order-refunded":  "REFUNDED",
}

consumer = Consumer({
    "bootstrap.servers": BOOTSTRAP,
    "group.id": "schema-merge-migrator",
    "auto.offset.reset": "earliest",
})
consumer.subscribe(SOURCE_TOPICS)

producer = Producer({"bootstrap.servers": BOOTSTRAP})

def delivery_report(err, msg):
    if err:
        print(f"Delivery failed: {err}")

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print(f"Consumer error: {msg.error()}")
        continue

    event_type = TOPIC_TO_EVENT_TYPE.get(msg.topic(), "UNKNOWN")
    original_value = json.loads(msg.value().decode("utf-8"))

    # 构造合并后的统一结构：核心字段 + payload
    merged = {
        "event_type": event_type,
        "event_id":   original_value.get("event_id", ""),
        "user_id":    original_value.get("user_id", ""),
        "ts":         original_value.get("ts", 0),
        "payload":    json.dumps(original_value),  # 原始数据整体塞进 payload
    }

    producer.produce(
        TARGET_TOPIC,
        key=merged["event_id"],
        value=json.dumps(merged).encode("utf-8"),
        callback=delivery_report,
    )
    producer.poll(0)

producer.flush()