# adform/stream-loader architecture

> Architecture diagram of the GitHub repository adform/stream-loader, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-09-27). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/adform/stream-loader
- Repository: https://github.com/adform/stream-loader

## Overview

Stream Loader is a Scala library for moving Kafka records into configurable storage sinks with exactly-once delivery goals. The principal workflow consumes records, groups and batches them, formats and encodes files where applicable, then commits data and stream positions using storage-specific strategies. The graph retains the documented CSV, Parquet, Vertica-native, and ClickHouse RowBinary formats, plus the major storage integrations. Some core wiring and unsampled integrations are shown only where the README documents the capability; implementation-level links are limited by the excerpts provided.

Main entry points: [`stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/FileBuilder.scala`](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/FileBuilder.scala) (File builder), [`stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/InDataOffsetBatchStorage.scala`](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/InDataOffsetBatchStorage.scala) (In-data offsets), [`stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatchingSink.scala`](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatchingSink.scala) (RecordBatchingSink), [`stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/TwoPhaseCommitBatchStorage.scala`](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/TwoPhaseCommitBatchStorage.scala) (Two-phase commit)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_runtime["Loader runtime"]
  node_streamloader["StreamLoader<br/>[StreamLoader.scala]"]
  node_kafkasource["KafkaSource<br/>[KafkaSource.scala]"]
  node_sink["RecordBatchingSink"]
  node_partitiongroups["Partition grouping"]
end

subgraph group_batch["Batching and files"]
  node_batcher["RecordBatcher"]
  node_filebatcher["Partitioned file batcher"]
  node_formatter["Record formatter"]
  node_filebuilder["File builder<br/>[FileBuilder.scala]"]
end

subgraph group_formats["Encodings"]
  node_csv["CSV encoding"]
  node_parquet["Parquet builders"]
  node_clickhouseformat["RowBinary encoding"]
  node_verticaformat["Vertica encodings"]
end

subgraph group_storage["Storage integrations"]
  node_s3storage["S3 file storage"]
  node_hadoopstorage["Hadoop file storage"]
  node_icebergstorage["Iceberg storage"]
  node_clickhousestorage["ClickHouse storage"]
  node_verticastorage["Vertica storage"]
end

subgraph group_coordination["Offset coordination"]
  node_batchstorage["Batch storage contract"]
  node_twophase["Two-phase commit"]
  node_indata["In-data offsets"]
end

node_application(("Loader application"))
node_kafka[("Kafka cluster")]
node_s3[("S3-compatible service")]
node_hdfs[("Hadoop filesystem")]
node_iceberg[("Iceberg table")]
node_clickhouse[("ClickHouse")]
node_vertica[("Vertica")]

node_application -->|"starts"| node_streamloader
node_streamloader -->|"consumes from"| node_kafkasource
node_streamloader -->|"sends records"| node_sink
node_kafkasource -->|"polls"| node_kafka
node_sink -->|"groups records"| node_partitiongroups
node_sink -->|"batches records"| node_batcher
node_batcher -->|"uses for files"| node_filebatcher
node_filebatcher -->|"formats records"| node_formatter
node_filebatcher -->|"builds files"| node_filebuilder
node_filebuilder -.->|"can use"| node_csv
node_filebuilder -.->|"can use"| node_parquet
node_filebuilder -.->|"can use"| node_clickhouseformat
node_filebuilder -.->|"can use"| node_verticaformat
node_sink -->|"commits batches"| node_batchstorage
node_batchstorage -.->|"implemented by"| node_twophase
node_batchstorage -.->|"implemented by"| node_indata
node_twophase -.->|"implemented by"| node_s3storage
node_twophase -.->|"implemented by"| node_hadoopstorage
node_s3storage -->|"uploads files"| node_s3
node_hadoopstorage -->|"stages and stores"| node_hdfs
node_indata -.->|"implemented by"| node_icebergstorage
node_indata -.->|"implemented by"| node_clickhousestorage
node_indata -.->|"implemented by"| node_verticastorage
node_icebergstorage -->|"appends data and offsets"| node_iceberg
node_clickhousestorage -->|"inserts and queries"| node_clickhouse
node_verticastorage -->|"writes and queries"| node_vertica
node_twophase -->|"stages and commits offsets"| node_kafka
node_indata -->|"commits offsets"| node_kafka

click node_streamloader "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/StreamLoader.scala"
click node_kafkasource "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/source/KafkaSource.scala"
click node_sink "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatchingSink.scala"
click node_partitiongroups "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/PartitionGroupingSink.scala"
click node_batcher "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatcher.scala"
click node_filebatcher "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/PartitioningFileRecordBatcher.scala"
click node_formatter "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordFormatter.scala"
click node_filebuilder "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/FileBuilder.scala"
click node_csv "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/encoding/csv/CsvRecordEncoder.scala"
click node_parquet "https://github.com/adform/stream-loader/blob/master/stream-loader-hadoop/src/main/scala/com/adform/streamloader/hadoop/parquet/AvroParquetFileBuilder.scala"
click node_clickhouseformat "https://github.com/adform/stream-loader/blob/master/stream-loader-clickhouse/src/main/scala/com/adform/streamloader/clickhouse/rowbinary/RowBinaryClickHouseFileBuilder.scala"
click node_verticaformat "https://github.com/adform/stream-loader/blob/master/stream-loader-vertica/src/main/scala/com/adform/streamloader/vertica/file/native/NativeVerticaFileBuilder.scala"
click node_batchstorage "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/RecordBatchStorage.scala"
click node_twophase "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/TwoPhaseCommitBatchStorage.scala"
click node_indata "https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/InDataOffsetBatchStorage.scala"
click node_s3storage "https://github.com/adform/stream-loader/blob/master/stream-loader-s3/src/main/scala/com/adform/streamloader/s3/S3FileStorage.scala"
click node_hadoopstorage "https://github.com/adform/stream-loader/blob/master/stream-loader-hadoop/src/main/scala/com/adform/streamloader/hadoop/HadoopFileStorage.scala"
click node_icebergstorage "https://github.com/adform/stream-loader/blob/master/stream-loader-iceberg/src/main/scala/com/adform/streamloader/iceberg/IcebergRecordBatchStorage.scala"
click node_clickhousestorage "https://github.com/adform/stream-loader/blob/master/stream-loader-clickhouse/src/main/scala/com/adform/streamloader/clickhouse/ClickHouseFileStorage.scala"
click node_verticastorage "https://github.com/adform/stream-loader/blob/master/stream-loader-vertica/src/main/scala/com/adform/streamloader/vertica/InRowOffsetVerticaFileStorage.scala"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_streamloader,node_kafkasource,node_sink,node_partitiongroups toneBlue
class node_batcher,node_filebatcher,node_formatter,node_filebuilder,node_kafka,node_s3,node_hdfs,node_iceberg,node_clickhouse,node_vertica toneAmber
class node_csv,node_parquet,node_clickhouseformat,node_verticaformat toneMint
class node_s3storage,node_hadoopstorage,node_icebergstorage,node_clickhousestorage,node_verticastorage toneRose
class node_batchstorage,node_twophase,node_indata,node_application toneIndigo
```

## Components

### Loader runtime

- [StreamLoader](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/StreamLoader.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/StreamLoader.scala`
- [KafkaSource](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/source/KafkaSource.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/source/KafkaSource.scala`
- [RecordBatchingSink](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatchingSink.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatchingSink.scala`
- [Partition grouping](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/PartitionGroupingSink.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/PartitionGroupingSink.scala`

### Batching and files

- [RecordBatcher](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatcher.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatcher.scala`
- [Partitioned file batcher](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/PartitioningFileRecordBatcher.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/PartitioningFileRecordBatcher.scala`
- [Record formatter](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordFormatter.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordFormatter.scala`
- [File builder](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/FileBuilder.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/FileBuilder.scala`

### Encodings

- [CSV encoding](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/encoding/csv/CsvRecordEncoder.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/encoding/csv/CsvRecordEncoder.scala`
- [Parquet builders](https://github.com/adform/stream-loader/blob/master/stream-loader-hadoop/src/main/scala/com/adform/streamloader/hadoop/parquet/AvroParquetFileBuilder.scala): `stream-loader-hadoop/src/main/scala/com/adform/streamloader/hadoop/parquet/AvroParquetFileBuilder.scala`
- [RowBinary encoding](https://github.com/adform/stream-loader/blob/master/stream-loader-clickhouse/src/main/scala/com/adform/streamloader/clickhouse/rowbinary/RowBinaryClickHouseFileBuilder.scala): `stream-loader-clickhouse/src/main/scala/com/adform/streamloader/clickhouse/rowbinary/RowBinaryClickHouseFileBuilder.scala`
- [Vertica encodings](https://github.com/adform/stream-loader/blob/master/stream-loader-vertica/src/main/scala/com/adform/streamloader/vertica/file/native/NativeVerticaFileBuilder.scala): `stream-loader-vertica/src/main/scala/com/adform/streamloader/vertica/file/native/NativeVerticaFileBuilder.scala`

### Storage integrations

- [S3 file storage](https://github.com/adform/stream-loader/blob/master/stream-loader-s3/src/main/scala/com/adform/streamloader/s3/S3FileStorage.scala): `stream-loader-s3/src/main/scala/com/adform/streamloader/s3/S3FileStorage.scala`
- [Hadoop file storage](https://github.com/adform/stream-loader/blob/master/stream-loader-hadoop/src/main/scala/com/adform/streamloader/hadoop/HadoopFileStorage.scala): `stream-loader-hadoop/src/main/scala/com/adform/streamloader/hadoop/HadoopFileStorage.scala`
- [Iceberg storage](https://github.com/adform/stream-loader/blob/master/stream-loader-iceberg/src/main/scala/com/adform/streamloader/iceberg/IcebergRecordBatchStorage.scala): `stream-loader-iceberg/src/main/scala/com/adform/streamloader/iceberg/IcebergRecordBatchStorage.scala`
- [ClickHouse storage](https://github.com/adform/stream-loader/blob/master/stream-loader-clickhouse/src/main/scala/com/adform/streamloader/clickhouse/ClickHouseFileStorage.scala): `stream-loader-clickhouse/src/main/scala/com/adform/streamloader/clickhouse/ClickHouseFileStorage.scala`
- [Vertica storage](https://github.com/adform/stream-loader/blob/master/stream-loader-vertica/src/main/scala/com/adform/streamloader/vertica/InRowOffsetVerticaFileStorage.scala): `stream-loader-vertica/src/main/scala/com/adform/streamloader/vertica/InRowOffsetVerticaFileStorage.scala`

### Offset coordination

- [Batch storage contract](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/RecordBatchStorage.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/RecordBatchStorage.scala`
- [Two-phase commit](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/TwoPhaseCommitBatchStorage.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/TwoPhaseCommitBatchStorage.scala`
- [In-data offsets](https://github.com/adform/stream-loader/blob/master/stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/InDataOffsetBatchStorage.scala): `stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/InDataOffsetBatchStorage.scala`

### Other components

- **Loader application**
- **Kafka cluster**
- **S3-compatible service**
- **Hadoop filesystem**
- **Iceberg table**
- **ClickHouse**
- **Vertica**

## Connections

- Loader application → StreamLoader: starts
- StreamLoader → KafkaSource: consumes from
- StreamLoader → RecordBatchingSink: sends records
- KafkaSource → Kafka cluster: polls
- RecordBatchingSink → Partition grouping: groups records
- RecordBatchingSink → RecordBatcher: batches records
- RecordBatcher → Partitioned file batcher: uses for files
- Partitioned file batcher → Record formatter: formats records
- Partitioned file batcher → File builder: builds files
- File builder → CSV encoding: can use
- File builder → Parquet builders: can use
- File builder → RowBinary encoding: can use
- File builder → Vertica encodings: can use
- RecordBatchingSink → Batch storage contract: commits batches
- Batch storage contract → Two-phase commit: implemented by
- Batch storage contract → In-data offsets: implemented by
- Two-phase commit → S3 file storage: implemented by
- Two-phase commit → Hadoop file storage: implemented by
- S3 file storage → S3-compatible service: uploads files
- Hadoop file storage → Hadoop filesystem: stages and stores
- In-data offsets → Iceberg storage: implemented by
- In-data offsets → ClickHouse storage: implemented by
- In-data offsets → Vertica storage: implemented by
- Iceberg storage → Iceberg table: appends data and offsets
- ClickHouse storage → ClickHouse: inserts and queries
- Vertica storage → Vertica: writes and queries
- Two-phase commit → Kafka cluster: stages and commits offsets
- In-data offsets → Kafka cluster: commits offsets