How adform/stream-loader works
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: FileBuilder.scala File builder · InDataOffsetBatchStorage.scala In-data offsets · RecordBatchingSink.scala RecordBatchingSink · TwoPhaseCommitBatchStorage.scala Two-phase commit
Components 27
Loader runtime
- StreamLoader
stream-loader-core/src/main/scala/com/adform/streamloader/StreamLoader.scala - KafkaSource
stream-loader-core/src/main/scala/com/adform/streamloader/source/KafkaSource.scala - RecordBatchingSink
stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatchingSink.scala - Partition grouping
stream-loader-core/src/main/scala/com/adform/streamloader/sink/PartitionGroupingSink.scala
Batching and files
- RecordBatcher
stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordBatcher.scala - Partitioned file batcher
stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/PartitioningFileRecordBatcher.scala - Record formatter
stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/RecordFormatter.scala - File builder
stream-loader-core/src/main/scala/com/adform/streamloader/sink/file/FileBuilder.scala
Encodings
- CSV encoding
stream-loader-core/src/main/scala/com/adform/streamloader/sink/encoding/csv/CsvRecordEncoder.scala - Parquet builders
stream-loader-hadoop/src/main/scala/com/adform/streamloader/hadoop/parquet/AvroParquetFileBuilder.scala - RowBinary encoding
stream-loader-clickhouse/src/main/scala/com/adform/streamloader/clickhouse/rowbinary/RowBinaryClickHouseFileBuilder.scala - Vertica encodings
stream-loader-vertica/src/main/scala/com/adform/streamloader/vertica/file/native/NativeVerticaFileBuilder.scala
Storage integrations
- S3 file storage
stream-loader-s3/src/main/scala/com/adform/streamloader/s3/S3FileStorage.scala - Hadoop file storage
stream-loader-hadoop/src/main/scala/com/adform/streamloader/hadoop/HadoopFileStorage.scala - Iceberg storage
stream-loader-iceberg/src/main/scala/com/adform/streamloader/iceberg/IcebergRecordBatchStorage.scala - ClickHouse storage
stream-loader-clickhouse/src/main/scala/com/adform/streamloader/clickhouse/ClickHouseFileStorage.scala - Vertica storage
stream-loader-vertica/src/main/scala/com/adform/streamloader/vertica/InRowOffsetVerticaFileStorage.scala
Offset coordination
- Batch storage contract
stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/RecordBatchStorage.scala - Two-phase commit
stream-loader-core/src/main/scala/com/adform/streamloader/sink/batch/storage/TwoPhaseCommitBatchStorage.scala - In-data offsets
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 28
- 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