🌊 第 17 章 · Kafka Streams 与 ksqlDB 可视化

① 流-表二元性（Stream-Table Duality）

左：KStream（事件流，每条都是一个事实）；右：KTable（按 key 取最新值的当前状态）。点击「+ 加事件」感受两者关系。

② 四种窗口类型对比动画

③ KStream-KTable Join（流加维度表）

KTable 维护用户档案（按 user_id 取最新）；KStream 进来订单时用 user_id 查 KTable 加上用户名 / 国家。

④ WordCount Topology 可视化

输入一句话，观察消息在拓扑节点之间流动。

(等待输入…)

⑤ KStream / KTable / GlobalKTable 选型

💡 KStream-KTable Join 要求两边同 key 同分区；KStream-GlobalKTable Join 不需要 co-partitioning，可按任意外键查找。

⑥ 三种 Join 对比

// KStream-KTable join 示例（不需要 window）
KStream<String, Order> orders = builder.stream("orders");
KTable<String, User>  users  = builder.table("users");
orders.join(users, (order, user) -> enrich(order, user))
      .to("orders.enriched");
  

⑦ 三种时间语义

// 自定义 Event Time 抽取
public class JsonTsExtractor implements TimestampExtractor {
    public long extract(ConsumerRecord<Object,Object> r, long partitionTime) {
        long ts = parseJson((String) r.value()).getLong("event_ts");
        return ts > 0 ? ts : partitionTime;   // 兜底：用上一次的时间
    }
}
props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, JsonTsExtractor.class);
  

💡 配置 grace.ms 接受迟到事件，但越长状态越大；金融场景常 grace=5min，普通分析 grace=10s。

⑧ State Store + Changelog Topic（容灾）

RocksDB 在本地磁盘存状态；每次 put 同步写入 changelog topic（compacted）。Task 迁移到新 Worker 时，从 changelog 重放重建本地 store。

[Task-0] -- put(k=alice, v=10) --> RocksDB (本地)
                                |
                                +--> producer.send(changelog-topic, k=alice, v=10)

崩溃后新 Worker 接手 Task-0：
  1. 创建空 RocksDB
  2. 从 changelog topic 头到尾 replay → 状态完全恢复
  3. 切换为正常处理消息

⚠ 优化：
  - changelog 是 compacted topic（log.cleanup.policy=compact），同 key 只保留最新
  - num.standby.replicas=1：另一台 Worker 实时跟随 changelog → 故障接管秒级
  - 关闭：StoreBuilder.withLoggingDisabled() → 只本地、不容灾（仅适合纯衍生数据）
  

⑨ EOS v2 在 Streams 中的开启

props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
          StreamsConfig.EXACTLY_ONCE_V2);   // ← 一行打开
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);   // 内部 topic ≥ 3
  

⚠ 下游消费者必须 isolation.level=read_committed，否则会看到 abort 的中间值。

⑩ ksqlDB 常见 SQL 速查

-- 创建 STREAM（基于 Topic）
CREATE STREAM clicks (user_id VARCHAR, page VARCHAR, ts BIGINT)
  WITH (KAFKA_TOPIC='learn.clicks', VALUE_FORMAT='JSON', TIMESTAMP='ts');

-- 创建 TABLE（按 key 取最新）
CREATE TABLE users (user_id VARCHAR PRIMARY KEY, name VARCHAR, country VARCHAR)
  WITH (KAFKA_TOPIC='learn.users', VALUE_FORMAT='JSON');

-- 滚动窗口聚合：每分钟每用户 PV
CREATE TABLE pv_per_min AS
  SELECT user_id, COUNT(*) AS pv
  FROM clicks
  WINDOW TUMBLING (SIZE 1 MINUTE)
  GROUP BY user_id
  EMIT CHANGES;

-- Stream-Table Join（流加维度）
CREATE STREAM enriched AS
  SELECT c.user_id, u.name, u.country, c.page
  FROM clicks c LEFT JOIN users u ON c.user_id = u.user_id
  EMIT CHANGES;

-- Pull Query（点查 TABLE 当前值）
SELECT * FROM users WHERE user_id = 'u-1';

-- 持续查询管理
SHOW QUERIES;
TERMINATE CSAS_PV_PER_MIN_3;
DROP TABLE pv_per_min DELETE TOPIC;
  

⑪ 常见坑速查

⑫ Streams / Flink / Spark Streaming 选型

💡 选型口诀：
· 处理 100% 在 Kafka 内部、想简单部署 → Kafka Streams / ksqlDB
· 多源 + 复杂窗口 + 大状态 + 金融级 EOS → Flink
· 已有 Spark 离线团队、需要批流统一 → Spark Structured Streaming

⑬ Streams 关键调优参数