Henry's xyz

外观

Kafka Streams ​

Kafka Streams 是 Kafka 官方的流处理库，作为一个 Java 库直接嵌入应用，无需独立的流处理集群（与 Flink、Spark Streaming 不同）。

架构 ​

               Kafka Cluster
          ┌─────────────────────┐
          │ input-topic          │
          │  (原始数据)           │
          └────────┬────────────┘
                   │
          ┌────────▼────────────┐
          │   Kafka Streams App │
          │  ┌───────────────┐  │
          │  │ Stream DSL    │  │
          │  │ Processor API │  │
          │  └───────────────┘  │
          │   (State Store)     │
          └────────┬────────────┘
                   │
          ┌────────▼────────────┐
          │ output-topic         │
          │  (处理结果)          │
          └─────────────────────┘
特性说明
无外部依赖仅依赖 Kafka 集群
有状态/无状态本地 RocksDB State Store + 可恢复（Changelog Topic）
Exactly-once天然支持 EOS 语义
弹性伸缩应用实例增减自动 Rebalance
KSQL流式 SQL 引擎，对标 Flink SQL

Stream DSL 示例 ​

java
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

StreamsBuilder builder = new StreamsBuilder();

// 数据流: input-topic → 过滤 → 转换 → 分组 → 聚合 → output-topic
KStream<String, String> source = builder.stream("raw-orders");

source
    .filter((key, orderJson) -> isValidOrder(orderJson))     // 过滤无效订单
    .mapValues(orderJson -> extractOrderAmount(orderJson))    // 提取金额
    .groupBy((key, amount) -> getRegion(amount),               // 按区域分组
             Grouped.with(Serdes.String(), Serdes.Double()))
    .aggregate(
        () -> 0.0,
        (region, amount, total) -> total + amount,             // 累加金额
        Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("region-sales")
            .withValueSerde(Serdes.Double()))
    .toStream()
    .to("region-sales-output", Produced.with(Serdes.String(), Serdes.Double()));

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

Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

KSQL ​

KSQL 是 Kafka Streams 的 SQL 引擎，无需写 Java 代码即可对流数据进行查询和处理。

sql
-- 创建 Stream（无界数据流）
CREATE STREAM orders (
    order_id VARCHAR,
    user_id VARCHAR,
    amount DOUBLE,
    region VARCHAR
) WITH (
    KAFKA_TOPIC = 'orders',
    VALUE_FORMAT = 'JSON'
);

-- 过滤
CREATE STREAM high_value_orders AS
    SELECT * FROM orders WHERE amount > 1000;

-- 聚合（窗口计算）
CREATE TABLE region_sales AS
    SELECT region,
           SUM(amount) AS total_sales,
           COUNT(*) AS order_count
    FROM orders
    WINDOW TUMBLING (SIZE 1 MINUTE)
    GROUP BY region;

-- 查询
SELECT * FROM high_value_orders EMIT CHANGES;
SELECT * FROM region_sales EMIT CHANGES;
Stream 与 Table 对比StreamTable
语义追加日志（事件流）变更日志（当前状态）
查询每次返回新事件返回当前最新值
存储不持久化本地 State Store + Changelog Topic

有状态操作 ​

操作说明状态存储
aggregate聚合（Sum/Count/Reduce）累加值
join流-流 或 流-表 Join窗口数据
count计数计数值
reduce归约归约结果
java
// Windowed Join — 按时间窗口 Join 两个 Stream
KStream<String, Order> orders = builder.stream("orders");
KStream<String, Payment> payments = builder.stream("payments");

orders.join(payments,
    (order, payment) -> new EnrichedOrder(order, payment),
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
    StreamJoined.with(Serdes.String(), orderSerde, paymentSerde))
    .to("enriched-orders");

核心配置 ​

参数说明
APPLICATION_ID_CONFIG应用 ID，同 Group 共享 State Store
NUM_STREAM_THREADS_CONFIG线程数，匹配 Partition 数
STATE_DIR_CONFIG本地状态存储目录
CACHE_MAX_BYTES_BUFFERING_CONFIG状态缓存大小
PROCESSING_GUARANTEE_CONFIGexactly_once_v2（推荐）或 at_least_once
DEFAULT_KEY_SERDE_CLASS_CONFIG默认 Key 序列化器
DEFAULT_VALUE_SERDE_CLASS_CONFIG默认 Value 序列化器

参考 ​