Pathway Kafka 连接器实战：用 pw.io.kafka 构建实时数据流式 ETL 管线

Pathway Live Data Framework 通过 pw.io.kafka 模块提供了 Kafka 输入/输出连接器，使流处理管线能够直接从 Kafka topic 消费消息并将计算结果写回 Kafka。本文基于官方教程文档与当前仓库源码，完整讲解 pw.io.kafka.read 与 pw.io.kafka.write 的全部参数、消息格式约定、端到端示例（含消息生成脚本），并结合 Rust 侧连接器实现剖析提交（commit）机制、消费组订阅与背压处理的底层原理。

模块概览：两个连接器入口

Pathway 提供的 Kafka 连接器位于 python/pathway/io/kafka/init.py，对外暴露三个核心 API：

• pw.io.kafka.read：从指定 topic 读取消息，返回一张随消息更新而增量演化的 Table；
• pw.io.kafka.simple_read：read 的简化封装，只需提供服务器地址与 topic 名即可；
• pw.io.kafka.write：将 Table 上的更新写入指定的 Kafka topic。

原始教程（docs/2.developers/4.user-guide/20.connect/99.connectors/30.kafka_connectors.md）说明 Kafka 连接器仅支持流式（streaming）模式；而在当前版本的源码中，read 进一步支持了 mode 参数，可取值 "streaming"（默认，持续等待新消息）或 "static"（只读取执行时刻已存在的数据），详见下文输入连接器一节。

此外仓库还提供了 Redpanda 连接器 pw.io.redpanda.read / pw.io.redpanda.write（python/pathway/io/redpanda/init.py）。从源码结构看，它直接复用 Kafka 的底层实现（from pathway.io import kafka），用法与 Kafka 完全相同，只需将 kafka 替换为 redpanda。

最短示例：实时求和场景

考虑一个典型场景：消息被发送到 Kafka 实例的 topic connector_example，每条消息是一行包含单列 value（CSV 格式）的数据；我们希望实时计算这些值的总和，并把结果流写回同一 Kafka 实例的 sum topic。完整脚本 realtime_sum.py 如下：

import pathway as pw

# Kafka settings
rdkafka_settings = {
    "bootstrap.servers": "server-address:9092",
    "security.protocol": "sasl_ssl",
    "sasl.mechanism": "SCRAM-SHA-256",
    "group.id": "$GROUP_NAME",
    "session.timeout.ms": "6000",
    "sasl.username": "username",
    "sasl.password": "********",
}

# We define a schema for the table
# It set all the columns and their types
class InputSchema(pw.Schema):
  value: int


# We use the Kafka connector to listen to the "connector_example" topic
t = pw.io.kafka.read(
    rdkafka_settings,
    topic="connector_example",
    schema=InputSchema,
    format="csv",
    autocommit_duration_ms=1000
)

# We compute the sum (this part is independent of the connectors).
t = t.reduce(sum=pw.reducers.sum(t.value))

# We use the Kafka connector to send the resulting output stream containing the sum
pw.io.kafka.write(t, rdkafka_settings, topic_name="sum", format="json")

# We launch the computation.
pw.run()

三个要点：

1. rdkafka_settings 遵循 librdkafka 配置格式：bootstrap.servers 指定 broker 地址，security.protocol / sasl.mechanism / sasl.username / sasl.password 配置 SASL/SSL 认证，group.id 指定消费者组。
2. 中间计算与连接器无关：t.reduce(sum=pw.reducers.sum(t.value)) 是标准的 Pathway 表操作，换成任何聚合逻辑都不影响连接器用法。
3. pw.run() 不可省略：调用后计算将永久运行直至进程被终止；没有它，整个管线不会启动。

输入连接器 pw.io.kafka.read 详解

数据流语义

Kafka 输入连接器把 topic 上收到的消息序列建模为一张流式表：一次更新（update）是一组消息，更新由一次提交（commit）触发。提交保证了每次更新的原子性，并按周期生成——周期即 autocommit_duration_ms 参数。

参数说明

当前版本 read 的完整签名（见 python/pathway/io/kafka/init.py）：

参数	说明	默认值
rdkafka_settings	librdkafka 格式的连接配置 dict	必填
topic	监听的 topic 名称（当前版本仅使用列表中的第一个元素）	必填
format	消息格式："raw"、"plaintext"、"json"	"raw"
schema	结果的表 Schema（列名、类型、主键定义）；非 raw 格式时必填	None
mode	"streaming" 持续消费新消息；"static" 只读取执行时刻已有数据	"streaming"
autocommit_duration_ms	两次提交之间的最大时间（毫秒），连接器在此周期内收到的更新将被提交并推入计算图	1500
schema_registry_settings	连接 Confluent Schema Registry 的设置	None
json_field_paths	JSON 格式下，将 schema 字段映射到 payload 中的 JSON Pointer 路径（RFC 6901）	None
autogenerate_key	为 True 时自动为主键生成唯一值；否则优先使用消息自身的 key（仅 raw/plaintext 格式生效）	False
with_metadata	为 True 时追加 _metadata 列（JSON 字段），包含 timestamp_millis、topic、partition、offset 及消息 headers 数组	False
start_from_timestamp_ms	从指定的历史时间戳（Unix 毫秒）开始读取	None
parallel_readers	并行读取器副本数；未指定时取 min{引擎线程数, 分区总数}，且不会超过引擎线程数	None
name	连接器的唯一名称，用于日志、监控面板，以及持久化开启时的进度快照命名	None
max_backlog_size	限制任意时刻从源读取并保留在处理中的条目数上限；达到上限时暂停读取，处理完成后恢复。适合应对大源初始数据洪峰、避免内存尖峰	None
debug_data	调试模式下替代真实数据的静态数据	None

版本差异提示：2023 年原始教程中 format 支持 raw、csv、json 三种取值；当前版本签名已演进为 raw、plaintext、json。其中 raw 将 key 与 payload 以原始字节形式落入 key / data 两列，plaintext 则将二者从 UTF-8 解析为文本字符串后落入同名两列。若消息为 CSV，可先用 plaintext 读入 data 列再做表内解析，或直接使用 json 格式。

参数校验与常见陷阱

源码在入口处做了严格的参数校验（python/pathway/io/kafka/init.py）：

• rdkafka_settings 必须包含非空的 bootstrap.servers（消费者靠它定位 broker）；
• 必须包含非空的 group.id——因为 Pathway 的 Kafka 读取器使用 subscribe 而非 assign，而 librdkafka 要求 subscribe 必须配置消费者组 id，否则直接抛错。这一点在测试用例 test_kafka_read_without_group_id_raises_clear_error（python/pathway/tests/test_io.py）中得到了验证；
• autocommit_duration_ms 与 max_backlog_size 必须为正数，start_from_timestamp_ms 必须非负，parallel_readers 必须为正；
• 同时设置 start_from_timestamp_ms 且用户提供的 auto.offset.reset 不是 “从头开始” 语义（earliest/beginning/smallest）时，会发出警告：该值会被改写为 earliest，以便时间戳 seek 可以回退到分区起点。

原始教程还特别强调了 CSV 格式的头部消息约定：第一条消息必须是以逗号分隔、顺序正确的列名头部；缺少它连接器无法正常工作，但它只能发送一次——如果发送两次，第二条会被当作普通数据行处理。这条约定在使用 dsv/csv 类消息源时仍然值得牢记。

简化入口 simple_read

pw.io.kafka.simple_read(server, topic, ...)（python/pathway/io/kafka/init.py）只要求服务器地址与 topic 名，内部自动构造 bootstrap.servers、随机 group.id 和 auto.offset.reset（默认 beginning，即从 topic 开头读起）。若设置 read_only_new=True，则从分区末尾开始，只读程序启动后新出现的消息。需要认证或细粒度调参时应直接使用 read——simple_read 不接受 rdkafka_settings 参数。

输入连接器的 Rust 侧实现原理

Python 层之下，Kafka 读取器由 src/connectors/data_storage/kafka.rs 中的 KafkaReader 实现，几个关键设计值得关注：

1. 两种模式下的分区获取策略不同。 源码注释（kafka.rs）说明：流式模式调用 consumer.subscribe([topic])，由消费组在各 worker 之间自动再平衡并持久化已提交 offset 用于故障恢复；静态模式则手动 assign，按 partition % reader_count == worker_index 把分区均匀分片给各并行读取器，直接和分区 leader 通信，不依赖 group coordinator——避免了新消费组必须完成 JoinGroup/SyncGroup 往返后才能首次拉取数据的竞态。

2. 背压时暂停分区而非丢弃消息。 keep_alive 方法（kafka.rs）在下游阻塞、消息无法推进时暂停已分配分区（librdkafka 会清空预取缓冲区，暂停期间不积累内存），但仍以零超时 poll：这不返回数据，却会重置 max.poll.interval 计时器并服务再平衡回调——否则停顿超过 max.poll.interval.ms 消费者就会被踢出消费组，重新加入后将从组提交 offset 处重投（产生重复）。当停顿期间发生再平衡而漏出一条新分区的消息时，它会被暂存到 pending_messages 并在恢复时先于后续消息投递，确保不丢失。

3. 时间戳起点的懒式 seek。 设置 start_from_timestamp_ms 后，seek_positions_for_timestamp 通过 offsets_for_times 计算各分区起点，但不立即 seek（seek 只对已分配分区有效，且不能绕过消费组的自动分配），而是记录目标 offset，等消费者真正收到该分区第一条消息时再执行 seek（kafka.rs）。若某分区的目标位置已在末尾之后，该分区在静态模式下不产出任何行，流式模式只读起点之后的新消息。

4. 启动期元数据探测的容错重试。 新建 topic 在集群选主/元数据传播期间可能瞬时返回 NotLeaderForPartition、LeaderNotAvailable、UnknownTopicOrPartition 等错误；total_partitions_for_topic 与 partition_watermarks（kafka.rs）对此类瞬时错误以 200ms 退避重试，最长 30 秒，超时才判定为 TopicNotFound 等启动错误。此外 max_allowed_consecutive_errors 设为 32，即允许连续 32 次读错误后才放弃。

输出连接器 pw.io.kafka.write 详解

pw.io.kafka.write 把表 t 上的每一次更新发送到 Kafka 实例的单个 topic。当前版本的完整参数（python/pathway/io/kafka/init.py）：

参数	说明	默认值
table	要发送到 Kafka 的表	必填
rdkafka_settings	librdkafka 格式的连接配置，必须含非空 bootstrap.servers	必填
topic_name	目标 topic；也接受一个字符串列的引用，此时每条消息写入该列值对应的 topic（动态路由）	必填
format	序列化格式："json"、"plaintext"、"raw"、"dsv"（分隔符值格式，是 CSV 的推广）	"json"
delimiter	dsv 格式下的字段分隔符	","
key	指定哪一列作为消息 key；留空则使用内部主键	None
value	raw/plaintext 格式下指定哪一列作为消息体（str 对应 plaintext，binary 对应 raw）；表只有一列时可自动推断	None
headers	指定哪些列作为 Kafka headers 转发（UTF-8 字符串；binary 列原样输出）	None
schema_registry_settings / subject	Confluent Schema Registry 连接设置与 subject 名	None
name	连接器唯一名称，用于日志与监控	None
sort_by	在每个 mini-batch 内按给定列升序排序输出；多列按元组字典序比较	None

最简用法即教程中的示例：

pw.io.kafka.write(t, rdkafka_settings, topic_name="sum", format="json")

两个实现层面的细节：

• 消息自带逻辑时间戳头：根据 docstring，产生的消息除 key 与 value 外，还会附带 pathway_time（条目的逻辑时间）和 pathway_diff（1 或 -1，表示插入/删除语义）两个 header，均以 UTF-8 编码——下游消费者可据此还原表的增删更新语义；
• 生产者队列满时重试而非失败：Rust 侧 KafkaWriter::write（kafka.rs）在 producer.send 返回 QueueFull 时以 10ms 间隔 poll 并重新提交同一条记录，形成自旋重试；writer 的 retriable() 返回 true，连接级错误同样可重试。Drop 时执行 producer.flush(None)，保证退出前所有缓冲消息落盘。

原始教程中输出格式当时支持 binary、json、dsv；当前版本的 raw 对应了原 binary 语义（表须恰含一个 binary 列，或显式用 value 指定目标 binary 列）。

端到端示例：消息生成脚本

与 realtime_sum.py 配对的 generate_stream.py 使用 Kafka 官方的 KafkaProducer API 向 topic 写入数据：

from kafka import KafkaProducer
import time

topic = "connector_example"

producer = KafkaProducer(
    bootstrap_servers=["server-address:9092"],
    sasl_mechanism="SCRAM-SHA-256",
    security_protocol="SASL_SSL",
    sasl_plain_username="username",
    sasl_plain_password="********",
)
producer.send(topic, ("value").encode("utf-8"), partition=0)

time.sleep(5)
for i in range(10):
    time.sleep(1)
    producer.send(
        topic, (str(i)).encode("utf-8"), partition=0
    )

producer.close()

注意第一行 ("value") 即 CSV 头消息（列名），随后按每秒一条的节奏发送 0 到 9 的数据行。教程还提醒：取决于 Kafka 版本，可能需要显式指定 API 版本才能让上述代码工作：

producer = KafkaProducer(
    bootstrap_servers=["server-address:9092"],
    sasl_mechanism="SCRAM-SHA-256",
    security_protocol="SASL_SSL",
    sasl_plain_username="username",
    sasl_plain_password="********",
    api_version=(0,10,2),
)

运行顺序：先启动 realtime_sum.py（其内部 pw.run() 会常驻），再运行 generate_stream.py；随后即可观察到 sum topic 中的 JSON 消息随累计值逐步更新。

Redpanda 与其他消息队列

如开头所述，pw.io.redpanda.read / pw.io.redpanda.write 的签名与参数同 Kafka 版本一一对应（python/pathway/io/redpanda/init.py），实现上直接复用 pathway.io.kafka 模块，因此本文所有参数说明与实现分析同样适用于 Redpanda 连接器。仓库的集成测试目录 integration_tests/kafka/ 还覆盖了一组消息队列场景（含 test_backpressure.py 背压测试、test_simple.py 等），可作为各队列连接器行为的参照。

小结

Pathway 的 Kafka 连接器以 librdkafka 为底层、以消费组订阅为流式默认策略，用 autocommit_duration_ms 把离散消息聚合成原子更新，配合 reduce 等操作即可组成“Kafka 进、Kafka 出”的完整实时 ETL 管线；start_from_timestamp_ms、parallel_readers、max_backlog_size 等参数则提供了回放、并行度与内存保护的控制手段。深入阅读 src/connectors/data_storage/kafka.rs 可以看到暂停分区保活、懒式 seek、分区分片等工程细节，这些机制共同保证了管线在下游阻塞与集群再平衡场景下的正确性。