--- title: 'From Kafka to Databend Cloud: Engineering a Trillion-Scale Agent Trace Ingestion Pipeline' date: 2026-08-12 slug: category-engineering/agent-trace-ingestion cover_url: 'https://static.databend.cn/img/202608/20260812-kafka-banner.png' description: 'This post is written for data engineers building Agent Trace pipelines, log analytics systems, Kafka-to-warehouse ingestion, or high-throughput semi-structured data platforms. It explains how Agent Trace data flows from Kafka into Databend Cloud, and how the open-source project [bend-ingest-kafka](https://github.com/databendcloud/bend-ingest-kafka) is designed for this workload.' tags: [databend] category: Engineering authors: - Jeremy --- ## When Agent Trace Becomes the Data Layer for Evals A leading AI model company recently faced a data engineering problem that is becoming common in the Agent era. {/* truncate */} After open-sourcing a trillion-scale reinforcement reasoning model, the company began serving workloads with million-token long contexts and complex agentic tasks. A single long-running task could involve thousands of tool calls and process millions of context tokens. As the model moved from single-turn chat to long-horizon task execution, each run produced a growing stream of request parameters, model inputs and outputs, token usage, first-token latency, tool calls, retrieval steps, errors, and deeply nested spans. These traces are not just troubleshooting logs. They explain why a model succeeded, where it failed, and how changes in prompts, tools, harnesses, and model versions affected the final outcome. In practice, Agent Trace becomes a core data asset for building Eval datasets, running attribution analysis, and improving the next generation of models and products. At production peak, the company's online trace ingestion reached terabytes per hour and accumulated toward trillion-record scale over time. They chose Databend Cloud to support the full Agent Trace pipeline, from raw ingestion and incremental processing to business modeling, Eval production, and attribution analysis. ## Why Agent Trace Ingestion Is Hard In an Agent Trace pipeline, the ingestion layer is the first place where pressure shows up. A single model invocation can generate many events. A complex Agent task can expand into multiple levels of spans and run across a long time window. During traffic peaks, large volumes of trace events arrive in Kafka and are distributed across topics and partitions. The ingestion layer needs to satisfy several constraints at the same time: - Catch up with Kafka lag quickly, so backlog does not continue to grow. - Avoid committing Kafka offsets before data is durably written, otherwise data can be lost. - Keep complex JSON parsing out of the Kafka consumer, because trace schemas change frequently. - Avoid one database write per message, since fixed write overhead becomes expensive at scale. - Remain recoverable during process failures, restarts, scaling events, and network instability. `bend-ingest-kafka` sits at this boundary. It continuously consumes raw trace data from Kafka, writes it into a Databend Cloud Raw Table through batched files, and leaves incremental processing to Databend Stream and `MERGE INTO`. At a high level, the pipeline looks like this: ![](https://static.databend.cn/img/202608/20260812-01-end-to-end-trace-ingestion.png) The architectural boundary is intentional: `bend-ingest-kafka` is responsible for moving raw data into Databend Cloud quickly, completely, and recoverably. Business-specific parsing, field extraction, deduplication, and modeling happen inside Databend. This separation keeps ingestion and transformation decoupled. When traffic grows, Kafka consumers and Databend Warehouses can scale independently. When trace fields change, teams do not need to redeploy every Kafka consumer just to keep ingestion alive. ## Extending Kafka Parallelism into Databend Writes Kafka already uses topics and partitions to split a large trace stream into many independently consumable data flows. To use that parallelism effectively, the consumer side cannot be a single-process, single-threaded writer. In production, multiple `bend-ingest-kafka` instances are deployed for the same topic. Those instances use the same Kafka consumer group. Inside each instance, multiple Workers can run in parallel. Each Worker owns its own Kafka Consumer, Batch Reader, and pending file queue. Kafka Consumer Group coordination assigns partitions across all instances and Workers. ![](https://static.databend.cn/img/202608/20260812-02-parallel-consumption-workers.png) The concurrency model inside one instance is straightforward. The program creates multiple Workers based on the `workers` configuration and runs each Worker in its own goroutine: ```Go 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() }() } ``` This is more than starting more Kafka Consumers. After records are read, NDJSON generation, Zstd compression, Stage upload, and `COPY INTO` execution also proceed independently per Worker. When the number of instances and Workers increases, the entire write channel becomes wider instead of pushing pressure into a single write path. The useful degree of parallelism is still bounded by Kafka partitions. For example, if a topic has 144 partitions and the deployment runs 12 instances with 4 Workers each, the system has 48 consuming units. On average, each Worker processes about 3 partitions. If more instances are added, Kafka rebalances partition ownership. If partitions are added online, `bend-ingest-kafka` periodically refreshes topic metadata and joins the new assignment. Relevant events are visible in logs: ```Plaintext Partitions revoked: [...] Partitions assigned: [...] ``` Whether scale-out has taken effect, and whether newly added partitions are being consumed, can be verified directly from rebalance logs and Kafka lag. ## Raw Mode: Capture the Full Event First, Model Later This production pipeline uses the Raw Mode of `bend-ingest-kafka`: ```JSON { "isJsonTransform": false } ``` In this mode, the ingestion process does not force each trace JSON object into a fixed business schema. Instead, it writes the raw payload and Kafka metadata into a Raw Table: ```SQL CREATE TABLE trace_raw ( uuid STRING, koffset BIGINT, kpartition INT, raw_data JSON, record_metadata JSON, add_time TIMESTAMP ); ``` A record in the Raw Table looks like this: ```JSON { "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 schemas evolve quickly. A model release may introduce reasoning tokens, cache-hit fields, or billing attributes. An Agent framework upgrade may change tool-call structures, context compression records, or error formats. If the ingestion process needs to understand every field before writing data, one upstream schema change can become a consumer upgrade. Raw Mode moves this problem out of the real-time ingestion path. As long as a message is valid JSON, `bend-ingest-kafka` first stores it completely. Field extraction, type conversion, and target-table modeling are handled later by Databend SQL. This design has three practical benefits: - The ingestion process uses less CPU because it avoids heavy JSON transformation. - Schema evolution does not block Kafka consumption. - Raw trace data remains intact, so teams can reprocess it when parsing rules change. The Raw Table also keeps `topic`, `partition`, and `offset`. Together, these fields form a stable identity for a Kafka message. If a failure causes replay, downstream jobs can deduplicate by this identity. If engineers need to trace a Databend record back to its Kafka source, the original consume position is still available. ## Batch File Writes: Do Not Turn Every Message into a Warehouse Insert Many small writes are inefficient for analytical databases. After reading messages from Kafka, `bend-ingest-kafka` groups them into batches and writes each batch as an NDJSON file. When traffic is high, a batch closes as soon as it reaches `batchSize`. When traffic is low, `batchMaxInterval` bounds the waiting time so that records do not sit indefinitely waiting for a full batch: ```JSON { "batchSize": 10000, "batchMaxInterval": 10 } ``` High-traffic topics can use large batches to amortize fixed costs. Low-traffic topics still get a predictable visibility delay. Raw trace records often contain many repeated JSON field names, so compression is effective. When `copyIntoUploadCompression` is enabled, the program uses a buffered writer to generate NDJSON and compress it with Zstd at the same time. The uploaded file is an `.ndjson.zst` file: ```Go zstdWriter, err := zstd.NewWriter(outputFile) writer = zstdWriter bufferedWriter := bufio.NewWriter(writer) ``` Databend uses `COMPRESSION = AUTO` in `COPY INTO` to detect and decompress the file automatically. This reduces network transfer and temporary Stage storage, which is especially useful for raw JSON trace data with repeated structure. Batching solves per-message writes, but executing one `COPY INTO` per file still pays fixed overhead too frequently. SQL request handling, task scheduling, file parsing initialization, and transaction commit are not perfectly proportional to the number of rows inside a file. When files are small, this fixed cost becomes visible. For that reason, `bend-ingest-kafka` adds two triggers for `COPY INTO`: ```JSON { "copyIntoFileCount": 128, "copyIntoMaxInterval": 5 } ``` These conditions use OR semantics: - Files are generated and uploaded to Stage as usual. - When 128 pending files are available, COPY runs immediately. - Starting from the first successfully uploaded file, if 5 seconds pass, COPY runs even if the pending file count is below 128. This gives high-throughput topics enough aggregation while also bounding write latency for lower-throughput topics. ![](https://static.databend.cn/img/202608/20260812-03-five-files-one-copy-into.png) The generated SQL looks like this: ```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; ``` Assume `batchSize` is 10,000. Under high traffic, COPY runs as soon as the pending file count reaches `copyIntoFileCount`. Under low traffic, the first uploaded file waits no longer than `copyIntoMaxInterval` before COPY runs. Individual files stay bounded in size, fixed overhead is amortized, and write latency has a clear upper bound. Each Worker maintains its own pending file queue. Workers do not mix files with each other. Multiple Workers and multiple instances can generate files, upload files to Stage, and execute independent `COPY INTO` statements at the same time. Kafka partition parallelism is therefore carried all the way into the Databend Cloud write side. ## Offset Commit: Wait Until Data Is Really in the Raw Table One of the easiest ways to lose data in a Kafka ingestion pipeline is to commit offsets too early. The `bend-ingest-kafka` processing order is: ![](https://static.databend.cn/img/202608/20260812-04-commit-offset-after-copy.png) Uploading a file to Stage does not mean the message has been processed. If the process commits offsets after upload but exits before `COPY INTO`, Kafka will treat those messages as consumed while the Raw Table still has no corresponding rows. That creates a gap that cannot be recovered automatically. To avoid this, each Worker keeps both the uploaded Stage files and their corresponding Kafka batches in memory. Only after the whole group of files is successfully written into the Raw Table does the Worker commit the offsets for those batches. If COPY fails, the program retries with the same already uploaded files. It does not regenerate and reupload the files. Until that pending group is completed, the Worker does not read more messages. This means downstream pressure stays in Kafka instead of growing without bound inside the process memory. The pipeline uses `at-least-once` semantics. It prioritizes no data loss, but one failure boundary can create duplicates: Databend may finish COPY, but the client process may fail before committing Kafka offsets. After restart, Kafka will redeliver that batch, and the Raw Table may contain duplicate rows. This is an intentional failure-tolerance trade-off. A later `MERGE INTO` step aggregates Stream increments by business key and uses `action` to decide the final state written into the target table. Compared with adding complex cross-system transactions to chase exactly-once ingestion at the consumer layer, this model is easier to recover and more suitable for a high-throughput pipeline. Network instability is handled with exponential backoff. If Stage upload or COPY fails, the process retries: ```Plaintext 1 second -> 2 seconds -> 4 seconds -> 8 seconds -> ... -> maxRetryDelay ``` During a normal deployment or scale-down, graceful shutdown handles the final boundary. Even if the pending queue contains fewer files than `copyIntoFileCount`, the Worker writes the remaining files into the Raw Table, commits the corresponding offsets, and only then closes the Kafka Consumer. ![](https://static.databend.cn/img/202608/20260812-05-graceful-shutdown-flush-before-close.png) Power loss, OOM, or `SIGKILL` cannot run the graceful shutdown path. In those cases, uncommitted messages are replayed by Kafka and duplicates are eventually removed by downstream MERGE logic. ## Removing Duplicate Effects with Stream and MERGE INTO After data reaches the Raw Table, the ingestion phase is finished. Field parsing, type conversion, and deduplication no longer consume Kafka Consumer resources. They continue inside Databend. These steps can be split into multiple Databend Cloud Tasks. An upstream Task can consume Stream increments and run `MERGE INTO`; downstream Tasks can continue with field expansion, data cleaning, and derived metric computation. Each Task can bind to a Warehouse and run on a schedule or as part of a dependency chain. If every transformation rescans the full Raw Table, cost grows quickly once historical data reaches tens of billions or trillions of records. Databend Stream tracks incremental changes on the Raw Table, so downstream jobs only process records added since the previous run. Conceptually, an append-only Stream can be created on the Raw Table: ```SQL CREATE STREAM trace_raw_stream ON TABLE trace_raw APPEND_ONLY = TRUE; ``` ![](https://static.databend.cn/img/202608/20260812-06-incremental-processing-after-raw-table.png) The `MERGE INTO` step after the Stream projects target-table fields from `raw_data`, selects the latest row for each business key within the same Stream increment, and applies update, delete, or insert behavior based on `action`. The SQL below follows the generation logic used by `GenerateMergeIntoSQL` in the project. Real column names and business keys depend on the trace schema: ```SQL 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 *; ``` If the same business key appears multiple times in one increment, `ROW_NUMBER()` with `QUALIFY` selects one row according to the project's generation logic. `action = 'update'` updates the target row. `action = 'delete'` deletes it. Other non-delete events are inserted when the target row does not exist. The ingestion layer remains `at-least-once`, while the business table reaches eventual idempotency. This also explains why the pipeline does not parse every trace field inside `bend-ingest-kafka`. The ingestion program performs stable, generic work. Databend handles incremental computation and structured processing. When a trace adds fields, teams can change downstream SQL without upgrading every Kafka Consumer. When parsing logic is wrong, the original records remain in the Raw Table and can be processed again with new rules. ![](https://static.databend.cn/img/202608/20260812-07-ingestion-processing-boundary.png) ## Verifying Multi-File COPY with an End-to-End Test To validate Raw Mode and multi-file COPY behavior, the project includes an end-to-end test with 100,000 records. The test connects to real Kafka, a real Databend Stage, and a real Raw Table. The test uses the following parameters: ```Plaintext Total messages: 100,000 isJsonTransform: false batchSize: 1,000 copyIntoFileCount: 5 Worker: 1 Kafka Partition: 1 ``` With 100,000 messages and 1,000 messages per batch, the run generates 100 files. With one COPY per 5 files, the expected number of `COPY INTO` executions is 20. The actual result matches that relationship: ```Plaintext Raw Table rows: 100,000 Unique Kafka offsets: 100,000 Offset range: 0 to 99,999 Stage uploads: 100 COPY INTO executions: 20 Write errors: 0 ``` In the local test environment, one Worker completed Kafka consumption, Raw data wrapping, Zstd compression, Stage upload, COPY, and offset commit in about 6.58 seconds, reaching roughly 15,208 rows per second. This test validates correctness and multi-file COPY behavior. It is not a benchmark of Databend Cloud's upper limit. Production throughput depends on average trace size, Kafka partition count, number of instances and Workers, network bandwidth, and Databend Warehouse size. The important relationship is still clear: 100,000 messages generated 100 files, only 20 COPY executions were needed, and the Raw Table contained 100,000 records with continuous offsets and complete metadata. ## Trillion Scale Does Not Come from One Large Machine Trillion-scale ingestion describes long-term accumulated data volume and the ability to keep up with production peaks. It does not come from one oversized process. It comes from each layer in the pipeline being able to scale. Kafka can add more topics and partitions. `bend-ingest-kafka` can add more instances and Workers. Databend Cloud can adjust Warehouses based on write and transformation pressure. Once Raw Table and business tables are decoupled by Stream, ingestion speed is no longer tied directly to complex JSON parsing and deduplication. ![](https://static.databend.cn/img/202608/20260812-08-horizontal-scaling.png) When traffic grows, the basic system shape does not need to change. New Consumers take over more partitions. More Workers generate and upload files in parallel. Multi-file COPY continues to reduce fixed overhead. After data enters the Raw Table, independent compute resources handle incremental transformation. From Kafka's perspective, `bend-ingest-kafka` is a Consumer. From Databend Cloud's perspective, it is a continuously running batch ingestion channel. What makes it suitable for large-scale trace ingestion is not one isolated optimization, but a set of small decisions working together: - Batch size controls individual file size. - Zstd reduces network transfer and temporary storage cost. - Multi-file `COPY INTO` amortizes scheduling and execution overhead. - Delayed offset commit protects data integrity. - Retry and graceful shutdown handle failure boundaries. - Multiple instances and Workers carry Kafka partition parallelism into the Databend write path. Raw Table, Stream, and `MERGE INTO` complete the second half of the pipeline. Raw traces are first preserved in full. New records then enter incremental processing. Duplicate messages are removed before landing in business tables. The resulting data serves Evals, attribution analysis, and large-model observability queries. When online trace ingestion reaches terabytes per hour and long-term data moves toward trillion scale, this division of responsibility is easier to scale and evolve than a Consumer that tries to own every business rule. ## Conclusion `bend-ingest-kafka` does not try to finish every job inside the Kafka Consumer. It focuses on the stable ingestion responsibilities that matter most: parallel consumption, batch file generation, compression, Stage upload, multi-file COPY, delayed offset commit, failure retry, and graceful shutdown. Databend Cloud takes over the second half of the pipeline. Raw Table keeps the facts. Stream tracks increments. `MERGE INTO` performs structured processing and eventual idempotency. Warehouses isolate write, transformation, and analytical workloads. For Agent Trace, the value of a data platform is not only storing logs. The harder problem is turning continuously growing, schema-drifting, and often sensitive execution records into a reusable data asset for engineering teams. In this pipeline, Databend Cloud acts as the unified data layer for Agent Trace: Raw Table preserves complete JSON and Kafka metadata, Stream and `MERGE INTO` support incremental processing for Evals, independent Warehouses isolate workloads, and full traces continue to support attribution, replay, training data production, and model iteration. For teams building Agent Trace pipelines, the reusable idea is simple: do not make the ingestion layer understand every business semantic. First capture data completely, durably, and recoverably. Then move complex transformation to a data platform designed for incremental computation and analytics. The `bend-ingest-kafka` project is open source: GitHub: [https://github.com/databendcloud/bend-ingest-kafka](https://github.com/databendcloud/bend-ingest-kafka)