kafka

07. Kafka Streams 流处理

Kafka Streams: KStream、KTable、窗口聚合与 Exactly-Once 语义

Leeting Yan 5 分钟阅读 2011 字

Kafka Streams 是一个轻量级的流处理库，与 Kafka 深度集成，无需部署独立的流处理集群。与 Flink 或 Spark Streaming 相比，Kafka Streams 的优势在于低延迟、无额外基础设施、以及与 Kafka 生态的无缝集成。它通过 KStream 和 KTable 两个核心抽象，将事件流（Event Stream）和状态表（Changelog Stream）统一为处理模型。

1. 核心抽象：KStream 与 KTable

Kafka Streams 的编程模型建立在两个互补的抽象之上：

KStream：不可变事件流

KStream 表示一系列独立的事件记录，每条记录都是一次追加操作。即使 key 相同，也是独立的记录。

KStream<orderId, status>
  ┌─────────┬───────────┐
  │ order:1 │ CREATED   │  ← 独立记录
  │ order:1 │ PAID      │  ← 独立记录（追加）
  │ order:1 │ SHIPPED   │  ← 独立记录（追加）
  └─────────┴───────────┘
→ 处理结果包含三条记录

KTable：可变更状态流

KTable 表示一个可以更新的状态表，底层是 log-compacted topic。相同 key 的新记录会覆盖旧记录，只保留最新状态。

KTable<userId, balance>
  ┌─────────┬───────────┐
  │ user:1  │ 100.00    │  ← 初始状态
  │ user:1  │ 85.50     │  ← 更新（覆盖）
  │ user:2  │ 200.00    │  ← 新 key 插入
  └─────────┴───────────┘
→ 最终状态: user:1=85.50, user:2=200.00

KStream 与 KTable 的互转

转换方法说明
KStream → KTable.toTable()按 key 保留最新值
KTable → KStream.toStream()输出变更流（每次更新一条记录）
KStream + KTable.join()流表关联（类似 SQL 的 Stream-Table Join）
KStream<String, Order> orders = builder.stream("orders");
KTable<String, User> users = builder.table("users");

// Stream-Table Join：将订单流与用户表关联
KStream<String, OrderWithUser> enriched = orders.join(users,
    (order, user) -> new OrderWithUser(order, user));

2. Topology 与 Processor API

Kafka Streams 将流处理逻辑建模为拓扑图（Topology），由 Source、Processor、Sink 三种节点组成：

   ┌──────────┐
   │  Source   │  ← 读取 Topic "orders"
   │ (orders)  │
   └────┬─────┘
        │
   ┌────▼──────┐
   │Filter     │  ← 过滤掉 amount < 100 的订单
   │(amount>100)│
   └────┬──────┘
        │
   ┌────▼──────┐
   │ MapValues │  ← 提取订单金额
   │ (amount)  │
   └────┬──────┘
        │
   ┌────▼──────┐
   │   Sink    │  ← 写入 Topic "large-orders"
   └───────────┘

DSL 表达与 Processor API 等价：

// High-Level DSL（声明式）
orders.filter((k, v) -> v.getAmount() > 100)
      .mapValues(Order::getAmount)
      .to("large-orders");

// Low-Level Processor API（命令式，更灵活）
topology.addSource("Source", "orders")
        .addProcessor("Filter", () -> new AmountFilterProcessor(), "Source")
        .addProcessor("Extractor", () -> new AmountExtractorProcessor(), "Filter")
        .addSink("Sink", "large-orders", "Extractor");

3. 窗口聚合与时间管理

流处理中的聚合需要窗口来界定计算范围。Kafka Streams 支持四种窗口类型：

窗口类型描述特点适用场景
Tumbling固定大小、不重叠每条记录只属于一个窗口每分钟统计
Hopping固定大小、可重叠advance < size，记录可属于多个窗口滑动平均值
Session动态大小、由活动间隙定义同一 key 的 gap timeout 内属于同一会话用户行为分析
Sliding（已废弃）基于记录的窗口已被 Hopping 替代—

Tumbling Window 示例

orders.groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))  // 5 分钟窗口
    .aggregate(
        () -> 0.0,                                         // 初始值
        (key, order, total) -> total + order.getAmount(),  // 聚合逻辑
        Materialized.with(Serdes.String(), Serdes.Double())
    )
    .toStream()
    .filter((k, v) -> v > 10000)  // 只输出金额 > 10000 的窗口
    .to("high-value-windows");

Grace Period（处理乱序数据）

orders.groupByKey()
    .windowedBy(
        TimeWindows.of(Duration.ofMinutes(5))
                   .grace(Duration.ofMinutes(1))  // 允许 1 分钟的迟到数据
    )
    .aggregate(...)

4. Exactly-Once 流处理

Kafka Streams 通过"事务性消费 + 事务性生产 + 偏移量原子提交"实现端到端的 Exactly-Once：

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "fraud-detection-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
          StreamsConfig.EXACTLY_ONCE_V2);  // Kafka 2.5+

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Transaction> transactions = builder.stream("transactions");

transactions
    .filter((k, v) -> v.getAmount() > 10000)
    .to("suspicious-transactions",
        Produced.with(Serdes.String(), new TransactionSerde()));

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Exactly-Once 的实现链路：

Consumer (Read) → Process → Producer (Write)
     │                           │
     └──────── Transaction ──────┘
              (原子提交)
  • 消费偏移量作为事务的一部分写入 __consumer_offsets
  • 如果应用崩溃，未提交的事务会被回滚，消息会重新消费
  • 已提交的事务在 Broker 端去重，保证幂等性

5. State Store 与本地状态

Kafka Streams 通过 State Store（状态存储）支持有状态的流处理，如聚合、Join、去重等：

内置 State Store 类型

类型特点适用场景
PersistentKeyValueStoreRocksDB 持久化，可恢复需要容错的状态
InMemoryKeyValueStore纯内存，性能最高可重建的临时状态
TimestampedKeyValueStore带时间戳的版本需要淘汰过期数据的场景
WindowStore按窗口分区的存储窗口聚合

持久化与容错

State Store 的数据通过 changelog topic 实现容错：

Application 写入 State Store:
  State Store (RocksDB) ──→ changelog topic (compact, replicated)
                              │
  崩溃恢复时:                 │
    重建 State Store ←───────┘  (从 changelog 恢复)

Standby Replica 机制：可以在另一个实例上预加载状态存储的副本，当主实例失败时，Standby 可以立即接管，避免重建延迟。

维度Kafka StreamsApache FlinkSpark Structured Streaming
架构嵌入式库独立集群Spark 集群
延迟毫秒级毫秒级秒级（微批次）
部署无需额外集群需要 JobManager/TaskManager需要 Spark 集群
状态后端RocksDB（内置）RocksDB/Heap内存/RocksDB（通过 State Store）
窗口语义事件时间 + 处理时间事件时间（Watermark）事件时间
生态集成Kafka 深度集成多源多汇Spark 生态
适用场景Kafka 为中心的流处理复杂流处理批流统一

Kafka Streams 状态存储与调优

Kafka Streams 的内部状态使用 RocksDB 存储：

// 自定义 RocksDB 配置
Properties props = new Properties();
props.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, 
          CustomRocksDBConfig.class.getName());

public class CustomRocksDBConfig implements RocksDBConfigSetter {
    @Override
    public void setConfig(String storeName, Options options, Map<String, Object> configs) {
        // 增大 block cache（默认 50MB → 256MB）
        BlockBasedTableConfig tableConfig = new BlockBasedTableConfig();
        tableConfig.setBlockCacheSize(256 * 1024 * 1024L);
        options.setTableFormatConfig(tableConfig);
        
        // 压缩策略：大状态用 Snappy，追求查询速度用 None
        options.setCompressionType(CompressionType.SNAPPY);
    }
}
调参项默认建议影响
block_cache_size50MB物理内存 10%减少磁盘 I/O
write_buffer_size4MB16MB减少 flush 频率
compactionleveluniversal（写入密集）平衡读写

Exactly-Once 的代价：启用 EOS 后吞吐量下降约 20-30%，因需要事务协调和幂等生产者。仅在金融交易等对一致性要求极高的场景启用。

延伸阅读

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「kafka」更多文章

  1. 集群升级与滚动重启实践
  2. Kafka 应用测试策略：Testcontainers 与集成测试
  3. 压缩算法选型：lz4、zstd、snappy 与 gzip