Benthos（现已更名为 Redpanda Connect）是一个用 YAML 声明流式数据管道的工具，能在不写一行代码的情况下完成消息接入、转换与分发，是构建实时数据流的轻量方案。本文带你从零跑通一条可落地的流式数据管道。
一、为什么需要流式数据管道
在微服务与事件驱动架构里，数据常常需要在系统之间持续流动：业务库变更要同步到搜索索引、日志要清洗后写入数据仓库、IoT 设备上报要实时告警。传统做法是为每个场景写一段胶水代码，结果便是大量难以维护的脚本。流式数据管道把”接入—处理—输出”抽象成可配置的连接器，让数据流像搭积木一样组合。
典型的几个刚需场景包括：
- 数据库 CDC 变更实时同步到缓存与搜索引擎，保证查询侧低延迟可见；
- 多源日志统一清洗、脱敏后投递到对象存储做审计与回溯；
- 交易事件流做风控规则过滤，命中即推送到告警通道；
- 不同团队的数据格式对齐，向下游提供稳定的契约化结构。
这些场景的共同点是”持续、可重放、需要转换”，正是流式管道的用武之地。
二、Benthos 是什么
Benthos 由 Redpanda 团队开源，2024 年起更名为 Redpanda Connect，但社区仍习惯称其为 Benthos，CLI 与配置格式完全兼容。它的核心卖点是一个统一的 YAML 配置描述整条管道，支持 100+ 种输入与输出（Kafka、Redpanda、MQTT、S3、HTTP、数据库 CDC 等），并用 Bloblang 做数据转换。相比 Flink 这类重量级引擎，它单机即可运行、资源占用低、学习曲线平缓。
需要强调的是，它并非要取代 Flink/Spark 这类分布式计算引擎。如果你的需求是复杂的窗口聚合、大规模状态计算，Flink 仍是首选；而当你只需要”可靠地把数据从 A 搬到 B 并做轻量转换”时，Benthos 的运维成本要低一个数量级——没有 ZooKeeper、没有 JobManager，一个二进制加一份配置就能跑。很多团队的实际做法是：用 Benthos 做 80% 的简单管道，仅在真正需要复杂计算时才上 Flink。
三、5 分钟跑通第一个管道
用 Docker 一行即可启动，无需安装运行时（关于 Docker 编排可参考《Docker 实战：容器编排与多服务部署》）：
# 直接运行官方镜像
docker run --rm -v $(pwd)/config.yaml:/benthos/config.yaml \
redpandadata/connect run
# 或本地安装（macOS）
brew install redpanda-data/tap/redpanda-connect
benthos list inputs # 查看支持的输入连接器
下面这条配置会每秒生成一条测试数据，做字段映射后打印到终端：
input:
generate:
interval: 1s
count: 5
mapping: |
root.id = uuid_v4()
root.name = "device-" + (counter() % 10).string()
pipeline:
processors:
- mapping: |
root.device = this.name
root.event_at = now()
root.payload = this
output:
stdout: {}
把内容保存为 config.yaml 后执行 benthos run ./config.yaml，即可看到转换后的 JSON 流。
四、核心概念：四段式结构
每条 Benthos 管道都由四段组成，职责清晰、可独立替换：
| 区段 | 作用 | 常见实现 |
|---|---|---|
| input | 数据来源 | kafka、redpanda、mqtt、file、generate |
| pipeline.processors | 逐条转换 | mapping、filter、awk、compress |
| buffer | 削峰与重放 | memory、file、bounded_memory |
| output | 数据去向 | kafka、s3、http、sql_insert |
其中 buffer 是稳定性关键：当下游短暂不可用时，消息先落盘缓冲，避免上游背压导致丢数据。
五、用 Bloblang 做数据转换
Bloblang 是 Benthos 内置的映射语言，比 JSONPath 更表达力强。下面示例演示字段改名、类型转换与条件分支：
# 原始 JSON：{"user":{"id":1,"name":"alice"},"ts":1710000000}
mapping: |
root.uid = this.user.id
root.uname = this.user.name.uppercase()
# 时间戳转成可读时间
root.happened_at = this.ts.ts_format("2006-01-02T15:04:05Z")
# 仅保留名字以 a 开头的记录
root.keep = this.user.name.regexp_find_all("^a") != []
更复杂时还能解析嵌套 JSON 字符串、做 base64 编解码、按字段聚合。所有转换都是纯函数式、可单测的，调试时用 benthos test 运行断言即可。
再举一个真实场景：上游偶尔把整段报文塞进一个叫 raw 的字符串字段里，我们需要在管道内先解析再展开。Bloblang 可以嵌套解析：
mapping: |
let parsed = this.raw.parse_json()
root.id = $parsed.user.id
root.tags = $parsed.user.tags.slice(0, 3)
# 字段缺失时给出默认值，避免下游空指针
root.region = $parsed.region.or("unknown")
这种”容错式解析”在生产环境非常重要：上游格式一旦抖动，健壮的默认值能让管道继续跑而不是整条崩溃。
六、对接 Kafka / Redpanda 实战
最常见的生产形态是从 Kafka/Redpanda 消费、清洗后再写回另一个 topic。关于 Redpanda 的部署与特性，可延伸阅读《Redpanda 实战：Kafka 兼容的轻量级流平台》，以及《Kafka 入门实战：消息队列选型与可靠性保障》。
input:
kafka:
addresses: [ "redpanda:9092" ]
topics: [ "orders.raw" ]
consumer_group: "benthos-cleaner"
checkpoint_limit: 1024
pipeline:
processors:
- mapping: |
root.order_id = this.id
root.amount = this.total.number()
root.cleaned_at = now()
output:
kafka:
addresses: [ "redpanda:9092" ]
topic: "orders.clean"
max_in_flight: 4
这条管道实时把 orders.raw 中的脏数据清洗成规整结构，投递到 orders.clean，下游服务直接消费干净流，省去了各自的解析逻辑。
七、错误重试、死信队列与可观测性
生产环境必须考虑下游失败。Benthos 提供重试与死信队列（DLQ）机制，避免单条坏消息阻塞整条流：
output:
kafka:
addresses: [ "redpanda:9092" ]
topic: "orders.clean"
retry_as_batch: true # 失败后整批重试
max_retries: 3
dead_letter: # 三次仍失败则转入死信 topic
kafka:
addresses: [ "redpanda:9092" ]
topic: "orders.dlq"
同时可暴露 Prometheus 指标（metrics: 段）查看吞吐、错误率与处理延迟，接入已有的监控体系。如需把配置随代码持续交付，可放进 GitHub Actions 流水线做校验与发布。
推荐的监控组合是：吞吐速率（每秒处理条数）、输出错误率、处理延迟 P99，以及 buffer 使用率。当 buffer 使用率持续攀升，往往意味着下游出现瓶颈，需要优先排查输出侧而非盲目加机器。日志建议统一输出为 JSON 格式（logger: 段配置），便于被 Loki 或 ELK 采集。
八、生产落地检查清单
上线前对照以下清单逐项确认：
| 检查项 | 建议 |
|---|---|
| 消费位移 | 用 consumer_group 托管，避免重复消费 |
| 缓冲策略 | 高吞吐场景用 file buffer 防丢 |
| 幂等输出 | 下游按主键去重，容忍重投 |
| 错误处理 | 必须配置 dead_letter |
| 资源配置 | max_in_flight 控制并发，防压垮下游 |
| 可观测性 | 开启 Prometheus 指标与结构化日志 |
九、小结
Benthos/Redpanda Connect 用一份 YAML 就把流式数据的接入、转换、分发串成可维护的管道，单机即可承载不小的吞吐，是事件驱动架构里性价比极高的”数据胶水”。当你需要在 Kafka、数据库与数据仓库之间搬运并清洗数据时，它往往比自研脚本更稳、更省心。下一步可以结合容器化部署与 CI 流水线，把它纳入标准基础设施。
