K2I: A High-Performance Streaming Ingestion Engine for Apache Iceberg

1. Introduction

The data lakehouse paradigm has emerged as the dominant architecture for modern analytical workloads, combining the flexibility of data lakes with the performance characteristics of data warehouses. Apache Iceberg has become the de facto open table format, providing ACID transactions, schema evolution, and time travel capabilities on top of commodity object storage.

However, ingesting streaming data into Iceberg tables presents significant challenges. Traditional batch ETL introduces unacceptable latency for real-time use cases, while naive streaming approaches create excessive small files that degrade query performance and increase metadata overhead.

K2I addresses these challenges through a purpose-built streaming ingestion engine that:

• Minimizes data latency: Sub-second freshness via in-memory hot buffer
• Optimizes file sizes: Intelligent buffering produces read-optimized Parquet files
• Ensures data integrity: Exactly-once semantics through write-ahead logging
• Reduces operational burden: Automated compaction, expiration, and cleanup
• Maximizes performance: Native Rust implementation with zero-copy data paths

1.1 Inspiration

K2I draws inspiration from the Moonlink architecture developed by Mooncake Labs, which pioneered the concept of combining in-memory buffering with Iceberg's durable storage. K2I extends this approach with:

• Kafka-native ingestion (vs. CDC-focused)
• Multiple catalog backend support
• Comprehensive maintenance automation
• Production-grade observability

2. Problem Statement

2.1 The Latency-Cost Trade-off

Organizations face a fundamental tension when building data pipelines:

Real-time systems require maintaining separate infrastructure, duplicating storage, and managing complex stateful processing. Batch systems are simpler but cannot serve time-sensitive use cases.

2.2 The Small File Problem

Streaming data into Iceberg creates a cascade of problems:

1. File proliferation: Each micro-batch creates new files, potentially thousands per hour
2. Metadata explosion: Each file requires manifest entries, increasing planning time
3. Query degradation: Small files increase I/O operations and reduce scan efficiency
4. Storage overhead: Per-file metadata and object storage API costs accumulate

2.3 Exactly-Once Challenges

Achieving exactly-once semantics in streaming systems requires coordination between:

• Message consumption acknowledgment
• State persistence
• External system commits

Failures at any point can result in data loss or duplication, requiring complex distributed transactions or idempotency mechanisms.

2.4 Operational Complexity

Production streaming pipelines require ongoing maintenance:

• File compaction to optimize query performance
• Snapshot expiration to manage metadata growth
• Orphan file cleanup from failed operations
• Schema evolution handling
• Monitoring and alerting

3. Design Philosophy

K2I is built on four foundational principles:

3.1 Single-Process Simplicity

Unlike distributed streaming frameworks (Flink, Spark Streaming), K2I runs as a single process per topic partition. This eliminates:

• Distributed coordination overhead (ZooKeeper, consensus protocols)
• Network serialization between processing stages
• Complex failure recovery across nodes
• Operational complexity of cluster management

Single-process design provides deterministic behavior, simpler debugging, and predictable resource consumption.

3.2 Hot/Cold Data Separation

K2I maintains two storage tiers:

Hot Tier (In-Memory)

• Apache Arrow columnar format
• Hash indexes for O(1) key lookup
• Sub-millisecond query latency
• Bounded by configurable memory limit

Cold Tier (Object Storage)

• Apache Parquet files on S3/GCS/Azure
• Optimized for analytical queries
• Unlimited retention
• Cost-efficient at scale

This separation allows immediate data availability while accumulating sufficient data for optimal file sizes.

3.3 Transaction Log Durability

All state-changing operations are recorded in an append-only transaction log before execution. This provides:

• Crash recovery: Reconstruct state from log replay
• Exactly-once semantics: Idempotency through operation tracking
• Audit trail: Complete history of data mutations
• Debugging: Forensic analysis of failures

3.4 Defensive Design

K2I assumes failures will occur and designs for resilience:

• Backpressure propagation: Pause consumption when buffer is full
• Circuit breakers: Prevent cascading failures to external systems
• Graceful degradation: Continue operating with reduced functionality
• Automatic recovery: Self-heal from transient failures

4. System Architecture

4.1 High-Level Overview

+-----------------------------------------------------------------------------+
|                           K2I Ingestion Engine                              |
+-----------------------------------------------------------------------------+
|                                                                             |
|   +------------------+    +------------------+    +------------------+      |
|   | SmartKafka       |    |    Hot Buffer    |    | Iceberg Writer   |      |
|   | Consumer         |--->| (Arrow + Index)  |--->| (Parquet)        |      |
|   |                  |    |                  |    |                  |      |
|   | - rdkafka        |    | - RecordBatch    |    | - Catalog        |      |
|   | - Backpressure   |    | - DashMap Index  |    | - Object Store   |      |
|   | - Retry Logic    |    | - TTL Eviction   |    | - Atomic Commit  |      |
|   +------------------+    +------------------+    +------------------+      |
|            |                       |                       |               |
|            v                       v                       v               |
|   +---------------------------------------------------------------------+  |
|   |                      Transaction Log                                |  |
|   | - Append-only entries with CRC32 checksums                          |  |
|   | - Periodic checkpoints for fast recovery                            |  |
|   | - Idempotency records for exactly-once semantics                    |  |
|   +---------------------------------------------------------------------+  |
|                                                                             |
+-----------------------------------------------------------------------------+

4.2 Component Responsibilities

4.3 Technology Stack

K2I is implemented entirely in Rust, leveraging:

5. Core Components

5.1 Smart Kafka Consumer

The SmartKafkaConsumer wraps rdkafka with production-grade features:

5.1.1 Batch Processing

Messages are collected into batches before processing:

pub struct KafkaBatch {
    messages: Vec<KafkaMessage>,
    partition_offsets: HashMap<i32, i64>,
    timestamp_range: (DateTime<Utc>, DateTime<Utc>),
}

Batching amortizes processing overhead and enables efficient Arrow conversion.

5.1.2 Backpressure Control

When the hot buffer reaches capacity, the consumer pauses:

+-------------+     BufferFull     +-------------+
|  Consuming  |------------------>|   Paused    |
|             |                   |             |
|  (polling)  |<------------------|  (waiting)  |
+-------------+     BufferSpace    +-------------+

This prevents memory exhaustion while maintaining offset consistency.

5.1.3 Retry Strategy

Transient failures trigger exponential backoff:

Delay = min(max_delay, base_delay × 2^attempt) + jitter

Default configuration:
  base_delay:   100ms
  max_delay:    30s
  max_attempts: 10
  jitter:       0-10% random

5.2 Hot Buffer

The hot buffer provides sub-millisecond access to recent data.

5.2.1 Data Structure

pub struct HotBuffer {
    // Columnar storage
    schema: SchemaRef,
    builders: RwLock<ColumnBuilders>,
    records: RwLock<Vec<BufferedRecord>>,

    // O(1) lookup indexes
    key_index: DashMap<Vec<u8>, RowId>,
    offset_index: DashMap<(i32, i64), RowId>,

    // Management
    stats: BufferStats,
    config: BufferConfig,
}

5.2.2 Index Design

Both indexes use DashMap for lock-free concurrent access.

5.2.3 Flush Triggers

The buffer flushes when any condition is met:

5.3 Iceberg Writer

The Iceberg writer handles the critical path of durable persistence.

5.3.1 Write Pipeline

Arrow RecordBatch
        |
        v
+-------------------+
| Parquet Encoder   |
| - Row group: 128K |
| - Compression     |
| - Statistics      |
+-------------------+
        |
        v
+-------------------+
| Object Storage    |
| - Multipart upload|
| - Retry on failure|
| - Checksum verify |
+-------------------+
        |
        v
+-------------------+
| Catalog Commit    |
| - CAS semantics   |
| - Conflict retry  |
| - Snapshot create |
+-------------------+

5.3.2 File Naming Convention

{warehouse}/{database}/{table}/
  event_date={YYYY-MM-DD}/
    kafka_partition={N}/
      part-{uuid}-{offset_min}-{offset_max}.parquet

This structure enables partition pruning by date, Kafka partition isolation for replay, and offset range identification for debugging.

6. Data Flow & Processing Pipeline

6.1 Write Path

The write path processes messages from Kafka to the hot buffer:

Step 1: Poll
+-------------+
|   Kafka     |---- poll() -----> Vec<RawMessage>
+-------------+

Step 2: Batch
+-------------+
|   Batcher   |---- collect() --> KafkaBatch {messages, offsets}
+-------------+

Step 3: Convert
+-------------+
|   Arrow     |---- convert() --> RecordBatch
+-------------+

Step 4: Buffer
+-------------+
| Hot Buffer  |---- append() ---> Updated indexes
+-------------+

Step 5: Log
+-------------+
| Transaction |---- write() ----> OffsetMarker entry
|    Log      |
+-------------+

Latency: < 10ms end-to-end for buffering

6.2 Flush Path

The flush path persists buffered data to Iceberg. Latency: 200-800ms typical flush cycle

6.3 Query Path

The hot buffer supports immediate queries:

// By message key
buffer.query_by_key("user-123") -> Option<KafkaMessage>

// By partition and offset
buffer.query_by_offset(partition=0, offset=12345) -> Option<KafkaMessage>

// Range query
buffer.query_range(partition=0, start=12000, end=12100) -> Vec<KafkaMessage>

Latency: < 1ms for indexed lookups

7. Exactly-Once Semantics

K2I provides exactly-once delivery guarantees through coordinated transaction management.

7.1 The Challenge

Exactly-once requires atomicity across three systems:

1. Kafka: Consume and commit offsets
2. Object Storage: Upload files
3. Iceberg Catalog: Commit metadata

Failure at any point must not result in data loss or duplication.

7.2 Solution: Write-Ahead Logging + Idempotency

All operations are logged before execution:

FlushStart(batch_id=abc, rows=1000)
  └── If crash here: recover knows flush was attempted

ParquetWritten(file=part-001.parquet, size=50MB)
  └── If crash here: file exists but not committed

IcebergSnapshot(snapshot_id=123)
  └── If crash here: snapshot exists, mark complete

FlushComplete(batch_id=abc, snapshot=123)
  └── Safe to commit Kafka offset

7.3 Guarantees

8. Crash Recovery

K2I automatically recovers from crashes without data loss or duplication.

8.1 Recovery Process

Startup
    |
    v
+---------------------------------+
| 1. Load last checkpoint         |
|    - Last Kafka offsets         |
|    - Last Iceberg snapshot      |
+---------------------------------+
    |
    v
+---------------------------------+
| 2. Replay log since checkpoint  |
|    - Process each entry         |
|    - Track incomplete flushes   |
+---------------------------------+
    |
    v
+---------------------------------+
| 3. Execute cleanup              |
|    - Delete orphan files        |
|    - Log cleanup actions        |
+---------------------------------+
    |
    v
+---------------------------------+
| 4. Resume normal operation      |
|    - Seek to starting offsets   |
|    - Begin consuming            |
+---------------------------------+

9. Performance Characteristics

9.1 Latency Profile

9.2 Query Freshness

9.3 Throughput

9.3.1 Throughput Tuning

# High throughput configuration
[kafka]
batch_size = 5000
batch_timeout_ms = 1000

[buffer]
max_size_mb = 2000
flush_batch_size = 50000

[iceberg]
compression = "lz4"
target_file_size_mb = 256

9.4 Memory Consumption

10. Maintenance Operations

K2I includes automated maintenance to ensure long-term health.

10.1 Compaction

Compaction merges small files into larger ones:

Before:
├── part-001.parquet (50MB)
├── part-002.parquet (30MB)
├── part-003.parquet (40MB)
├── part-004.parquet (60MB)
└── part-005.parquet (20MB)

After compaction:
├── compacted-001.parquet (200MB)

10.1.1 Configuration

[maintenance]
compaction_enabled = true
compaction_interval_seconds = 3600    # Every hour
compaction_threshold_mb = 100         # Files smaller than this
compaction_target_mb = 512            # Target output size

10.2 Snapshot Expiration

Removes old snapshots to control metadata growth.

[maintenance]
snapshot_expiration_enabled = true
snapshot_retention_days = 7

10.3 Orphan File Cleanup

Removes unreferenced files from storage with safety measures:

• Retention buffer: Only delete files older than orphan_retention_days (default: 3)
• Reference check: Verify file not in any snapshot
• Modified time: Don't delete recently modified files

11. Catalog Integration

K2I supports multiple Iceberg catalog backends through a unified interface.

11.1 REST Catalog

[iceberg]
catalog_type = "rest"
rest_uri = "http://catalog.example.com:8181"

[iceberg.rest]
credential_type = "oauth2"
oauth2_token_endpoint = "https://auth.example.com/token"
oauth2_client_id = "k2i-client"

11.2 Hive Metastore

[iceberg]
catalog_type = "hive"
hive_metastore_uri = "thrift://hive-metastore:9083"

11.3 AWS Glue

[iceberg]
catalog_type = "glue"
aws_region = "us-east-1"

[iceberg.glue]
role_arn = "arn:aws:iam::123456789012:role/K2IRole"

11.4 Nessie

[iceberg]
catalog_type = "nessie"
nessie_uri = "http://nessie:19120/api/v1"

[iceberg.nessie]
default_branch = "main"

12. Observability

K2I provides comprehensive observability through metrics, health checks, and structured logging.

12.1 Prometheus Metrics

Counters

Gauges

12.2 Health Checks

Health Response Example

{
  "status": "healthy",
  "components": {
    "kafka": { "status": "healthy", "message": "Connected to 3 brokers" },
    "buffer": { "status": "healthy", "message": "45% capacity" },
    "iceberg": { "status": "healthy", "message": "Last commit 5s ago" }
  },
  "uptime_seconds": 3600,
  "version": "0.1.0"
}

13. Deployment Models

13.1 Docker

FROM ghcr.io/osodevops/k2i:latest

COPY config.toml /etc/k2i/config.toml

HEALTHCHECK --interval=30s --timeout=5s \
  CMD curl -f http://localhost:8080/healthz || exit 1

EXPOSE 8080 9090

ENTRYPOINT ["k2i", "ingest", "--config", "/etc/k2i/config.toml"]

13.2 Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: k2i
spec:
  replicas: 1  # Single process per partition set
  template:
    spec:
      containers:
      - name: k2i
        image: ghcr.io/osodevops/k2i:v0.1.0
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "2000m"
        ports:
        - containerPort: 8080
          name: health
        - containerPort: 9090
          name: metrics
        livenessProbe:
          httpGet:
            path: /healthz
            port: 8080
        readinessProbe:
          httpGet:
            path: /readyz
            port: 8080

13.3 Scaling Considerations

14. Comparison with Alternatives

14.1 Feature Matrix

14.2 Use Case Fit

14.3 Performance Comparison

15. Future Directions

15.1 Planned Features

15.2 Research Areas

• Adaptive flush triggers: ML-based flush timing optimization
• Tiered hot buffer: NVMe-backed overflow for larger buffers
• Distributed mode: Coordinated multi-process deployment
• Change data capture: Direct database CDC support

16. Conclusion

K2I represents a new approach to streaming data ingestion that prioritizes simplicity without sacrificing capability. By combining single-process architecture with intelligent buffering and comprehensive durability mechanisms, K2I delivers:

• Sub-second data freshness through in-memory hot buffering
• Cost-efficient storage via optimized Parquet files on object storage
• Data integrity with exactly-once semantics and crash recovery
• Operational simplicity through automated maintenance
• Production readiness with comprehensive observability

K2I is ideal for organizations seeking to bridge the gap between real-time streaming and batch analytics without the complexity of distributed stream processing frameworks.

17. References

1. Apache Iceberg Specification. https://iceberg.apache.org/spec/
2. Apache Arrow Columnar Format. https://arrow.apache.org/docs/format/Columnar.html
3. Apache Parquet Format. https://parquet.apache.org/docs/
4. Mooncake Labs - Moonlink. https://www.mooncake.dev/moonlink/
5. rdkafka - Rust Kafka Client. https://github.com/fede1024/rust-rdkafka
6. Iceberg Rust Implementation. https://github.com/apache/iceberg-rust