From Kafka to Databend Cloud: Engineering a Trillion-Scale Agent Trace Ingestion Pipeline

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.

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

MERGE INTO

At a high level, the pipeline looks like this:

The architectural boundary is intentional:

bend-ingest-kafka

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

The concurrency model inside one instance is straightforward. The program creates multiple Workers based on the

workers

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

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

Relevant events are visible in logs:

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

{
  "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:

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:

{
  "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

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

offset

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

When traffic is high, a batch closes as soon as it reaches

batchSize

batchMaxInterval

{
  "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

.ndjson.zst

zstdWriter, err := zstd.NewWriter(outputFile)
writer = zstdWriter
bufferedWriter := bufio.NewWriter(writer)

Databend uses

COMPRESSION = AUTO

COPY INTO

Batching solves per-message writes, but executing one

COPY INTO

For that reason,

bend-ingest-kafka

COPY INTO

{
  "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.

The generated SQL looks like this:

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

copyIntoFileCount

copyIntoMaxInterval

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

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

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

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

This is an intentional failure-tolerance trade-off. A later

MERGE INTO

action

Network instability is handled with exponential backoff. If Stage upload or COPY fails, the process retries:

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

Power loss, OOM, or

SIGKILL

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

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:

CREATE STREAM trace_raw_stream
ON TABLE trace_raw
APPEND_ONLY = TRUE;

The

MERGE INTO

raw_data

action

The SQL below follows the generation logic used by

GenerateMergeIntoSQL

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()

QUALIFY

action = 'update'

action = 'delete'

at-least-once

This also explains why the pipeline does not parse every trace field inside

bend-ingest-kafka

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:

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

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

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

• 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

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

Databend Cloud takes over the second half of the pipeline. Raw Table keeps the facts. Stream tracks increments.

MERGE INTO

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

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

GitHub: https://github.com/databendcloud/bend-ingest-kafka