导读本文是《从万亿级大模型到全线应用Databend Cloud 助力头部 AI 企业构建全链路 Trace 数据管道》的技术延伸面向正在建设 Agent Trace、日志分析、Kafka 入仓或高吞吐半结构化数据管道的数据工程师重点拆解 Agent Trace 从 Kafka 进入 Databend Cloud 的接入链路以及开源项目 bend-ingest-kafka 在其中的设计与实现。从案例说起当 Agent Trace 成为 Evals 的数据底座某中国头部 AI 大模型企业在开源万亿级强化推理大模型后开始面对一个典型的 Agent 时代数据工程问题。该模型面向百万级超长上下文与 Agent 级复杂任务处理一次长任务可能包含数千次工具调用累计处理数百万 context tokens。随着模型能力从单轮对话走向长程任务执行运行过程中会持续产生请求参数、模型输入输出、Token 用量、首字延迟、工具调用、检索过程、错误信息以及多层嵌套 Span。这些 Trace 不只是排查问题的日志。它们记录了模型为什么成功、在哪里失败以及 Prompt、工具、Harness 和模型版本的变化如何影响最终结果是构建 Evals 数据集、开展归因分析和推动模型迭代的重要数据资产。在生产高峰期该企业的在线 Trace 写入达到每小时 TB 级并长期累积到万亿级规模。他们选择 Databend Cloud 承载全链路 Agent Trace 数据构建一条贯穿原始数据接入、增量处理、业务建模、Evals 生产与归因分析的统一数据管道。Agent Trace 接入难在哪里在 Agent Trace 数据管道中最先承受压力的是接入链路。一次模型调用会产生大量事件。复杂 Agent 任务还会继续展开为多层 Span并跨越较长时间窗口。业务高峰到来时大量 Trace 同时涌入 Kafka被分散到多个 Topic 和 Partition。接入端需要同时满足几个要求尽快追平 Kafka lag不能让积压持续扩大。不能因为追求吞吐而提前提交 offset造成数据丢失。不能把复杂 JSON 解析全部放在 Consumer 里否则 Schema 频繁变化会拖慢链路演进。不能用一条消息触发一次数据库写入否则固定开销会被放大。需要在故障、重启、缩容、网络抖动时保持可恢复。bend-ingest-kafka工作在这个位置。它从 Kafka 持续消费原始 Trace以批量文件写入 Databend Cloud 的 Raw Table再由 Databend Stream 跟踪增量交给MERGE INTO解析、去重并写入最终业务表。整条链路可以概括成下面这张图这套架构的核心边界很清晰bend-ingest-kafka负责把原始数据快速、完整、可恢复地送入 Databend Cloud复杂的业务解析、字段抽取和去重留在 Databend 内部完成。这种分工让接入与加工解耦。流量增长时可以独立扩容 Consumer 和 Databend WarehouseTrace 字段变化时也不需要频繁修改 Kafka 消费程序。把 Kafka 的并行度延伸到 Databend 写入端Kafka 已经用 Topic 和 Partition 把海量 Trace 切成许多可以并行处理的数据流。要真正利用这份并行能力消费端不能停留在单进程、单线程模型。生产环境会部署多个bend-ingest-kafka实例。同一 Topic 对应的实例使用相同 consumer group。每个实例内部还可以启动多个 Worker每个 Worker 都拥有独立的 Kafka Consumer、Batch Reader 和待写入文件队列。Kafka Consumer Group 负责在所有实例和 Worker 之间分配 Partition。源码中的实例内并发结构很直接。程序按照workers配置创建多个 Worker并让它们各自在 Goroutine 中运行wg.Add(cfg.Workers) for i : 0; i cfg.Workers; i { w : NewConsumeWorker(cfg, fmt.Sprintf(worker-%d, i), ig) go func() { w.Run(ctx) wg.Done() }() }这不只是多开几个 Kafka Consumer。数据读取之后的 NDJSON 生成、Zstd 压缩、Stage 上传和COPY INTO同样会随 Worker 并行展开。实例数和 Worker 数增加后整个写入通道一起变宽而不是把压力集中到某个单线程写入点。有效并行度仍然受 Kafka Partition 数量约束。假设一个 Topic 有 144 个 Partition部署 12 个实例、每个实例运行 4 个 Worker就会形成 48 个消费单元平均每个 Worker 处理约 3 个 Partition。继续增加实例时Kafka 会通过 rebalance 重新分配 Partition在线增加 Partition 后bend-ingest-kafka也会定期刷新 Topic metadata发现变化并重新加入分配。相关过程会写入日志Partitions revoked: [...] Partitions assigned: [...]扩容是否生效、新增 Partition 是否已经开始消费都可以从 rebalance 日志和 Kafka lag 中直接确认。Raw Mode先完整接住变化再做业务建模这条生产链路使用bend-ingest-kafka的 Raw Mode{ isJsonTransform: false }在这个模式下接入程序不会把 Trace JSON 强行映射到固定业务表而是将原始数据连同 Kafka 元数据一起写入 Raw TableCREATE TABLE trace_raw ( uuid STRING, koffset BIGINT, kpartition INT, raw_data JSON, record_metadata JSON, add_time TIMESTAMP );一条落入 Raw Table 的记录大致如下{ uuid: 0bc7efb9-8e4c-4a8a-a0c9-6a442b0523fe, koffset: 18273645, kpartition: 37, record_metadata: { topic: llm-traces, partition: 37, offset: 18273645, key: trace-key, create_time: 2026-08-10T10:00:00Z }, add_time: 2026-08-10T10:00:01Z, raw_data: { trace_id: trace-001, span_id: span-001, model: large-model-v3, usage: { prompt_tokens: 1234, completion_tokens: 356 } } }Agent Trace 的结构变化很快。模型版本更新后可能新增推理 Token、缓存命中、计费字段Agent 框架升级后工具调用、上下文压缩和错误结构也可能发生变化。如果接入程序需要逐字段理解所有数据上游的一次字段变更就可能变成一次消费者升级。Raw Mode 把这个问题移出了实时接入的关键路径。只要消息仍是合法 JSONbend-ingest-kafka就先把它完整保存下来。字段如何解析、类型如何转换、最终进入哪张分析表交给后续 Databend SQL 完成。这样做有三个收益接入端 CPU 消耗更低不被复杂 JSON 解析拖慢。Schema 演进不会阻塞 Kafka 消费。原始 Trace 被完整保留后续解析规则调整后仍然可以重新加工。Raw Table 还保留了topic、partition和offset。这三个字段组合起来就是一条 Kafka 消息稳定的身份标识。发生故障重放时下游可以据此去重需要回溯数据时也能把 Databend 中的记录准确对应回 Kafka 消费位置。批量文件写入不要让一条消息触发一次入仓海量小请求是分析型数据库写入最不经济的方式。bend-ingest-kafka从 Kafka 读取消息后会先组成 Batch再生成 NDJSON 文件。流量充足时消息数达到batchSizeBatch 立即结束流量较低时batchMaxInterval会限制等待时间避免数据因为迟迟攒不满而长时间停留。{ batchSize: 10000, batchMaxInterval: 10 }高流量 Topic 可以依靠大 Batch 摊薄固定开销低流量 Topic 也能守住可见延迟。Raw Trace 里有大量重复的 JSON 字段名压缩收益通常很可观。开启copyIntoUploadCompression后程序用带缓冲的 Writer 一边生成 NDJSON一边完成 Zstd 压缩最后上传的是.ndjson.zst文件zstdWriter, err : zstd.NewWriter(outputFile) writer zstdWriter bufferedWriter : bufio.NewWriter(writer)Databend 在COPY INTO中使用COMPRESSION AUTO自动识别并解压。网络传输和 Stage 临时存储体积随之下降特别适合原始 Trace 这类字段重复度较高的 JSON 数据。Batch 解决了逐条写入的问题但如果每生成一个文件就执行一次COPY INTO固定开销仍然会被频繁支付。SQL 请求、任务调度、文件解析初始化和事务提交与文件里有多少条数据并不是完全成正比。文件较小时这些固定成本尤其明显。因此bend-ingest-kafka增加了文件数与等待时间两个触发条件{ copyIntoFileCount: 128, copyIntoMaxInterval: 5 }它们是 OR 关系文件照常生成后立即上传到 Stage。Pending 文件达到 128 个时立即 COPY不必等待 5 秒。从第一个文件上传成功开始达到 5 秒时即使不足 128 个也立即 COPY。这样既能在高流量下充分聚合文件也能限制低流量场景的写入延迟。最终生成的 SQL 类似这样COPY INTO trace_raw FROM ~/batch/ FILES ( file-1.ndjson.zst, file-2.ndjson.zst, file-3.ndjson.zst, file-4.ndjson.zst, file-5.ndjson.zst ) FILE_FORMAT ( TYPE NDJSON MISSING_FIELD_AS FIELD_DEFAULT COMPRESSION AUTO ) PURGE TRUE FORCE FALSE DISABLE_VARIANT_CHECK TRUE;假设batchSize为 10,000高流量下pending 文件达到copyIntoFileCount就立即执行 COPY低流量下第一个已上传文件等待达到copyIntoMaxInterval也会触发 COPY。单文件大小不会失控固定开销得到摊薄写入延迟也有明确上限。每个 Worker 都维护自己的文件队列不同 Worker 不会把 pending 文件混在一起。多个 Worker、多个实例可以同时生成文件、上传 Stage并向 Raw Table 发起相互独立的 COPY。Kafka 侧的多 Partition 并发由此一路延伸到了 Databend Cloud 写入端。Offset 提交必须等数据真正进入 Raw Table接入链路里最容易被忽视也最容易造成数据丢失的是 offset 提交时机。bend-ingest-kafka的处理顺序始终是文件已经上传到 Stage并不意味着消息已经处理完成。假如这时提前提交 offset进程却在COPY INTO之前退出Kafka 会认为消息已经消费而 Raw Table 里还没有数据最终留下无法自动恢复的缺口。所以Worker 在内存中同时保留 Stage 文件和对应的 Kafka Batch。只有整组文件成功写入 Raw Table才依次提交这些 Batch 的 offset。COPY 失败时程序继续使用已经上传的同一组文件重试不会重复生成和上传这组 pending 文件处理完成之前Worker 也不会继续读取更多消息。这意味着下游异常时积压留在 Kafka而不是无边界地堆进进程内存。这条链路采用的是at-least-once语义。它优先保证数据不丢但在一个特殊边界上可能产生重复Databend 已经完成 COPY客户端还未来得及提交 Kafka offset进程就发生故障。重启后这批消息会再次消费Raw Table 里可能出现两份相同数据。这是有意保留的容错空间。后续MERGE INTO会按照业务主键聚合同一批 Stream 增量并结合action取出需要落入目标表的最新状态。相比为了追求接入端“恰好一次”而引入复杂的跨系统事务这种方式更适合高吞吐链路也更容易在故障后恢复。网络抖动同样不会立刻打断消费程序。Stage 上传和 COPY 失败后程序按指数退避重试1 秒 → 2 秒 → 4 秒 → 8 秒 → …… → maxRetryDelay正常发布或实例缩容时优雅退出会接住最后一个边界。即使 pending 队列里不足copyIntoFileCountWorker 也会在退出前把剩余文件写入 Raw Table、提交对应 offset然后再关闭 Kafka Consumer。节点断电、OOM 或SIGKILL无法执行这段流程但未提交的消息仍会由 Kafka 重放最终再由下游 MERGE 消除重复。用 Stream 和 MERGE INTO 消除重复写入影响数据写进 Raw Table 后接入阶段就结束了。Trace 的字段解析、类型转换和去重不再占用 Kafka Consumer 的资源而是在 Databend 内部继续完成。这些步骤可以在 Databend Cloud 中拆成多个 Task 执行前置 Task 消费 Stream 增量并运行MERGE INTO后续 Task 继续完成字段展开、数据清洗和派生指标计算每个 Task 都可以绑定独立 Warehouse并按计划或依赖关系自动运行。如果每次加工都扫描整张 Raw Table历史数据达到百亿、万亿级后成本会越来越高。Databend Stream 会跟踪 Raw Table 的增量变化让下游任务只读取上次处理之后新增的数据。概念上可以在 Raw Table 上创建 Append-only StreamCREATE STREAM trace_raw_stream ON TABLE trace_raw APPEND_ONLY TRUE;Stream 后面的MERGE INTO会从raw_data中投影目标表字段并在同一批 Stream 增量中按业务主键取最新一条再根据action完成更新、删除或插入。下面的 SQL 与项目中GenerateMergeIntoSQL的生成逻辑一致真实列名和业务主键会随 Trace Schema 调整MERGE INTO trace_detail a USING ( SELECT raw_data:id AS id, raw_data:trace_id AS trace_id, raw_data:span_id AS span_id, raw_data:model AS model, raw_data:prompt_tokens AS prompt_tokens, raw_data:completion_tokens AS completion_tokens, action FROM trace_raw_stream QUALIFY ROW_NUMBER() OVER (PARTITION BY id ORDER BY add_time) 1 ) b ON a.id b.id WHEN MATCHED AND b.action update THEN UPDATE * WHEN MATCHED AND b.action delete THEN DELETE WHEN NOT MATCHED AND b.action ! delete THEN INSERT *;同一个业务主键在一批增量里出现多次时ROW_NUMBER()配合QUALIFY按项目中的生成逻辑筛出一条记录。actionupdate更新目标记录actiondelete删除目标记录其他非删除事件在目标表不存在时插入。接入端的at-least-once由此转化为业务表上的最终幂等。这也解释了为什么整条链路没有在bend-ingest-kafka里解析所有 Trace 字段。接入程序只做稳定且通用的工作Databend 负责它更擅长的增量计算和结构化处理。Trace 增加字段时可以修改后面的 SQL而不必升级全部 Kafka Consumer解析规则发生错误时Raw Table 中的原始数据仍然保留也能按新规则重新加工。用真实链路验证多文件 COPY 的行为为了验证 Raw Mode 和多文件 COPY 的实际行为项目加入了一组 10 万条数据的全链路测试。测试连接真实 Kafka、真实 Databend Stage 和真实 Raw Table使用下面的参数消息总数 100,000 isJsonTransform false batchSize 1,000 copyIntoFileCount 5 Worker 1 Kafka Partition 1100,000 条消息按每批 1,000 条生成 100 个文件每 5 个文件执行一次 COPY理论上应当产生 20 次COPY INTO。实际运行结果与这个关系完全一致Raw Table 行数 100,000 唯一 Kafka offset 100,000 offset 范围 0 99,999 Stage 上传次数 100 COPY INTO 次数 20 写入错误 0本地测试环境中单 Worker 完成 Kafka 消费、Raw 数据封装、Zstd 压缩、Stage 上传、COPY 和 offset commit总耗时约 6.58 秒吞吐约为 15,208 rows/s。这组数据用于验证全链路正确性和多文件 COPY 的执行效果不代表 Databend Cloud 的性能上限。生产吞吐还取决于 Trace 平均大小、Kafka Partition 数、实例与 Worker 数、网络带宽以及 Databend Warehouse 规格。不过它清楚地验证了一个关键关系100,000 条消息生成 100 个文件最终只需要 20 次 COPYRaw Table 中仍然得到 100,000 条 offset 连续、元数据完整的记录。万亿级不是一台机器跑出来的万亿级描述的是长期累积的数据规模以及持续面对高峰流量时的系统能力。它不依赖某个配置惊人的单机进程而是来自整条链路每一层都能扩展。Kafka 可以拆分更多 Topic、增加 Partitionbend-ingest-kafka可以增加实例和 WorkerDatabend Cloud 可以根据写入和加工压力调整 Warehouse。Raw Table 与业务表之间由 Stream 解耦后接入速度也不再和复杂 JSON 解析、去重任务绑定在一起。流量增长时这条路径不需要改变系统的基本形态。新的 Consumer 接管更多 Partition更多 Worker 并行生成和上传文件多文件 COPY 继续降低固定开销数据进入 Raw Table 后再由独立的计算资源完成增量加工。从 Kafka 看bend-ingest-kafka是一个 Consumer从 Databend Cloud 看它是一条持续运行的批量写入通道。真正让它能够承担海量 Trace 接入的并不是某一个孤立优化而是这些细节共同形成的结果Batch 控制单文件大小。Zstd 降低传输和临时存储成本。多文件COPY INTO摊薄调度开销。offset 延迟提交守住数据完整性。重试和优雅退出处理故障边界。多实例与多 Worker 把 Kafka 的 Partition 并行度延伸到 Databend 写入端。Raw Table、Stream 和MERGE INTO则补上了链路的后半程。原始 Trace 先被完整保存新增数据随后进入增量处理重复消息在业务表落地前被消除最终服务于 Evals、归因分析和大模型可观测查询。当在线 Trace 写入达到每小时 TB 级长期数据走向万亿规模这种“接入先求稳、加工留在后端”的分工会比一个包办所有逻辑的 Consumer 更容易扩展也更容易演进。小结bend-ingest-kafka的设计并不追求在 Consumer 里完成所有工作。它把最关键、最通用的接入职责做好并行消费、批量文件写入、压缩、Stage 上传、多文件 COPY、offset 延迟提交、失败重试和优雅退出。Databend Cloud 则接过后半程Raw Table 保留事实Stream 追踪增量MERGE INTO完成结构化处理与最终幂等Warehouse 隔离写入、加工和分析负载。对 Agent Trace 来说数据平台的价值不只是“把日志存下来”。更关键的是它要把持续增长、结构漂移、包含敏感信息的原始执行记录转化成可以被工程团队反复使用的数据资产。Databend Cloud 在这条链路中承担的是 Agent Trace 的统一数据层Raw Table 保留完整 JSON 与 Kafka 元数据Stream 和MERGE INTO支撑面向 Evals 的增量加工独立 Warehouse 隔离写入、加工和分析负载完整 Trace 则可以继续服务归因、回放、训练数据生产和模型迭代。Databend Cloud 不只是分析型数据库而是面向现代 AI 应用的 Agent Trace 数据底座。它让 Trace 从“事后排查日志”变成“持续反馈系统”帮助团队更快定位问题、更快构建评测数据集也更快推动模型和 Agent 产品进化。对于正在建设 Agent Trace 数据管道的团队这套架构提供了一个可复用的工程思路不要让接入层理解所有业务语义先完整、稳定、可恢复地接住数据再把复杂加工交给更适合做增量计算和分析的数据平台。bend-ingest-kafka项目已开源GitHubhttps://github.com/databendcloud/bend-ingest-kafka相关阅读《从万亿级大模型到全线应用Databend Cloud 助力头部 AI 企业构建全链路 Trace 数据管道》《Agent 轨迹分析与归因的数据工程实践》
