Add storm-iceberg module: an Apache Iceberg sink bolt - #8950
Draft
GGraziadei wants to merge 7 commits into
Draft
Conversation
New external module writing Trident batches to Apache Iceberg tables directly from a topology, with exactly-once semantics. Each batch is committed in a single Iceberg transaction that atomically appends the data files and records the transaction id in the table property storm.trident.<topologyName>.<partitionIndex>.last-committed-txid, so replayed batches are detected and skipped even across worker crashes and commits whose outcome was unknown to the writer. Includes: - IcebergOptions: catalog properties passed verbatim to Iceberg's CatalogUtil.buildIcebergCatalog, so any catalog works with its standard keys; file format, target file size and table auto-creation. - RecordMapper with a field-name based default doing the standard primitive conversions; required columns without a value fail loudly. - Partitioned tables through Iceberg's fanout writer, one open data file per partition. - Per-batch table refresh, so schema and partition spec evolution is picked up without restarting the workers. - Metrics-v2 instrumentation: records written, data files and bytes committed, commit latency, commit failures, skipped replays. - A shutdown hook releasing the catalog client, since Trident's State has no lifecycle callback. - Documentation and two runnable example topologies, unpartitioned and partitioned.
…dent # Conflicts: # storm-dist/binary/final-package/src/main/assembly/binary.xml
…ingested per commit.
The Iceberg version was pinned in the module's own properties, where a project-wide dependency bump would not see it. It now lives in the root pom alongside hadoop.version, with the artifacts managed in dependencyManagement; storm-iceberg declares them without versions. This does not widen the dependency's reach. dependencyManagement fixes versions without adding dependencies, so modules that do not declare Iceberg still resolve none of it: storm-client and storm-server both show zero Iceberg artifacts. The direction also prevents it, since storm-iceberg depends on storm-client (provided) and nothing in core depends on storm-iceberg. Neither binary distribution bundles it either: like every other external/* connector, storm-iceberg ships only its README in both the full and the lite assembly, so Iceberg reaches only topologies that ask for it. The catalog implementations (iceberg-aws, -gcp, -hive, -nessie, ...) are deliberately absent from the managed set: they are supplied by the topology rather than pulled in transitively.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
Adds
storm-iceberg, a sink that writes tuples from a Storm topology directlyinto an Apache Iceberg table, with no Kafka Connect or Spark job in between.
The guarantee is atomic commits with at-least-once delivery. A batch becomes
visible in one Iceberg append or not at all, so readers never see a partial
batch; tuples are acked only after the commit containing them has landed, so a
crash costs orphan data files and a replay rather than lost rows. Duplicates are
possible on replay and are not removed — the sink is append-only and writes no
equality deletes — so they stay visible until something downstream dedupes.
Exactly-once is deliberately not claimed. It would require a deterministic
identity of the input, which comes from the source rather than the sink, and a
general-purpose module cannot assume every user has a replayable,
deterministically addressed source. An extractor SPI for sources that can do
better is a possible follow-up.
Commit atomicity is made recoverable by a write-ahead log: the data files are
made durable, an entry naming them is written under the table's own metadata
location with a freshly minted commit id, the files are appended in one
operation that stamps that id on the snapshot summary, and only then is the
entry cleared and the tuples acked. On startup a task asks the table whether
each pending commit landed, replaying only those that did not — the table
answers the question, so no identity from the source is needed.
Scope and packaging:
upserts and merge-on-read are out of scope.
remove_orphan_files,rewrite_data_files,expire_snapshots) is out of scope but documented as still necessary.external/*connector it ships only its README, in both the full and the lite assembly.
iceberg-aws,-gcp,-hive,-nessie, ...) aresupplied by the topology, not pulled in transitively.
dependencyManagement, so a project-wide dependency bump sees it.An earlier revision of this PR exposed a Trident
State. It is now a bolt,following feedback on the dev list: the original mail argued for tuple-at-a-time
control over what is durably persisted and then proposed a micro-batch API. The
catalog configuration, writer factory, partitioned fanout and commit logic
contain no Trident types and live in
org.apache.storm.iceberg.common.How was the change tested
34 tests, all green, plus checkstyle and PMD, on JDK 25
(
mvn verify -pl external/storm-iceberg,examples/storm-iceberg-examples):IcebergCommitterTestFieldNameRecordMapperTestIcebergBoltTestIcebergOptionsTestIcebergWriterTestCommitWalTestThe tests run against a real Iceberg table (
HadoopCatalogover a JUnit@TempDir), not mocks of Iceberg. What they cover:IcebergGenericsand asserted.appeared is replayed on startup; one that is already visible is dropped
rather than appended twice.
successful and its tuples acked; one that did not land clears its WAL entry
before failing, so the source's replay writes those rows exactly once. Both
are exercised with a real table and only the append made flaky.
the threshold commits and acks the whole batch, a tick tuple flushes a partial
batch, and a failed commit fails the buffered tuples instead of acking them.
empty-batch handling, buffered-byte accounting.
iceberg-commit-failuresincrement corresponds to replayed tuples.Not covered, and worth knowing before merge:
Glue and Nessie catalogs, and S3 / object-store
FileIO, are untested here.their default 10M-row scale on a real cluster.
optimistic retries.