From b5c15ac671775554d05663d3ad42a2590bdc74a1 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 26 Jul 2026 00:48:25 +0200 Subject: [PATCH 1/4] Add storm-iceberg Trident sink 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...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. --- docs/storm-iceberg.md | 144 +++++ examples/pom.xml | 1 + examples/storm-iceberg-examples/pom.xml | 99 ++++ ...bergPartitionedTridentExampleTopology.java | 100 ++++ .../IcebergTridentExampleTopology.java | 81 +++ external/pom.xml | 1 + external/storm-iceberg/README.md | 76 +++ external/storm-iceberg/pom.xml | 109 ++++ .../trident/FieldNameRecordMapper.java | 99 ++++ .../storm/iceberg/trident/IcebergOptions.java | 150 ++++++ .../storm/iceberg/trident/IcebergState.java | 289 ++++++++++ .../iceberg/trident/IcebergStateFactory.java | 46 ++ .../iceberg/trident/IcebergStateMetrics.java | 82 +++ .../iceberg/trident/IcebergStateUpdater.java | 37 ++ .../trident/PartitionedRecordWriter.java | 54 ++ .../storm/iceberg/trident/RecordMapper.java | 40 ++ .../trident/FieldNameRecordMapperTest.java | 143 +++++ .../iceberg/trident/IcebergOptionsTest.java | 88 +++ .../trident/IcebergStateFactoryTest.java | 101 ++++ .../iceberg/trident/IcebergStateTest.java | 504 ++++++++++++++++++ .../trident/RecordingMetricsContext.java | 102 ++++ .../src/main/assembly/binary.xml | 7 + 22 files changed, 2353 insertions(+) create mode 100644 docs/storm-iceberg.md create mode 100644 examples/storm-iceberg-examples/pom.xml create mode 100644 examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedTridentExampleTopology.java create mode 100644 examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergTridentExampleTopology.java create mode 100644 external/storm-iceberg/README.md create mode 100644 external/storm-iceberg/pom.xml create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/FieldNameRecordMapper.java create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergOptions.java create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergState.java create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateFactory.java create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateMetrics.java create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateUpdater.java create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/PartitionedRecordWriter.java create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/RecordMapper.java create mode 100644 external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/FieldNameRecordMapperTest.java create mode 100644 external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergOptionsTest.java create mode 100644 external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateFactoryTest.java create mode 100644 external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateTest.java create mode 100644 external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/RecordingMetricsContext.java diff --git a/docs/storm-iceberg.md b/docs/storm-iceberg.md new file mode 100644 index 00000000000..362f0f88347 --- /dev/null +++ b/docs/storm-iceberg.md @@ -0,0 +1,144 @@ +--- +title: Storm Apache Iceberg Integration +layout: documentation +documentation: true +--- + +Trident state implementation for writing data to [Apache Iceberg](https://iceberg.apache.org/) tables +directly from a Storm topology — no Kafka Connect or Spark job in between — with exactly-once semantics. + +## Usage + +```java +Map catalogProps = new HashMap<>(); +catalogProps.put("type", "rest"); +catalogProps.put("uri", "http://rest-catalog:8181"); + +IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events") + .build(); + +TridentTopology topology = new TridentTopology(); +topology.newStream("spout", spout) + .partitionPersist(new IcebergStateFactory(options), + new Fields("id", "name", "ts"), + new IcebergStateUpdater()); +``` + +The catalog properties are passed verbatim to Iceberg's `CatalogUtil.buildIcebergCatalog(...)`, +so every Iceberg catalog works with its standard configuration keys: `type` = `hive`, `hadoop`, +`rest`, or `catalog-impl` for Glue, Nessie, JDBC, etc. + +### Options + +| Option | Default | Description | +|---|---|---| +| `withCatalogProperties(Map)` | required | Iceberg catalog configuration | +| `withTable(String)` | required | Target table identifier, e.g. `db.events` | +| `withRecordMapper(RecordMapper)` | `FieldNameRecordMapper` | Tuple → `Record` conversion | +| `withFileFormat(FileFormat)` | `PARQUET` | Data file format | +| `withTargetFileSizeBytes(long)` | table property `write.target-file-size-bytes` | Rolling file size | +| `withAutoCreate(Schema, PartitionSpec)` | disabled | Create the table on first use if missing | + +### Tuple mapping + +By default tuple fields are matched to table columns **by name** (`FieldNameRecordMapper`): +numeric values are widened to the column type, `Instant` / `java.util.Date` / epoch-millis +`Long` values are converted for timestamp columns, `byte[]` becomes `ByteBuffer`. A required +column with no tuple value fails the topology loudly. For anything custom (structs, lists, +renames), implement `RecordMapper`: + +```java +public interface RecordMapper extends Serializable { + Record map(TridentTuple tuple, Schema schema); +} +``` + +### Partitioned tables + +Partitioned tables need no extra configuration: the state derives each record's partition from +the table's current `PartitionSpec` and keeps one open data file per partition (Iceberg's fanout +writer), so a single batch can span any number of partitions. + +```java +Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "region", Types.StringType.get()), + Types.NestedField.required(3, "event_time", Types.TimestampType.withZone())); + +PartitionSpec spec = PartitionSpec.builderFor(schema) + .identity("region") + .day("event_time") + .build(); + +IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events") + .withAutoCreate(schema, spec) + .build(); +``` + +The `PartitionSpec` above is only used when the table is auto-created; for an existing table the +spec stored in the catalog wins. Note that a batch spread over many partitions opens many files +at once — partition the stream on the partition columns (`partitionBy`) to keep file counts down. + +### Metrics + +The state registers these metrics-v2 metrics per `partitionPersist` task, so they show up in +whatever reporter the cluster is configured with: + +| Metric | Type | Meaning | +|---|---|---| +| `iceberg-records-written` | counter | Records handed to the Iceberg writer | +| `iceberg-data-files-committed` | counter | Data files made visible by commits | +| `iceberg-bytes-committed` | counter | Bytes in those data files | +| `iceberg-commit-latency` | timer | Duration of the Iceberg commit transaction | +| `iceberg-commit-failures` | counter | Commits that failed (including unknown state) | +| `iceberg-batches-skipped` | counter | Replayed batches skipped as already committed | + +A steadily rising `iceberg-data-files-committed` with a flat `iceberg-bytes-committed` is the +small-files signature: consider fewer, larger batches or a compaction job. + +### Table refresh + +Each batch refreshes the table metadata before writing, so schema and partition spec evolution +performed outside the topology is picked up without restarting the workers. This costs one +catalog round-trip per batch on top of the commit's own. + +### Resource cleanup + +Trident's `State` has no shutdown callback, so the state registers a JVM shutdown hook to close +the catalog (REST/HTTP client, Hive metastore connection, object store client) when the worker +exits. + +## Examples + +`examples/storm-iceberg-examples` contains two runnable topologies writing to a local Hadoop +catalog: `IcebergTridentExampleTopology` (unpartitioned) and +`IcebergPartitionedTridentExampleTopology` (partitioned by `identity(region)` and +`days(event_time)`). + +## Exactly-once semantics + +Each Trident batch is committed to Iceberg with a single atomic `Transaction` that both appends +the batch's data files and records the transaction id in the table property +`storm.trident...last-committed-txid`. When Trident replays a +batch whose txid is already recorded, the state skips it. Because the txid marker and the data +commit are atomic, this stays correct across worker crashes, replays, and commits whose outcome +was unknown to the writer. + +### Caveats + +- Exactly-once requires a **transactional spout** (the same txid must carry the same batch + content on replay), exactly as for `HdfsState`. +- **Resubmitting a topology from scratch** (fresh transactional state in ZooKeeper) restarts + Trident txids from low values, so markers from the previous run would wrongly skip batches. + Resubmit under a new topology name, or clear the `storm.trident..*` table + properties first. +- Each state partition performs its own Iceberg commit per batch. Iceberg resolves concurrent + append commits with optimistic retries (tune with the table property + `commit.retry.num-retries`), but beyond roughly 10–20 partitions consider reducing the + parallelism of the `partitionPersist`. +- Batches that fail before commit can leave never-committed (invisible) data files behind. + Standard Iceberg maintenance (`remove_orphan_files`) cleans them up. diff --git a/examples/pom.xml b/examples/pom.xml index 03517afedce..e596c88c597 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -36,6 +36,7 @@ storm-kafka-client-examples storm-jdbc-examples storm-hdfs-examples + storm-iceberg-examples storm-jms-examples storm-perf diff --git a/examples/storm-iceberg-examples/pom.xml b/examples/storm-iceberg-examples/pom.xml new file mode 100644 index 00000000000..e1808076dea --- /dev/null +++ b/examples/storm-iceberg-examples/pom.xml @@ -0,0 +1,99 @@ + + + + 4.0.0 + + + storm-examples + org.apache.storm + 3.0.0-SNAPSHOT + ../pom.xml + + + storm-iceberg-examples + + + + Storm Iceberg Examples + + + org.apache.storm + storm-client + ${project.version} + ${provided.scope} + + + org.apache.storm + storm-iceberg + ${project.version} + + + + + + + org.apache.maven.plugins + maven-shade-plugin + + true + + + *:* + + META-INF/*.SF + META-INF/*.sf + META-INF/*.DSA + META-INF/*.dsa + META-INF/*.RSA + META-INF/*.rsa + META-INF/*.EC + META-INF/*.ec + META-INF/MSFTSIG.SF + META-INF/MSFTSIG.RSA + + + + + + + package + + shade + + + + + + + + + + + + + org.apache.maven.plugins + maven-checkstyle-plugin + + + + org.apache.maven.plugins + maven-pmd-plugin + + + + diff --git a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedTridentExampleTopology.java b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedTridentExampleTopology.java new file mode 100644 index 00000000000..2725244fa1c --- /dev/null +++ b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedTridentExampleTopology.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.examples; + +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.HashMap; +import java.util.Map; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.types.Types; +import org.apache.storm.Config; +import org.apache.storm.StormSubmitter; +import org.apache.storm.iceberg.trident.IcebergOptions; +import org.apache.storm.iceberg.trident.IcebergStateFactory; +import org.apache.storm.iceberg.trident.IcebergStateUpdater; +import org.apache.storm.trident.TridentTopology; +import org.apache.storm.trident.testing.FixedBatchSpout; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Values; + +/** + * Writes a small fixed stream into a partitioned Iceberg table on a local Hadoop catalog. + * + *

The table is partitioned by {@code identity(region)} and {@code days(event_time)}, so a single + * batch spans several partitions and the state opens one data file per partition through the + * Iceberg fanout writer. + * + *

Run locally with: + * {@code storm local storm-iceberg-examples-*.jar + * org.apache.storm.iceberg.examples.IcebergPartitionedTridentExampleTopology} + * then inspect the warehouse directory (default {@code file:///tmp/storm-iceberg-warehouse}): the + * data files are laid out under {@code example/events/data/region=.../event_time_day=...}. + */ +public final class IcebergPartitionedTridentExampleTopology { + + private IcebergPartitionedTridentExampleTopology() { + } + + public static void main(String[] args) throws Exception { + String warehouse = args.length > 0 ? args[0] : "file:///tmp/storm-iceberg-warehouse"; + String topologyName = args.length > 1 ? args[1] : "iceberg-partitioned-example"; + + Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "region", Types.StringType.get()), + Types.NestedField.required(3, "event_time", Types.TimestampType.withZone())); + + PartitionSpec spec = PartitionSpec.builderFor(schema) + .identity("region") + .day("event_time") + .build(); + + Map catalogProps = new HashMap<>(); + catalogProps.put(CatalogUtil.ICEBERG_CATALOG_TYPE, CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP); + catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse); + + IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("example.events") + .withAutoCreate(schema, spec) + .build(); + + // Two regions x two days -> four partitions, written by a single state instance. + Instant today = Instant.now(); + Instant yesterday = today.minus(1, ChronoUnit.DAYS); + FixedBatchSpout spout = new FixedBatchSpout(new Fields("id", "region", "event_time"), 3, + new Values(1L, "eu-west", yesterday), new Values(2L, "us-east", yesterday), + new Values(3L, "eu-west", today), new Values(4L, "us-east", today), + new Values(5L, "eu-west", today), new Values(6L, "us-east", yesterday)); + spout.setCycle(false); + + TridentTopology topology = new TridentTopology(); + topology.newStream("events", spout) + .partitionPersist(new IcebergStateFactory(options), + new Fields("id", "region", "event_time"), new IcebergStateUpdater()); + + Config conf = new Config(); + conf.setMaxSpoutPending(3); + StormSubmitter.submitTopology(topologyName, conf, topology.build()); + } +} diff --git a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergTridentExampleTopology.java b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergTridentExampleTopology.java new file mode 100644 index 00000000000..ecd45dafb7b --- /dev/null +++ b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergTridentExampleTopology.java @@ -0,0 +1,81 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.examples; + +import java.util.HashMap; +import java.util.Map; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.types.Types; +import org.apache.storm.Config; +import org.apache.storm.StormSubmitter; +import org.apache.storm.iceberg.trident.IcebergOptions; +import org.apache.storm.iceberg.trident.IcebergStateFactory; +import org.apache.storm.iceberg.trident.IcebergStateUpdater; +import org.apache.storm.trident.TridentTopology; +import org.apache.storm.trident.testing.FixedBatchSpout; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Values; + +/** + * Writes a small fixed stream into an Iceberg table on a local Hadoop catalog. + * + *

Run locally with: + * {@code storm local storm-iceberg-examples-*.jar org.apache.storm.iceberg.examples.IcebergTridentExampleTopology} + * then inspect the warehouse directory (default {@code file:///tmp/storm-iceberg-warehouse}). + */ +public final class IcebergTridentExampleTopology { + + private IcebergTridentExampleTopology() { + } + + public static void main(String[] args) throws Exception { + String warehouse = args.length > 0 ? args[0] : "file:///tmp/storm-iceberg-warehouse"; + String topologyName = args.length > 1 ? args[1] : "iceberg-example"; + + Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "word", Types.StringType.get())); + + Map catalogProps = new HashMap<>(); + catalogProps.put(CatalogUtil.ICEBERG_CATALOG_TYPE, CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP); + catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse); + + IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("example.words") + .withAutoCreate(schema, PartitionSpec.unpartitioned()) + .build(); + + FixedBatchSpout spout = new FixedBatchSpout(new Fields("id", "word"), 3, + new Values(1L, "storm"), new Values(2L, "iceberg"), new Values(3L, "trident"), + new Values(4L, "lakehouse"), new Values(5L, "parquet"), new Values(6L, "snapshot")); + spout.setCycle(false); + + TridentTopology topology = new TridentTopology(); + topology.newStream("words", spout) + .partitionPersist(new IcebergStateFactory(options), new Fields("id", "word"), new IcebergStateUpdater()); + + Config conf = new Config(); + conf.setMaxSpoutPending(3); + StormSubmitter.submitTopology(topologyName, conf, topology.build()); + } +} diff --git a/external/pom.xml b/external/pom.xml index 2b209673f2d..9f3e645e895 100644 --- a/external/pom.xml +++ b/external/pom.xml @@ -35,6 +35,7 @@ storm-hdfs storm-hdfs-blobstore storm-hdfs-oci + storm-iceberg storm-jdbc storm-jms storm-kafka-client diff --git a/external/storm-iceberg/README.md b/external/storm-iceberg/README.md new file mode 100644 index 00000000000..4d1eecf86ae --- /dev/null +++ b/external/storm-iceberg/README.md @@ -0,0 +1,76 @@ +# Storm Iceberg + +Trident state implementation for writing data to [Apache Iceberg](https://iceberg.apache.org/) tables +directly from a Storm topology — no Kafka Connect or Spark job in between — with exactly-once semantics. + +## Usage + +```java +Map catalogProps = new HashMap<>(); +catalogProps.put("type", "rest"); +catalogProps.put("uri", "http://rest-catalog:8181"); + +IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events") + .build(); + +TridentTopology topology = new TridentTopology(); +topology.newStream("spout", spout) + .partitionPersist(new IcebergStateFactory(options), + new Fields("id", "name", "ts"), + new IcebergStateUpdater()); +``` + +The catalog properties are passed verbatim to Iceberg's `CatalogUtil.buildIcebergCatalog(...)`, +so every Iceberg catalog works with its standard configuration keys: `type` = `hive`, `hadoop`, +`rest`, or `catalog-impl` for Glue, Nessie, JDBC, etc. + +### Options + +| Option | Default | Description | +|---|---|---| +| `withCatalogProperties(Map)` | required | Iceberg catalog configuration | +| `withTable(String)` | required | Target table identifier, e.g. `db.events` | +| `withRecordMapper(RecordMapper)` | `FieldNameRecordMapper` | Tuple → `Record` conversion | +| `withFileFormat(FileFormat)` | `PARQUET` | Data file format | +| `withTargetFileSizeBytes(long)` | table property `write.target-file-size-bytes` | Rolling file size | +| `withAutoCreate(Schema, PartitionSpec)` | disabled | Create the table on first use if missing | + +### Tuple mapping + +By default tuple fields are matched to table columns **by name** (`FieldNameRecordMapper`): +numeric values are widened to the column type, `Instant` / `java.util.Date` / epoch-millis +`Long` values are converted for timestamp columns, `byte[]` becomes `ByteBuffer`. A required +column with no tuple value fails the topology loudly. For anything custom (structs, lists, +renames), implement `RecordMapper`: + +```java +public interface RecordMapper extends Serializable { + Record map(TridentTuple tuple, Schema schema); +} +``` + +## Exactly-once semantics + +Each Trident batch is committed to Iceberg with a single atomic `Transaction` that both appends +the batch's data files and records the transaction id in the table property +`storm.trident...last-committed-txid`. When Trident replays a +batch whose txid is already recorded, the state skips it. Because the txid marker and the data +commit are atomic, this stays correct across worker crashes, replays, and commits whose outcome +was unknown to the writer. + +### Caveats + +- Exactly-once requires a **transactional spout** (the same txid must carry the same batch + content on replay), exactly as for `HdfsState`. +- **Resubmitting a topology from scratch** (fresh transactional state in ZooKeeper) restarts + Trident txids from low values, so markers from the previous run would wrongly skip batches. + Resubmit under a new topology name, or clear the `storm.trident..*` table + properties first. +- Each state partition performs its own Iceberg commit per batch. Iceberg resolves concurrent + append commits with optimistic retries (tune with the table property + `commit.retry.num-retries`), but beyond roughly 10–20 partitions consider reducing the + parallelism of the `partitionPersist`. +- Batches that fail before commit can leave never-committed (invisible) data files behind. + Standard Iceberg maintenance (`remove_orphan_files`) cleans them up. diff --git a/external/storm-iceberg/pom.xml b/external/storm-iceberg/pom.xml new file mode 100644 index 00000000000..2b29aed834f --- /dev/null +++ b/external/storm-iceberg/pom.xml @@ -0,0 +1,109 @@ + + + + 4.0.0 + + + storm-external + org.apache.storm + 3.0.0-SNAPSHOT + ../pom.xml + + + storm-iceberg + + Storm Iceberg + + + 1.11.0 + + + + + org.apache.storm + storm-client + ${project.version} + ${provided.scope} + + + org.apache.iceberg + iceberg-api + ${iceberg.version} + + + org.apache.iceberg + iceberg-core + ${iceberg.version} + + + org.apache.iceberg + iceberg-data + ${iceberg.version} + + + org.apache.iceberg + iceberg-parquet + ${iceberg.version} + + + + org.apache.iceberg + iceberg-orc + ${iceberg.version} + + + org.apache.hadoop + hadoop-client-api + ${hadoop.version} + + + org.apache.hadoop + hadoop-client-runtime + ${hadoop.version} + + + org.mockito + mockito-junit-jupiter + test + + + + + + org.apache.maven.plugins + maven-jar-plugin + + + + test-jar + + + + + + org.apache.maven.plugins + maven-checkstyle-plugin + + + + org.apache.maven.plugins + maven-pmd-plugin + + + + diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/FieldNameRecordMapper.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/FieldNameRecordMapper.java new file mode 100644 index 00000000000..ca39dee59e5 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/FieldNameRecordMapper.java @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import java.nio.ByteBuffer; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.Date; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.storm.trident.tuple.TridentTuple; + +/** + * Default {@link RecordMapper}: matches tuple fields to Iceberg columns by name. + * + *

Standard primitive conversions are applied (numeric widening, {@link Instant}/epoch-millis to + * timestamp types, byte[] to ByteBuffer). Values of other types — including java.time values that + * already match the Iceberg representation, and struct/list/map values — are passed through as-is. + * A required column with no tuple value fails fast with {@link IllegalArgumentException}. + */ +public class FieldNameRecordMapper implements RecordMapper { + + private static final long serialVersionUID = 1L; + + @Override + public Record map(TridentTuple tuple, Schema schema) { + GenericRecord record = GenericRecord.create(schema); + for (Types.NestedField field : schema.columns()) { + Object value = tuple.contains(field.name()) ? tuple.getValueByField(field.name()) : null; + if (value == null) { + if (field.isRequired()) { + throw new IllegalArgumentException( + "Tuple has no value for required Iceberg column '" + field.name() + "'"); + } + continue; + } + record.setField(field.name(), convert(field.type(), value)); + } + return record; + } + + private Object convert(Type type, Object value) { + switch (type.typeId()) { + case INTEGER: + return value instanceof Number ? ((Number) value).intValue() : value; + case LONG: + return value instanceof Number ? ((Number) value).longValue() : value; + case FLOAT: + return value instanceof Number ? ((Number) value).floatValue() : value; + case DOUBLE: + return value instanceof Number ? ((Number) value).doubleValue() : value; + case STRING: + return value instanceof CharSequence ? value.toString() : value; + case TIMESTAMP: + return convertTimestamp((Types.TimestampType) type, value); + case BINARY: + return value instanceof byte[] ? ByteBuffer.wrap((byte[]) value) : value; + default: + return value; + } + } + + private Object convertTimestamp(Types.TimestampType type, Object value) { + Instant instant; + if (value instanceof Instant) { + instant = (Instant) value; + } else if (value instanceof Date) { + instant = ((Date) value).toInstant(); + } else if (value instanceof Long) { + instant = Instant.ofEpochMilli((Long) value); + } else { + return value; // already an OffsetDateTime / LocalDateTime + } + return type.shouldAdjustToUTC() + ? OffsetDateTime.ofInstant(instant, ZoneOffset.UTC) + : LocalDateTime.ofInstant(instant, ZoneOffset.UTC); + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergOptions.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergOptions.java new file mode 100644 index 00000000000..1a3f978af89 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergOptions.java @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import java.io.Serial; +import java.io.Serializable; +import java.util.HashMap; +import java.util.Map; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; + +/** + * Serializable configuration for {@link IcebergState}. + * + *

The catalog properties are passed verbatim to + * {@code CatalogUtil.buildIcebergCatalog(...)}, so any Iceberg catalog (hive, hadoop, rest, + * glue, nessie, ...) can be configured with the same property keys documented by Iceberg. + */ +public class IcebergOptions implements Serializable { + + @Serial + private static final long serialVersionUID = 1L; + + private final Map catalogProperties; + private final String tableIdentifier; + private final RecordMapper recordMapper; + private final FileFormat fileFormat; + private final Long targetFileSizeBytes; + private final Schema autoCreateSchema; + private final PartitionSpec autoCreateSpec; + + private IcebergOptions(Builder builder) { + this.catalogProperties = builder.catalogProperties; + this.tableIdentifier = builder.tableIdentifier; + this.recordMapper = builder.recordMapper; + this.fileFormat = builder.fileFormat; + this.targetFileSizeBytes = builder.targetFileSizeBytes; + this.autoCreateSchema = builder.autoCreateSchema; + this.autoCreateSpec = builder.autoCreateSpec; + } + + public Map getCatalogProperties() { + return catalogProperties; + } + + public String getTableIdentifier() { + return tableIdentifier; + } + + public RecordMapper getRecordMapper() { + return recordMapper; + } + + public FileFormat getFileFormat() { + return fileFormat; + } + + public Long getTargetFileSizeBytes() { + return targetFileSizeBytes; + } + + public Schema getAutoCreateSchema() { + return autoCreateSchema; + } + + public PartitionSpec getAutoCreateSpec() { + return autoCreateSpec; + } + + public static class Builder { + private Map catalogProperties; + private String tableIdentifier; + private RecordMapper recordMapper = new FieldNameRecordMapper(); + private FileFormat fileFormat = FileFormat.PARQUET; + private Long targetFileSizeBytes; + private Schema autoCreateSchema; + private PartitionSpec autoCreateSpec; + + public Builder withCatalogProperties(Map properties) { + this.catalogProperties = properties == null ? null : new HashMap<>(properties); + return this; + } + + public Builder withTable(String identifier) { + this.tableIdentifier = identifier; + return this; + } + + public Builder withRecordMapper(RecordMapper mapper) { + this.recordMapper = mapper; + return this; + } + + public Builder withFileFormat(FileFormat format) { + this.fileFormat = format; + return this; + } + + public Builder withTargetFileSizeBytes(long bytes) { + this.targetFileSizeBytes = bytes; + return this; + } + + /** + * Create the table on first use when it does not exist, with the given schema and + * partition spec. A null spec means unpartitioned. + */ + public Builder withAutoCreate(Schema schema, PartitionSpec spec) { + this.autoCreateSchema = schema; + this.autoCreateSpec = spec; + return this; + } + + public IcebergOptions build() { + if (catalogProperties == null || catalogProperties.isEmpty()) { + throw new IllegalStateException("Catalog properties must be specified."); + } + if (tableIdentifier == null || tableIdentifier.isBlank()) { + throw new IllegalStateException("Table identifier must be specified."); + } + if (recordMapper == null) { + throw new IllegalStateException("RecordMapper must not be null."); + } + if (fileFormat == null) { + throw new IllegalStateException("FileFormat must not be null."); + } + if (targetFileSizeBytes != null && targetFileSizeBytes <= 0) { + throw new IllegalStateException("Target file size must be positive."); + } + return new IcebergOptions(this); + } + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergState.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergState.java new file mode 100644 index 00000000000..89119fcd8d4 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergState.java @@ -0,0 +1,289 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import java.io.Closeable; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.Transaction; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericAppenderFactory; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.exceptions.CommitStateUnknownException; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.TaskWriter; +import org.apache.iceberg.io.UnpartitionedWriter; +import org.apache.iceberg.util.PropertyUtil; +import org.apache.storm.Config; +import org.apache.storm.task.IMetricsContext; +import org.apache.storm.topology.FailedException; +import org.apache.storm.trident.operation.TridentCollector; +import org.apache.storm.trident.state.State; +import org.apache.storm.trident.tuple.TridentTuple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Trident {@link State} that appends batches to an Apache Iceberg table with exactly-once + * semantics. + * + *

Each Trident batch is committed with an Iceberg {@link org.apache.iceberg.Transaction} that + * atomically appends the batch's data files and records the transaction id in the table property + * {@code storm.trident.<topologyName>.<partitionIndex>.last-committed-txid}. On replay, + * {@link #beginCommit(Long)} detects already-committed transaction ids and skips the batch. + */ +public class IcebergState implements State { + + static final String TXID_PROPERTY_FORMAT = "storm.trident.%s.%d.last-committed-txid"; + private static final Logger LOG = LoggerFactory.getLogger(IcebergState.class); + private static final String CATALOG_NAME = "storm-iceberg"; + + private final IcebergOptions options; + private final int partitionIndex; + + private IcebergStateMetrics metrics; + private Catalog catalog; + private Table table; + private String txidPropertyKey; + private long lastCommittedTxId; + private boolean skipBatch; + private TaskWriter writer; + private Thread shutdownHook; + private volatile boolean closed; + + IcebergState(IcebergOptions options, int partitionIndex) { + this.options = options; + this.partitionIndex = partitionIndex; + } + + void prepare(Map topoConf, IMetricsContext metricsContext) { + String topologyName = (String) topoConf.get(Config.TOPOLOGY_NAME); + this.metrics = new IcebergStateMetrics(metricsContext); + this.txidPropertyKey = String.format(TXID_PROPERTY_FORMAT, topologyName, partitionIndex); + this.catalog = CatalogUtil.buildIcebergCatalog(CATALOG_NAME, options.getCatalogProperties(), new Configuration()); + TableIdentifier identifier = TableIdentifier.parse(options.getTableIdentifier()); + this.table = loadOrCreateTable(identifier); + String lastTxid = table.properties().get(txidPropertyKey); + this.lastCommittedTxId = lastTxid == null ? 0L : Long.parseLong(lastTxid); + // Trident's State has no lifecycle callback, so a shutdown hook is the only place left to + // release the catalog's client (REST/HTTP, Hive metastore, object store) on worker exit. + this.shutdownHook = new Thread(this::close, "iceberg-state-close-" + partitionIndex); + Runtime.getRuntime().addShutdownHook(shutdownHook); + LOG.info("Prepared IcebergState for table {}, partition {}, lastCommittedTxId {}", + identifier, partitionIndex, lastCommittedTxId); + } + + private Table loadOrCreateTable(TableIdentifier identifier) { + if (options.getAutoCreateSchema() != null && !catalog.tableExists(identifier)) { + PartitionSpec spec = options.getAutoCreateSpec() == null + ? PartitionSpec.unpartitioned() + : options.getAutoCreateSpec(); + try { + LOG.info("Auto-creating Iceberg table {}", identifier); + return catalog.createTable(identifier, options.getAutoCreateSchema(), spec); + } catch (AlreadyExistsException e) { + LOG.info("Table {} was concurrently created by another state partition", identifier); + } + } + return catalog.loadTable(identifier); + } + + long getLastCommittedTxId() { + return lastCommittedTxId; + } + + @Override + public void beginCommit(Long txId) { + // A failed batch attempt can leave a writer with partially written records; drop it so + // the replayed attempt starts from a clean writer. + abortWriter(); + refreshTable(); + skipBatch = txId <= lastCommittedTxId; + if (skipBatch) { + LOG.info("txId {} is already committed (lastCommittedTxId {}), skipping replayed batch", + txId, lastCommittedTxId); + metrics.batchSkipped(); + } + } + + /** + * Pick up schema and partition spec evolution before the batch's writer is created. Without + * this the state would keep writing against the metadata read at {@link #prepare} until the + * worker restarts. Costs one catalog round-trip per batch, alongside the commit's own. + */ + private void refreshTable() { + try { + table.refresh(); + } catch (RuntimeException e) { + throw new FailedException("Failed refreshing Iceberg table metadata, failing batch", e); + } + } + + public void updateState(List tuples, TridentCollector collector) { + if (skipBatch) { + return; + } + try { + if (writer == null) { + writer = createWriter(); + } + for (TridentTuple tuple : tuples) { + writer.write(options.getRecordMapper().map(tuple, table.schema())); + } + metrics.recordsWritten(tuples.size()); + } catch (IOException e) { + abortWriter(); + throw new FailedException("Failed writing records to Iceberg, failing batch", e); + } catch (RuntimeException e) { + abortWriter(); + throw e; + } + } + + @Override + public void commit(Long txId) { + if (skipBatch) { + skipBatch = false; + return; + } + DataFile[] dataFiles = completeWriter(); + long startNanos = System.nanoTime(); + try { + Transaction txn = table.newTransaction(); + txn.updateProperties().set(txidPropertyKey, String.valueOf(txId)).commit(); + if (dataFiles.length > 0) { + AppendFiles append = txn.newAppend(); + for (DataFile dataFile : dataFiles) { + append.appendFile(dataFile); + } + append.commit(); + } + txn.commitTransaction(); + lastCommittedTxId = txId; + metrics.committed(dataFiles, System.nanoTime() - startNanos); + } catch (CommitStateUnknownException e) { + // The commit may or may not have reached the catalog. Do not delete the data files: + // beginCommit on the replayed batch reads the txid property (atomic with the append) + // and resolves the ambiguity. + metrics.commitFailed(); + throw new FailedException("Iceberg commit state unknown, failing batch for replay", e); + } catch (RuntimeException e) { + metrics.commitFailed(); + throw new FailedException("Iceberg commit failed, failing batch", e); + } + } + + private TaskWriter createWriter() { + Schema schema = table.schema(); + PartitionSpec spec = table.spec(); + FileFormat format = options.getFileFormat(); + long targetFileSize = options.getTargetFileSizeBytes() != null + ? options.getTargetFileSizeBytes() + : PropertyUtil.propertyAsLong(table.properties(), + TableProperties.WRITE_TARGET_FILE_SIZE_BYTES, + TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT); + GenericAppenderFactory appenderFactory = new GenericAppenderFactory(schema, spec); + // A fresh OutputFileFactory per batch: its random operation id keeps file names from + // replayed batches unique. + OutputFileFactory fileFactory = OutputFileFactory + .builderFor(table, partitionIndex, System.currentTimeMillis()) + .format(format) + .build(); + if (spec.isUnpartitioned()) { + return new UnpartitionedWriter<>(spec, format, appenderFactory, fileFactory, table.io(), targetFileSize); + } + return new PartitionedRecordWriter(spec, format, appenderFactory, fileFactory, table.io(), targetFileSize, schema); + } + + private DataFile[] completeWriter() { + try { + return writer == null ? new DataFile[0] : writer.complete().dataFiles(); + } catch (IOException e) { + // Files already closed by the writer may remain as never-committed orphans; they are + // invisible to readers and removed by standard Iceberg orphan-file maintenance. + throw new FailedException("Failed closing Iceberg writer, failing batch", e); + } finally { + writer = null; + } + } + + private void abortWriter() { + if (writer == null) { + return; + } + try { + writer.abort(); + } catch (IOException e) { + LOG.warn("Failed aborting Iceberg writer; uncommitted files may remain as orphans", e); + } finally { + writer = null; + } + } + + synchronized void close() { + if (closed) { + return; + } + closed = true; + removeShutdownHook(); + abortWriter(); + if (catalog instanceof Closeable) { + try { + ((Closeable) catalog).close(); + } catch (IOException e) { + LOG.warn("Failed closing Iceberg catalog", e); + } + } + } + + boolean isClosed() { + return closed; + } + + Thread getShutdownHook() { + return shutdownHook; + } + + private void removeShutdownHook() { + Thread hook = shutdownHook; + shutdownHook = null; + // Removing a hook from within the hook itself is both pointless and illegal. + if (hook == null || hook == Thread.currentThread()) { + return; + } + try { + Runtime.getRuntime().removeShutdownHook(hook); + } catch (IllegalStateException e) { + // The JVM is already shutting down and will run the hook itself; close() is idempotent. + LOG.debug("JVM already shutting down, leaving the close hook in place", e); + } + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateFactory.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateFactory.java new file mode 100644 index 00000000000..1c3b8a7d47d --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateFactory.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import java.util.Map; +import org.apache.storm.task.IMetricsContext; +import org.apache.storm.trident.state.State; +import org.apache.storm.trident.state.StateFactory; + +/** + * {@link StateFactory} for {@link IcebergState}. Use with + * {@code stream.partitionPersist(new IcebergStateFactory(options), fields, new IcebergStateUpdater())}. + */ +public class IcebergStateFactory implements StateFactory { + + private static final long serialVersionUID = 1L; + + private final IcebergOptions options; + + public IcebergStateFactory(IcebergOptions options) { + this.options = options; + } + + @Override + public State makeState(Map conf, IMetricsContext metrics, int partitionIndex, int numPartitions) { + IcebergState state = new IcebergState(options, partitionIndex); + state.prepare(conf, metrics); + return state; + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateMetrics.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateMetrics.java new file mode 100644 index 00000000000..99515216082 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateMetrics.java @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Timer; +import java.util.concurrent.TimeUnit; +import org.apache.iceberg.DataFile; +import org.apache.storm.task.IMetricsContext; + +/** + * Storm metrics-v2 instrumentation for {@link IcebergState}. + * + *

When no {@link IMetricsContext} is available the metrics are still recorded, on unregistered + * instances, so callers never need a null check. + */ +class IcebergStateMetrics { + + static final String RECORDS_WRITTEN = "iceberg-records-written"; + static final String DATA_FILES_COMMITTED = "iceberg-data-files-committed"; + static final String BYTES_COMMITTED = "iceberg-bytes-committed"; + static final String COMMIT_LATENCY = "iceberg-commit-latency"; + static final String COMMIT_FAILURES = "iceberg-commit-failures"; + static final String BATCHES_SKIPPED = "iceberg-batches-skipped"; + + private final Counter recordsWritten; + private final Counter dataFilesCommitted; + private final Counter bytesCommitted; + private final Timer commitLatency; + private final Counter commitFailures; + private final Counter batchesSkipped; + + IcebergStateMetrics(IMetricsContext metrics) { + this.recordsWritten = counter(metrics, RECORDS_WRITTEN); + this.dataFilesCommitted = counter(metrics, DATA_FILES_COMMITTED); + this.bytesCommitted = counter(metrics, BYTES_COMMITTED); + this.commitFailures = counter(metrics, COMMIT_FAILURES); + this.batchesSkipped = counter(metrics, BATCHES_SKIPPED); + this.commitLatency = metrics == null ? new Timer() : metrics.registerTimer(COMMIT_LATENCY); + } + + private static Counter counter(IMetricsContext metrics, String name) { + return metrics == null ? new Counter() : metrics.registerCounter(name); + } + + void recordsWritten(long count) { + recordsWritten.inc(count); + } + + /** Counts the files and bytes made visible by a successful commit. */ + void committed(DataFile[] dataFiles, long durationNanos) { + dataFilesCommitted.inc(dataFiles.length); + for (DataFile dataFile : dataFiles) { + bytesCommitted.inc(dataFile.fileSizeInBytes()); + } + commitLatency.update(durationNanos, TimeUnit.NANOSECONDS); + } + + void commitFailed() { + commitFailures.inc(); + } + + void batchSkipped() { + batchesSkipped.inc(); + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateUpdater.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateUpdater.java new file mode 100644 index 00000000000..1e0b5ec77a9 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateUpdater.java @@ -0,0 +1,37 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import java.util.List; +import org.apache.storm.trident.operation.TridentCollector; +import org.apache.storm.trident.state.BaseStateUpdater; +import org.apache.storm.trident.tuple.TridentTuple; + +/** + * {@link BaseStateUpdater} that forwards each Trident batch to {@link IcebergState}. + */ +public class IcebergStateUpdater extends BaseStateUpdater { + + private static final long serialVersionUID = 1L; + + @Override + public void updateState(IcebergState state, List tuples, TridentCollector collector) { + state.updateState(tuples, collector); + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/PartitionedRecordWriter.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/PartitionedRecordWriter.java new file mode 100644 index 00000000000..fee8321aa55 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/PartitionedRecordWriter.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionKey; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.InternalRecordWrapper; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.io.FileAppenderFactory; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.PartitionedFanoutWriter; + +/** + * Fanout writer that routes generic {@link Record}s to one open file per table partition. + */ +class PartitionedRecordWriter extends PartitionedFanoutWriter { + + private final PartitionKey partitionKey; + private final InternalRecordWrapper wrapper; + + PartitionedRecordWriter(PartitionSpec spec, FileFormat format, FileAppenderFactory appenderFactory, + OutputFileFactory fileFactory, FileIO io, long targetFileSize, Schema schema) { + super(spec, format, appenderFactory, fileFactory, io, targetFileSize); + this.partitionKey = new PartitionKey(spec, schema); + this.wrapper = new InternalRecordWrapper(schema.asStruct()); + } + + @Override + protected PartitionKey partition(Record record) { + // PartitionedFanoutWriter copies the key when it caches a new partition's writer, + // so reusing one mutable key here is safe. + partitionKey.partition(wrapper.wrap(record)); + return partitionKey; + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/RecordMapper.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/RecordMapper.java new file mode 100644 index 00000000000..0a11361a7e5 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/RecordMapper.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import java.io.Serializable; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.Record; +import org.apache.storm.trident.tuple.TridentTuple; + +/** + * Converts a {@link TridentTuple} into an Iceberg {@link Record} matching the target table schema. + * Implementations must be serializable: they are shipped with the topology. + */ +public interface RecordMapper extends Serializable { + + /** + * Convert a tuple to a record for the given table schema. + * + * @param tuple the input tuple + * @param schema the current schema of the target Iceberg table + * @return the record to write; never null + */ + Record map(TridentTuple tuple, Schema schema); +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/FieldNameRecordMapperTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/FieldNameRecordMapperTest.java new file mode 100644 index 00000000000..8865f00887c --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/FieldNameRecordMapperTest.java @@ -0,0 +1,143 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.types.Types; +import org.apache.storm.trident.tuple.TridentTuple; +import org.junit.jupiter.api.Test; + +class FieldNameRecordMapperTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get()), + Types.NestedField.optional(3, "score", Types.DoubleType.get()), + Types.NestedField.optional(4, "ts", Types.TimestampType.withZone())); + + private final FieldNameRecordMapper mapper = new FieldNameRecordMapper(); + + private TridentTuple mockTuple() { + TridentTuple tuple = mock(TridentTuple.class); + when(tuple.contains(anyString())).thenReturn(false); + return tuple; + } + + private void field(TridentTuple tuple, String name, Object value) { + when(tuple.contains(name)).thenReturn(true); + when(tuple.getValueByField(name)).thenReturn(value); + } + + @Test + void mapsFieldsByName() { + TridentTuple tuple = mockTuple(); + field(tuple, "id", 42L); + field(tuple, "name", "storm"); + field(tuple, "score", 0.5d); + + Record record = mapper.map(tuple, SCHEMA); + + assertEquals(42L, record.getField("id")); + assertEquals("storm", record.getField("name")); + assertEquals(0.5d, record.getField("score")); + assertNull(record.getField("ts")); + } + + @Test + void widensNumericTypes() { + Schema schema = new Schema( + Types.NestedField.required(1, "i", Types.IntegerType.get()), + Types.NestedField.required(2, "l", Types.LongType.get()), + Types.NestedField.required(3, "f", Types.FloatType.get()), + Types.NestedField.required(4, "d", Types.DoubleType.get())); + TridentTuple tuple = mockTuple(); + field(tuple, "i", (short) 7); + field(tuple, "l", 7); + field(tuple, "f", 7); + field(tuple, "d", 7.0f); + + Record record = mapper.map(tuple, schema); + + assertEquals(7, record.getField("i")); + assertEquals(7L, record.getField("l")); + assertEquals(7.0f, record.getField("f")); + assertEquals((double) 7.0f, record.getField("d")); + } + + @Test + void convertsInstantAndEpochMillisToTimestamptz() { + Instant instant = Instant.parse("2026-07-12T10:15:30Z"); + TridentTuple tuple = mockTuple(); + field(tuple, "id", 1L); + field(tuple, "name", "x"); + field(tuple, "ts", instant); + + Record record = mapper.map(tuple, SCHEMA); + assertEquals(OffsetDateTime.ofInstant(instant, ZoneOffset.UTC), record.getField("ts")); + + TridentTuple tuple2 = mockTuple(); + field(tuple2, "id", 1L); + field(tuple2, "name", "x"); + field(tuple2, "ts", instant.toEpochMilli()); + + Record record2 = mapper.map(tuple2, SCHEMA); + assertEquals(OffsetDateTime.ofInstant(instant, ZoneOffset.UTC), record2.getField("ts")); + } + + @Test + void missingRequiredFieldThrows() { + TridentTuple tuple = mockTuple(); + field(tuple, "id", 1L); + // "name" (required) absent from the tuple + + IllegalArgumentException e = assertThrows(IllegalArgumentException.class, () -> mapper.map(tuple, SCHEMA)); + assertTrue(e.getMessage().contains("name")); + } + + @Test + void nullForRequiredFieldThrows() { + TridentTuple tuple = mockTuple(); + field(tuple, "id", 1L); + field(tuple, "name", null); + + assertThrows(IllegalArgumentException.class, () -> mapper.map(tuple, SCHEMA)); + } + + @Test + void missingOptionalFieldIsNull() { + TridentTuple tuple = mockTuple(); + field(tuple, "id", 1L); + field(tuple, "name", "x"); + + Record record = mapper.map(tuple, SCHEMA); + assertNull(record.getField("score")); + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergOptionsTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergOptionsTest.java new file mode 100644 index 00000000000..ee12d18b3e4 --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergOptionsTest.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectOutputStream; +import java.util.Map; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; + +class IcebergOptionsTest { + + private static final Map CATALOG_PROPS = Map.of("type", "hadoop", "warehouse", "file:///tmp/wh"); + + private IcebergOptions.Builder validBuilder() { + return new IcebergOptions.Builder() + .withCatalogProperties(CATALOG_PROPS) + .withTable("db.events"); + } + + @Test + void buildsWithDefaults() { + IcebergOptions options = validBuilder().build(); + + assertEquals(CATALOG_PROPS, options.getCatalogProperties()); + assertEquals("db.events", options.getTableIdentifier()); + assertInstanceOf(FieldNameRecordMapper.class, options.getRecordMapper()); + assertEquals(FileFormat.PARQUET, options.getFileFormat()); + assertNull(options.getTargetFileSizeBytes()); + assertNull(options.getAutoCreateSchema()); + assertNull(options.getAutoCreateSpec()); + } + + @Test + void rejectsMissingCatalogProperties() { + IcebergOptions.Builder builder = new IcebergOptions.Builder().withTable("db.events"); + assertThrows(IllegalStateException.class, builder::build); + } + + @Test + void rejectsMissingTable() { + IcebergOptions.Builder builder = new IcebergOptions.Builder().withCatalogProperties(CATALOG_PROPS); + assertThrows(IllegalStateException.class, builder::build); + } + + @Test + void rejectsNonPositiveTargetFileSize() { + assertThrows(IllegalStateException.class, () -> validBuilder().withTargetFileSizeBytes(0).build()); + } + + @Test + void isJavaSerializableWithAutoCreate() throws IOException { + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.LongType.get())); + IcebergOptions options = validBuilder() + .withAutoCreate(schema, PartitionSpec.unpartitioned()) + .withTargetFileSizeBytes(1024L) + .build(); + + try (ObjectOutputStream out = new ObjectOutputStream(new ByteArrayOutputStream())) { + out.writeObject(options); // must not throw NotSerializableException + } + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateFactoryTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateFactoryTest.java new file mode 100644 index 00000000000..c98f84b1313 --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateFactoryTest.java @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.Schema; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Types; +import org.apache.storm.Config; +import org.apache.storm.trident.state.State; +import org.apache.storm.trident.tuple.TridentTuple; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class IcebergStateFactoryTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get())); + private static final TableIdentifier TABLE_ID = TableIdentifier.of("db", "events"); + + @TempDir + Path tempDir; + + @Test + void makeStatePreparesAndUpdaterWritesBatch() throws IOException { + String warehouse = tempDir.toUri().toString(); + try (HadoopCatalog catalog = new HadoopCatalog(new Configuration(), warehouse)) { + catalog.createTable(TABLE_ID, SCHEMA); + + Map catalogProps = new HashMap<>(); + catalogProps.put(CatalogUtil.ICEBERG_CATALOG_TYPE, CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP); + catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse); + IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events") + .build(); + + Map conf = new HashMap<>(); + conf.put(Config.TOPOLOGY_NAME, "factory-test"); + + State state = new IcebergStateFactory(options).makeState(conf, null, 0, 1); + IcebergState icebergState = assertInstanceOf(IcebergState.class, state); + try { + TridentTuple tuple = mock(TridentTuple.class); + when(tuple.contains(anyString())).thenReturn(false); + when(tuple.contains("id")).thenReturn(true); + when(tuple.contains("name")).thenReturn(true); + when(tuple.getValueByField("id")).thenReturn(99L); + when(tuple.getValueByField("name")).thenReturn("via-factory"); + + icebergState.beginCommit(1L); + new IcebergStateUpdater().updateState(icebergState, List.of(tuple), null); + icebergState.commit(1L); + + List rows = new ArrayList<>(); + try (CloseableIterable iterable = IcebergGenerics.read(catalog.loadTable(TABLE_ID)).build()) { + iterable.forEach(rows::add); + } + assertEquals(1, rows.size()); + assertEquals(99L, rows.get(0).getField("id")); + } finally { + icebergState.close(); + } + } + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateTest.java new file mode 100644 index 00000000000..5c087a00da1 --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateTest.java @@ -0,0 +1,504 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.NoSuchTableException; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Types; +import org.apache.storm.Config; +import org.apache.storm.topology.FailedException; +import org.apache.storm.trident.tuple.TridentTuple; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class IcebergStateTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get())); + private static final TableIdentifier TABLE_ID = TableIdentifier.of("db", "events"); + private static final String TOPOLOGY_NAME = "test-topology"; + + @TempDir + Path tempDir; + + private String warehouse; + private HadoopCatalog verifyCatalog; + + @BeforeEach + void setUp() { + warehouse = tempDir.toUri().toString(); + verifyCatalog = new HadoopCatalog(new Configuration(), warehouse); + } + + @AfterEach + void tearDown() throws IOException { + verifyCatalog.close(); + } + + private IcebergOptions.Builder baseOptions() { + Map catalogProps = new HashMap<>(); + catalogProps.put(CatalogUtil.ICEBERG_CATALOG_TYPE, CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP); + catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse); + return new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events"); + } + + private Map topoConf() { + Map conf = new HashMap<>(); + conf.put(Config.TOPOLOGY_NAME, TOPOLOGY_NAME); + return conf; + } + + private IcebergState newState(IcebergOptions options) { + return newState(options, null); + } + + private IcebergState newState(IcebergOptions options, RecordingMetricsContext metrics) { + IcebergState state = new IcebergState(options, 0); + state.prepare(topoConf(), metrics); + return state; + } + + private TridentTuple tuple(long id, String name) { + TridentTuple tuple = mock(TridentTuple.class); + when(tuple.contains(anyString())).thenReturn(false); + when(tuple.contains("id")).thenReturn(true); + when(tuple.contains("name")).thenReturn(true); + when(tuple.getValueByField("id")).thenReturn(id); + when(tuple.getValueByField("name")).thenReturn(name); + return tuple; + } + + private TridentTuple tupleWithExtra(long id, String name, String extra) { + TridentTuple tuple = tuple(id, name); + when(tuple.contains("extra")).thenReturn(true); + when(tuple.getValueByField("extra")).thenReturn(extra); + return tuple; + } + + private List readRows() throws IOException { + Table table = verifyCatalog.loadTable(TABLE_ID); + List rows = new ArrayList<>(); + try (CloseableIterable iterable = IcebergGenerics.read(table).build()) { + iterable.forEach(rows::add); + } + return rows; + } + + @Test + void prepareFailsWhenTableMissingAndNoAutoCreate() { + IcebergState state = new IcebergState(baseOptions().build(), 0); + assertThrows(NoSuchTableException.class, () -> state.prepare(topoConf(), null)); + } + + @Test + void prepareAutoCreatesTable() { + IcebergState state = newState(baseOptions() + .withAutoCreate(SCHEMA, PartitionSpec.unpartitioned()) + .build()); + try { + assertTrue(verifyCatalog.tableExists(TABLE_ID)); + assertEquals(0L, state.getLastCommittedTxId()); + } finally { + state.close(); + } + } + + @Test + void prepareLoadsExistingTableAndReadsTxid() { + Table table = verifyCatalog.createTable(TABLE_ID, SCHEMA); + table.updateProperties() + .set("storm.trident." + TOPOLOGY_NAME + ".0.last-committed-txid", "7") + .commit(); + + IcebergState state = newState(baseOptions().build()); + try { + assertEquals(7L, state.getLastCommittedTxId()); + } finally { + state.close(); + } + } + + @Test + void writesBatchAndCommitsAtomically() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a"), tuple(2L, "b")), null); + state.commit(1L); + + List rows = readRows(); + assertEquals(2, rows.size()); + Table table = verifyCatalog.loadTable(TABLE_ID); + assertEquals("1", table.properties().get("storm.trident." + TOPOLOGY_NAME + ".0.last-committed-txid")); + } finally { + state.close(); + } + } + + @Test + void emptyBatchStillAdvancesTxid() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + try { + state.beginCommit(1L); + state.commit(1L); + + assertEquals(0, readRows().size()); + Table table = verifyCatalog.loadTable(TABLE_ID); + assertEquals("1", table.properties().get("storm.trident." + TOPOLOGY_NAME + ".0.last-committed-txid")); + assertEquals(1L, state.getLastCommittedTxId()); + } finally { + state.close(); + } + } + + @Test + void consecutiveBatchesAppend() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + + state.beginCommit(2L); + state.updateState(List.of(tuple(2L, "b")), null); + state.commit(2L); + + assertEquals(2, readRows().size()); + } finally { + state.close(); + } + } + + @Test + void replayedBatchIsSkipped() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + + // Trident replays txid 1 (e.g. commit acknowledgement was lost) + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + + assertEquals(1, readRows().size()); + } finally { + state.close(); + } + } + + @Test + void replayAfterRestartIsSkipped() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergOptions options = baseOptions().build(); + + IcebergState first = newState(options); + try { + first.beginCommit(1L); + first.updateState(List.of(tuple(1L, "a")), null); + first.commit(1L); + } finally { + first.close(); + } + + // Worker restarts: a fresh state must recover the txid from the table and skip the replay. + IcebergState second = newState(options); + try { + assertEquals(1L, second.getLastCommittedTxId()); + second.beginCommit(1L); + second.updateState(List.of(tuple(1L, "a")), null); + second.commit(1L); + + assertEquals(1, readRows().size()); + + // The next batch proceeds normally. + second.beginCommit(2L); + second.updateState(List.of(tuple(2L, "b")), null); + second.commit(2L); + assertEquals(2, readRows().size()); + } finally { + second.close(); + } + } + + @Test + void crashBeforeCommitLeavesNoVisibleData() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergOptions options = baseOptions().build(); + + IcebergState first = newState(options); + first.beginCommit(1L); + first.updateState(List.of(tuple(1L, "a")), null); + // Simulated crash: no commit(1L). Written files are never committed. + first.close(); + + assertEquals(0, readRows().size()); + + IcebergState second = newState(options); + try { + second.beginCommit(1L); + second.updateState(List.of(tuple(1L, "a")), null); + second.commit(1L); + + assertEquals(1, readRows().size()); + } finally { + second.close(); + } + } + + @Test + void writesToPartitionedTable() throws IOException { + PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("name").build(); + verifyCatalog.createTable(TABLE_ID, SCHEMA, spec); + + IcebergState state = newState(baseOptions().build()); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "alpha"), tuple(2L, "beta"), tuple(3L, "alpha")), null); + state.commit(1L); + + assertEquals(3, readRows().size()); + Table table = verifyCatalog.loadTable(TABLE_ID); + // identity("name") with two distinct values -> one data file per partition + assertEquals("2", table.currentSnapshot().summary().get("added-data-files")); + } finally { + state.close(); + } + } + + @Test + void failedAttemptThenReplayDoesNotDuplicateWithinBatch() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + try { + // First attempt of txid 1 writes some tuples, then the batch fails before commit + // (e.g. another component of the topology failed the batch). + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + + // Replayed attempt of the same txid: beginCommit must discard the leftover writer. + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + + assertEquals(1, readRows().size()); + } finally { + state.close(); + } + } + + /** Mapper that fails on tuples whose "name" field equals "poison". */ + private static class PoisonRecordMapper extends FieldNameRecordMapper { + private static final long serialVersionUID = 1L; + + @Override + public Record map(TridentTuple tuple, Schema schema) { + if ("poison".equals(tuple.getValueByField("name"))) { + throw new IllegalStateException("poisoned tuple"); + } + return super.map(tuple, schema); + } + } + + @Test + void mappingFailureAbortsWriterAndPropagates() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions() + .withRecordMapper(new PoisonRecordMapper()) + .build()); + try { + state.beginCommit(1L); + assertThrows(IllegalStateException.class, + () -> state.updateState(List.of(tuple(1L, "ok"), tuple(2L, "poison")), null)); + + // Replay of the same txid with clean data succeeds and contains no leftovers + // from the failed attempt. + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "ok"), tuple(2L, "fixed")), null); + state.commit(1L); + + assertEquals(2, readRows().size()); + } finally { + state.close(); + } + } + + @Test + void closeIsIdempotentAndDeregistersTheShutdownHook() { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + Thread hook = state.getShutdownHook(); + assertNotNull(hook, "prepare must register a shutdown hook to release the catalog"); + + state.close(); + assertTrue(state.isClosed()); + assertNull(state.getShutdownHook()); + // false means the JVM no longer holds the hook, i.e. close() really deregistered it. + assertFalse(Runtime.getRuntime().removeShutdownHook(hook)); + + state.close(); + assertTrue(state.isClosed()); + } + + @Test + void picksUpSchemaEvolutionBetweenBatches() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "before")), null); + state.commit(1L); + + // The table gains a column while the topology is running. + verifyCatalog.loadTable(TABLE_ID).updateSchema() + .addColumn("extra", Types.StringType.get()) + .commit(); + + state.beginCommit(2L); + state.updateState(List.of(tupleWithExtra(2L, "after", "value")), null); + state.commit(2L); + + List rows = readRows(); + assertEquals(2, rows.size()); + // Without a refresh the state would still map against the two-column schema captured + // at prepare() and silently drop the new column. + assertTrue(rows.stream().anyMatch(r -> "value".equals(r.getField("extra")))); + } finally { + state.close(); + } + } + + @Test + void metricsCountRecordsFilesAndCommits() { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + RecordingMetricsContext metrics = new RecordingMetricsContext(); + IcebergState state = newState(baseOptions().build(), metrics); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a"), tuple(2L, "b"), tuple(3L, "c")), null); + state.commit(1L); + + assertEquals(3L, metrics.counter(IcebergStateMetrics.RECORDS_WRITTEN)); + assertEquals(1L, metrics.counter(IcebergStateMetrics.DATA_FILES_COMMITTED)); + assertTrue(metrics.counter(IcebergStateMetrics.BYTES_COMMITTED) > 0L); + assertEquals(1L, metrics.timerCount(IcebergStateMetrics.COMMIT_LATENCY)); + assertEquals(0L, metrics.counter(IcebergStateMetrics.COMMIT_FAILURES)); + assertEquals(0L, metrics.counter(IcebergStateMetrics.BATCHES_SKIPPED)); + } finally { + state.close(); + } + } + + @Test + void metricsCountSkippedReplays() { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + RecordingMetricsContext metrics = new RecordingMetricsContext(); + IcebergState state = newState(baseOptions().build(), metrics); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + + // Replay of an already committed txid. + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + + assertEquals(1L, metrics.counter(IcebergStateMetrics.BATCHES_SKIPPED)); + assertEquals(1L, metrics.counter(IcebergStateMetrics.RECORDS_WRITTEN)); + assertEquals(1L, metrics.timerCount(IcebergStateMetrics.COMMIT_LATENCY)); + } finally { + state.close(); + } + } + + @Test + void metricsCountCommitFailures() { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + RecordingMetricsContext metrics = new RecordingMetricsContext(); + IcebergState state = newState(baseOptions().build(), metrics); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + // The table disappears underneath the state before it can commit. + verifyCatalog.dropTable(TABLE_ID, false); + + assertThrows(FailedException.class, () -> state.commit(1L)); + assertEquals(1L, metrics.counter(IcebergStateMetrics.COMMIT_FAILURES)); + assertEquals(0L, metrics.timerCount(IcebergStateMetrics.COMMIT_LATENCY)); + } finally { + state.close(); + } + } + + @Test + void requiredNullFieldFailsLoudly() { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + try { + state.beginCommit(1L); + TridentTuple bad = mock(TridentTuple.class); + when(bad.contains(anyString())).thenReturn(false); + when(bad.contains("id")).thenReturn(true); + when(bad.getValueByField("id")).thenReturn(1L); + // required "name" missing -> IllegalArgumentException (not FailedException: this is + // a topology programming error that replays cannot fix) + assertThrows(IllegalArgumentException.class, + () -> state.updateState(List.of(bad), null)); + } finally { + state.close(); + } + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/RecordingMetricsContext.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/RecordingMetricsContext.java new file mode 100644 index 00000000000..e31a9b1caad --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/RecordingMetricsContext.java @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import com.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.MetricSet; +import com.codahale.metrics.Timer; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.metric.api.CombinedMetric; +import org.apache.storm.metric.api.ICombiner; +import org.apache.storm.metric.api.IMetric; +import org.apache.storm.metric.api.IReducer; +import org.apache.storm.metric.api.ReducedMetric; +import org.apache.storm.task.IMetricsContext; + +/** + * Minimal {@link IMetricsContext} that keeps the registered metrics in maps so tests can assert on + * them. Like the real registry, registering the same name twice returns the same instance. + */ +class RecordingMetricsContext implements IMetricsContext { + + private final Map counters = new HashMap<>(); + private final Map timers = new HashMap<>(); + + long counter(String name) { + Counter counter = counters.get(name); + return counter == null ? 0L : counter.getCount(); + } + + long timerCount(String name) { + Timer timer = timers.get(name); + return timer == null ? 0L : timer.getCount(); + } + + @Override + public Counter registerCounter(String name) { + return counters.computeIfAbsent(name, n -> new Counter()); + } + + @Override + public Timer registerTimer(String name) { + return timers.computeIfAbsent(name, n -> new Timer()); + } + + @Override + public Histogram registerHistogram(String name) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + public Meter registerMeter(String name) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + public Gauge registerGauge(String name, Gauge gauge) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + public void registerMetricSet(String prefix, MetricSet set) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + @Deprecated + public T registerMetric(String name, T metric, int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + @Deprecated + public ReducedMetric registerMetric(String name, IReducer reducer, int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + @Deprecated + public CombinedMetric registerMetric(String name, ICombiner combiner, int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("not used by IcebergState"); + } +} diff --git a/storm-dist/binary/final-package/src/main/assembly/binary.xml b/storm-dist/binary/final-package/src/main/assembly/binary.xml index ee5cb60e3ed..3f8343d9bc8 100644 --- a/storm-dist/binary/final-package/src/main/assembly/binary.xml +++ b/storm-dist/binary/final-package/src/main/assembly/binary.xml @@ -130,6 +130,13 @@ README.* + + ${project.basedir}/../../../external/storm-iceberg + external/storm-iceberg + + README.* + + ${project.basedir}/../../../external/storm-eventhubs external/storm-eventhubs From c81a74770769a320076122d6b28e34bd791e0051 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 26 Jul 2026 10:53:53 +0200 Subject: [PATCH 2/4] Fix small files problems in apache iceberg ingestion. Estimate bytes ingested per commit. --- docs/storm-iceberg.md | 42 +++++ .../trident/CountingAppenderFactory.java | 109 +++++++++++++ .../storm/iceberg/trident/IcebergOptions.java | 50 ++++++ .../storm/iceberg/trident/IcebergState.java | 74 ++++++++- .../iceberg/trident/IcebergStateMetrics.java | 16 ++ .../iceberg/trident/IcebergStateTest.java | 144 ++++++++++++++++++ 6 files changed, 431 insertions(+), 4 deletions(-) create mode 100644 external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/CountingAppenderFactory.java diff --git a/docs/storm-iceberg.md b/docs/storm-iceberg.md index 362f0f88347..5399a141860 100644 --- a/docs/storm-iceberg.md +++ b/docs/storm-iceberg.md @@ -40,6 +40,8 @@ so every Iceberg catalog works with its standard configuration keys: `type` = `h | `withFileFormat(FileFormat)` | `PARQUET` | Data file format | | `withTargetFileSizeBytes(long)` | table property `write.target-file-size-bytes` | Rolling file size | | `withAutoCreate(Schema, PartitionSpec)` | disabled | Create the table on first use if missing | +| `withCommitIntervalBytes(long)` | disabled | Buffer batches until this many bytes, then commit once | +| `withCommitIntervalMillis(long)` | disabled | Flush the buffered window once it is this old | ### Tuple mapping @@ -83,6 +85,44 @@ The `PartitionSpec` above is only used when the table is auto-created; for an ex spec stored in the catalog wins. Note that a batch spread over many partitions opens many files at once — partition the stream on the partition columns (`partitionBy`) to keep file counts down. +### Commit batching (trades exactly-once for fewer snapshots) + +By default **every Trident batch is one Iceberg commit and one snapshot**. On a high-rate stream +with small batches that produces thousands of snapshots and as many small files, which degrades +metastore and reader planning. + +`withCommitIntervalBytes` and `withCommitIntervalMillis` buffer batches and commit them together: + +```java +IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events") + .withCommitIntervalBytes(128L * 1024 * 1024) + .withCommitIntervalMillis(60_000) + .build(); +``` + +The writer stays open across the buffered batches, so data files also grow towards +`write.target-file-size-bytes` instead of being one-file-per-batch. Whichever threshold is +reached first flushes the window; setting neither keeps the default one-commit-per-batch. + +> **This weakens the delivery guarantee, and cannot be made not to.** Trident treats a batch as +> delivered the moment `commit()` returns. Batches that are buffered have already been +> acknowledged, so if the worker dies before the flush **they are lost** — not duplicated, lost. +> Use it only where losing the last window on a crash is acceptable, and leave both settings +> unset when you need the exactly-once guarantee described below. + +Two further consequences of buffering: + +- A batch that fails after earlier batches were buffered forces the whole window to be dropped, + since an open writer cannot un-write the failed attempt's records. Watch + `iceberg-windows-dropped`. +- The time threshold is evaluated when the *next* batch is committed, so a stream that stops + entirely leaves its last window uncommitted until data resumes. This matches how + `TimedRotationPolicy` behaves for `HdfsState`; Trident delivers no tick tuples to states and + skips empty batches, so there is no in-state timer that could do better without introducing + concurrency on the commit path. + ### Metrics The state registers these metrics-v2 metrics per `partitionPersist` task, so they show up in @@ -96,6 +136,8 @@ whatever reporter the cluster is configured with: | `iceberg-commit-latency` | timer | Duration of the Iceberg commit transaction | | `iceberg-commit-failures` | counter | Commits that failed (including unknown state) | | `iceberg-batches-skipped` | counter | Replayed batches skipped as already committed | +| `iceberg-batches-buffered` | counter | Batches held back waiting for the commit threshold | +| `iceberg-windows-dropped` | counter | Buffered windows discarded; their batches were lost | A steadily rising `iceberg-data-files-committed` with a flat `iceberg-bytes-committed` is the small-files signature: consider fewer, larger batches or a compaction job. diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/CountingAppenderFactory.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/CountingAppenderFactory.java new file mode 100644 index 00000000000..372764dbea7 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/CountingAppenderFactory.java @@ -0,0 +1,109 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.trident; + +import java.util.ArrayList; +import java.util.List; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.EqualityDeleteWriter; +import org.apache.iceberg.deletes.PositionDeleteWriter; +import org.apache.iceberg.encryption.EncryptedOutputFile; +import org.apache.iceberg.io.DataWriter; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.io.FileAppenderFactory; +import org.apache.iceberg.io.OutputFile; + +/** + * Wraps a {@link FileAppenderFactory} and remembers every writer it hands out, so the state can + * ask how many bytes the currently buffered window has produced. + * + *

The figure is an estimate: {@link FileAppender#length()} reflects what the + * underlying format has flushed, and columnar formats such as Parquet keep a sizeable in-memory + * buffer before writing a row group. It therefore under-reports until a file is closed, which for + * a commit threshold only means committing slightly later than the configured size. + * + *

Closed writers are kept in the list on purpose: a rolled-over file still counts towards the + * bytes accumulated since the last commit. {@link #reset()} drops them when the window is flushed. + */ +class CountingAppenderFactory implements FileAppenderFactory { + + private final FileAppenderFactory delegate; + private final List> appenders = new ArrayList<>(); + private final List> dataWriters = new ArrayList<>(); + + CountingAppenderFactory(FileAppenderFactory delegate) { + this.delegate = delegate; + } + + /** Bytes written by every writer created since the last {@link #reset()}. */ + long estimatedBytes() { + long total = 0L; + for (FileAppender appender : appenders) { + total += appender.length(); + } + for (DataWriter dataWriter : dataWriters) { + total += dataWriter.length(); + } + return total; + } + + /** Forget the writers of the window that was just committed or aborted. */ + void reset() { + appenders.clear(); + dataWriters.clear(); + } + + @Override + public FileAppender newAppender(OutputFile outputFile, FileFormat format) { + FileAppender appender = delegate.newAppender(outputFile, format); + appenders.add(appender); + return appender; + } + + @Override + public FileAppender newAppender(EncryptedOutputFile outputFile, FileFormat format) { + FileAppender appender = delegate.newAppender(outputFile, format); + appenders.add(appender); + return appender; + } + + @Override + public DataWriter newDataWriter(EncryptedOutputFile file, FileFormat format, + StructLike partition) { + DataWriter dataWriter = delegate.newDataWriter(file, format, partition); + dataWriters.add(dataWriter); + return dataWriter; + } + + @Override + public EqualityDeleteWriter newEqDeleteWriter(EncryptedOutputFile file, + FileFormat format, StructLike partition) { + // The sink is append-only; delete writers are never requested. + return delegate.newEqDeleteWriter(file, format, partition); + } + + @Override + public PositionDeleteWriter newPosDeleteWriter(EncryptedOutputFile file, + FileFormat format, + StructLike partition) { + return delegate.newPosDeleteWriter(file, format, partition); + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergOptions.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergOptions.java index 1a3f978af89..180693295dc 100644 --- a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergOptions.java +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergOptions.java @@ -45,6 +45,8 @@ public class IcebergOptions implements Serializable { private final Long targetFileSizeBytes; private final Schema autoCreateSchema; private final PartitionSpec autoCreateSpec; + private final Long commitIntervalBytes; + private final Long commitIntervalMillis; private IcebergOptions(Builder builder) { this.catalogProperties = builder.catalogProperties; @@ -54,6 +56,8 @@ private IcebergOptions(Builder builder) { this.targetFileSizeBytes = builder.targetFileSizeBytes; this.autoCreateSchema = builder.autoCreateSchema; this.autoCreateSpec = builder.autoCreateSpec; + this.commitIntervalBytes = builder.commitIntervalBytes; + this.commitIntervalMillis = builder.commitIntervalMillis; } public Map getCatalogProperties() { @@ -76,6 +80,19 @@ public Long getTargetFileSizeBytes() { return targetFileSizeBytes; } + public Long getCommitIntervalBytes() { + return commitIntervalBytes; + } + + public Long getCommitIntervalMillis() { + return commitIntervalMillis; + } + + /** True when batches are buffered across commits instead of committed one by one. */ + public boolean isCommitBatchingEnabled() { + return commitIntervalBytes != null || commitIntervalMillis != null; + } + public Schema getAutoCreateSchema() { return autoCreateSchema; } @@ -92,6 +109,8 @@ public static class Builder { private Long targetFileSizeBytes; private Schema autoCreateSchema; private PartitionSpec autoCreateSpec; + private Long commitIntervalBytes; + private Long commitIntervalMillis; public Builder withCatalogProperties(Map properties) { this.catalogProperties = properties == null ? null : new HashMap<>(properties); @@ -128,6 +147,31 @@ public Builder withAutoCreate(Schema schema, PartitionSpec spec) { return this; } + /** + * Buffer batches until roughly this many bytes have been written, then commit them all in + * one Iceberg transaction. Without it every Trident batch produces its own commit and + * snapshot. + * + *

This weakens the delivery guarantee. Trident considers a batch + * delivered as soon as {@code commit()} returns, so batches buffered by this setting are + * lost if the worker dies before the flush. Leave it unset to keep exactly-once. + */ + public Builder withCommitIntervalBytes(long bytes) { + this.commitIntervalBytes = bytes; + return this; + } + + /** + * Flush the buffered batches once the oldest one is older than this, evaluated when the + * next batch is committed. A stalled stream therefore leaves the last window uncommitted + * until data resumes. Carries the same durability caveat as + * {@link #withCommitIntervalBytes(long)}. + */ + public Builder withCommitIntervalMillis(long millis) { + this.commitIntervalMillis = millis; + return this; + } + public IcebergOptions build() { if (catalogProperties == null || catalogProperties.isEmpty()) { throw new IllegalStateException("Catalog properties must be specified."); @@ -144,6 +188,12 @@ public IcebergOptions build() { if (targetFileSizeBytes != null && targetFileSizeBytes <= 0) { throw new IllegalStateException("Target file size must be positive."); } + if (commitIntervalBytes != null && commitIntervalBytes <= 0) { + throw new IllegalStateException("Commit interval bytes must be positive."); + } + if (commitIntervalMillis != null && commitIntervalMillis <= 0) { + throw new IllegalStateException("Commit interval millis must be positive."); + } return new IcebergOptions(this); } } diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergState.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergState.java index 89119fcd8d4..b08096f1c3b 100644 --- a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergState.java +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergState.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.AppendFiles; import org.apache.iceberg.CatalogUtil; @@ -76,6 +77,9 @@ public class IcebergState implements State { private long lastCommittedTxId; private boolean skipBatch; private TaskWriter writer; + private CountingAppenderFactory countingAppenderFactory; + private long highestBufferedTxId; + private long windowStartNanos; private Thread shutdownHook; private volatile boolean closed; @@ -122,9 +126,20 @@ long getLastCommittedTxId() { @Override public void beginCommit(Long txId) { - // A failed batch attempt can leave a writer with partially written records; drop it so - // the replayed attempt starts from a clean writer. - abortWriter(); + // A failed batch attempt can leave a writer with partially written records. Without + // buffering that is always the case here, so the writer is simply dropped. With buffering + // the writer also carries earlier, successfully delivered batches, so it may only be + // dropped when this txId is itself a replay of something already in the window: an open + // writer cannot un-write the failed attempt's records, and keeping them would duplicate + // rows once the replay writes them again. + if (!options.isCommitBatchingEnabled() || txId <= highestBufferedTxId) { + if (writer != null && txId <= highestBufferedTxId) { + LOG.warn("txId {} replays a batch already buffered (up to {}); dropping the " + + "buffered window, its batches are lost", txId, highestBufferedTxId); + metrics.windowDropped(); + } + abortWriter(); + } refreshTable(); skipBatch = txId <= lastCommittedTxId; if (skipBatch) { @@ -174,6 +189,43 @@ public void commit(Long txId) { skipBatch = false; return; } + highestBufferedTxId = txId; + if (windowStartNanos == 0L) { + windowStartNanos = System.nanoTime(); + } + if (!shouldFlush()) { + metrics.batchBuffered(); + LOG.debug("Buffering txId {} ({} bytes so far in the window)", txId, bufferedBytes()); + return; + } + flush(txId); + } + + /** + * Decide whether the buffered window has to be committed now. With no interval configured + * every batch is flushed, which is the exactly-once behaviour. + */ + private boolean shouldFlush() { + if (!options.isCommitBatchingEnabled()) { + return true; + } + Long intervalBytes = options.getCommitIntervalBytes(); + if (intervalBytes != null && bufferedBytes() >= intervalBytes) { + return true; + } + Long intervalMillis = options.getCommitIntervalMillis(); + if (intervalMillis != null) { + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - windowStartNanos); + return elapsedMillis >= intervalMillis; + } + return false; + } + + private long bufferedBytes() { + return countingAppenderFactory == null ? 0L : countingAppenderFactory.estimatedBytes(); + } + + private void flush(Long txId) { DataFile[] dataFiles = completeWriter(); long startNanos = System.nanoTime(); try { @@ -210,7 +262,9 @@ private TaskWriter createWriter() { : PropertyUtil.propertyAsLong(table.properties(), TableProperties.WRITE_TARGET_FILE_SIZE_BYTES, TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT); - GenericAppenderFactory appenderFactory = new GenericAppenderFactory(schema, spec); + CountingAppenderFactory appenderFactory = + new CountingAppenderFactory(new GenericAppenderFactory(schema, spec)); + this.countingAppenderFactory = appenderFactory; // A fresh OutputFileFactory per batch: its random operation id keeps file names from // replayed batches unique. OutputFileFactory fileFactory = OutputFileFactory @@ -232,11 +286,15 @@ private DataFile[] completeWriter() { throw new FailedException("Failed closing Iceberg writer, failing batch", e); } finally { writer = null; + // The window's writers are done either way: if the commit below fails, its files + // become orphans and the next batch must start counting from zero. + resetWindow(); } } private void abortWriter() { if (writer == null) { + resetWindow(); return; } try { @@ -245,6 +303,14 @@ private void abortWriter() { LOG.warn("Failed aborting Iceberg writer; uncommitted files may remain as orphans", e); } finally { writer = null; + resetWindow(); + } + } + + private void resetWindow() { + windowStartNanos = 0L; + if (countingAppenderFactory != null) { + countingAppenderFactory.reset(); } } diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateMetrics.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateMetrics.java index 99515216082..8871ef4fad9 100644 --- a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateMetrics.java +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/trident/IcebergStateMetrics.java @@ -38,6 +38,8 @@ class IcebergStateMetrics { static final String COMMIT_LATENCY = "iceberg-commit-latency"; static final String COMMIT_FAILURES = "iceberg-commit-failures"; static final String BATCHES_SKIPPED = "iceberg-batches-skipped"; + static final String BATCHES_BUFFERED = "iceberg-batches-buffered"; + static final String WINDOWS_DROPPED = "iceberg-windows-dropped"; private final Counter recordsWritten; private final Counter dataFilesCommitted; @@ -45,6 +47,8 @@ class IcebergStateMetrics { private final Timer commitLatency; private final Counter commitFailures; private final Counter batchesSkipped; + private final Counter batchesBuffered; + private final Counter windowsDropped; IcebergStateMetrics(IMetricsContext metrics) { this.recordsWritten = counter(metrics, RECORDS_WRITTEN); @@ -52,6 +56,8 @@ class IcebergStateMetrics { this.bytesCommitted = counter(metrics, BYTES_COMMITTED); this.commitFailures = counter(metrics, COMMIT_FAILURES); this.batchesSkipped = counter(metrics, BATCHES_SKIPPED); + this.batchesBuffered = counter(metrics, BATCHES_BUFFERED); + this.windowsDropped = counter(metrics, WINDOWS_DROPPED); this.commitLatency = metrics == null ? new Timer() : metrics.registerTimer(COMMIT_LATENCY); } @@ -79,4 +85,14 @@ void commitFailed() { void batchSkipped() { batchesSkipped.inc(); } + + /** A batch was written but held back, waiting for the commit threshold. */ + void batchBuffered() { + batchesBuffered.inc(); + } + + /** A buffered window was discarded; the batches it held were lost. */ + void windowDropped() { + windowsDropped.inc(); + } } diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateTest.java index 5c087a00da1..77ad5dd6472 100644 --- a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateTest.java +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/trident/IcebergStateTest.java @@ -122,6 +122,16 @@ private TridentTuple tupleWithExtra(long id, String name, String extra) { return tuple; } + private int countSnapshots() { + Table table = verifyCatalog.loadTable(TABLE_ID); + table.refresh(); + int count = 0; + for (Object ignored : table.snapshots()) { + count++; + } + return count; + } + private List readRows() throws IOException { Table table = verifyCatalog.loadTable(TABLE_ID); List rows = new ArrayList<>(); @@ -483,6 +493,140 @@ void metricsCountCommitFailures() { } } + @Test + void withoutCommitIntervalEveryBatchIsCommitted() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions().build()); + try { + for (long txId = 1; txId <= 3; txId++) { + state.beginCommit(txId); + state.updateState(List.of(tuple(txId, "row" + txId)), null); + state.commit(txId); + } + assertEquals(3, readRows().size()); + assertEquals(3, countSnapshots()); + } finally { + state.close(); + } + } + + @Test + void commitIntervalBytesBuffersBatchesIntoASingleSnapshot() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + RecordingMetricsContext metrics = new RecordingMetricsContext(); + // Far larger than the few hundred bytes these batches produce, so only the last batch, + // driven by the time interval below, triggers the flush. + IcebergState state = newState(baseOptions() + .withCommitIntervalBytes(10L * 1024 * 1024) + .build(), metrics); + try { + for (long txId = 1; txId <= 3; txId++) { + state.beginCommit(txId); + state.updateState(List.of(tuple(txId, "row" + txId)), null); + state.commit(txId); + } + // Nothing is visible yet: three batches are buffered, no snapshot exists. + assertEquals(0, countSnapshots()); + assertEquals(3L, metrics.counter(IcebergStateMetrics.BATCHES_BUFFERED)); + assertEquals(0L, metrics.timerCount(IcebergStateMetrics.COMMIT_LATENCY)); + } finally { + state.close(); + } + } + + @Test + void bufferedWindowIsFlushedOnceTheByteThresholdIsCrossed() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + RecordingMetricsContext metrics = new RecordingMetricsContext(); + // One byte: the first batch already exceeds it, but only once its bytes are counted. + IcebergState state = newState(baseOptions().withCommitIntervalBytes(1L).build(), metrics); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + + assertEquals(1, readRows().size()); + assertEquals(1, countSnapshots()); + assertEquals(1L, metrics.timerCount(IcebergStateMetrics.COMMIT_LATENCY)); + assertEquals(0L, metrics.counter(IcebergStateMetrics.BATCHES_BUFFERED)); + } finally { + state.close(); + } + } + + @Test + void commitIntervalMillisFlushesTheWindowOnTheNextBatch() throws IOException, InterruptedException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + IcebergState state = newState(baseOptions() + .withCommitIntervalBytes(10L * 1024 * 1024) + .withCommitIntervalMillis(50L) + .build()); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + assertEquals(0, countSnapshots(), "the first batch only opens the window"); + + // Once the deadline has passed, the next batch commits the whole window. + Thread.sleep(80L); + state.beginCommit(2L); + state.updateState(List.of(tuple(2L, "b")), null); + state.commit(2L); + + assertEquals(2, readRows().size()); + assertEquals(1, countSnapshots(), "both batches must land in one snapshot"); + } finally { + state.close(); + } + } + + @Test + void replayOfABufferedBatchDropsTheWindow() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA); + RecordingMetricsContext metrics = new RecordingMetricsContext(); + IcebergState state = newState(baseOptions() + .withCommitIntervalBytes(10L * 1024 * 1024) + .withCommitIntervalMillis(300L) + .build(), metrics); + try { + state.beginCommit(1L); + state.updateState(List.of(tuple(1L, "a")), null); + state.commit(1L); + state.beginCommit(2L); + state.updateState(List.of(tuple(2L, "b")), null); + state.commit(2L); + assertEquals(0, countSnapshots(), "both batches are still buffered"); + + // txId 2 is replayed: it is buffered but not committed, so the window has to be + // discarded rather than replayed on top of itself. + state.beginCommit(2L); + assertEquals(1L, metrics.counter(IcebergStateMetrics.WINDOWS_DROPPED)); + + state.updateState(List.of(tuple(2L, "b")), null); + state.commit(2L); + state.beginCommit(3L); + state.updateState(List.of(tuple(3L, "c")), null); + state.commit(3L); + + // Past the 300 ms deadline the next batch flushes the window. Batch 1 was lost with + // the dropped window (the documented cost of commit batching); 2, 3 and 4 land once. + Thread.sleep(350L); + state.beginCommit(4L); + state.updateState(List.of(tuple(4L, "d")), null); + state.commit(4L); + + List names = new ArrayList<>(); + readRows().forEach(r -> names.add((String) r.getField("name"))); + names.sort(String::compareTo); + assertEquals(List.of("b", "c", "d"), names); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } finally { + state.close(); + } + } + @Test void requiredNullFieldFailsLoudly() { verifyCatalog.createTable(TABLE_ID, SCHEMA); From f27efc1cc267dcb32735ac8f3d33885feb67a1d4 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 26 Jul 2026 12:15:04 +0200 Subject: [PATCH 3/4] Fix examples iceberg (shuffling and max batches pending) --- ...bergPartitionedTridentExampleTopology.java | 55 +++++++++++++--- .../IcebergTridentExampleTopology.java | 64 ++++++++++++++++--- 2 files changed, 100 insertions(+), 19 deletions(-) diff --git a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedTridentExampleTopology.java b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedTridentExampleTopology.java index 2725244fa1c..8d062c1a64b 100644 --- a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedTridentExampleTopology.java +++ b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedTridentExampleTopology.java @@ -33,16 +33,30 @@ import org.apache.storm.iceberg.trident.IcebergStateFactory; import org.apache.storm.iceberg.trident.IcebergStateUpdater; import org.apache.storm.trident.TridentTopology; -import org.apache.storm.trident.testing.FixedBatchSpout; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Values; /** - * Writes a small fixed stream into a partitioned Iceberg table on a local Hadoop catalog. + * Ingests 10 million generated rows into a partitioned Iceberg table on a local Hadoop catalog, + * committing roughly every 1 MB written instead of once per Trident batch. * *

The table is partitioned by {@code identity(region)} and {@code days(event_time)}, so a single * batch spans several partitions and the state opens one data file per partition through the - * Iceberg fanout writer. + * Iceberg fanout writer. The threshold covers all four partitions together, so expect roughly + * four files per commit. + * + *

Commit batching weakens the delivery guarantee: batches buffered towards the + * next commit are lost if the worker dies, and a final partial window is never flushed, so the + * committed row count may end up slightly below the count ingested. See the unpartitioned example + * and {@code docs/storm-iceberg.md} for the full trade-off. + * + *

The state runs at {@value #STATE_PARALLELISM}-way parallelism: each partition buffers and + * commits independently against its own share of the stream, so the 1 MB threshold is crossed + * roughly {@value #STATE_PARALLELISM} times more slowly per partition than it would be at + * parallelism 1. + * + *

The row count and the commit threshold can be overridden on the command line: + * {@code }. * *

Run locally with: * {@code storm local storm-iceberg-examples-*.jar @@ -52,12 +66,21 @@ */ public final class IcebergPartitionedTridentExampleTopology { + private static final int BATCH_SIZE = 50_000; + private static final long TOTAL_TUPLES = 10_000_000L; + private static final long COMMIT_INTERVAL_BYTES = 1024L * 1024; + private static final int STATE_PARALLELISM = 4; + private static final String[] REGIONS = {"eu-west", "us-east"}; + private IcebergPartitionedTridentExampleTopology() { } public static void main(String[] args) throws Exception { String warehouse = args.length > 0 ? args[0] : "file:///tmp/storm-iceberg-warehouse"; String topologyName = args.length > 1 ? args[1] : "iceberg-partitioned-example"; + long totalTuples = args.length > 2 ? Long.parseLong(args[2]) : TOTAL_TUPLES; + long commitIntervalBytes = + args.length > 3 ? Long.parseLong(args[3]) : COMMIT_INTERVAL_BYTES; Schema schema = new Schema( Types.NestedField.required(1, "id", Types.LongType.get()), @@ -77,24 +100,36 @@ public static void main(String[] args) throws Exception { .withCatalogProperties(catalogProps) .withTable("example.events") .withAutoCreate(schema, spec) + .withCommitIntervalBytes(commitIntervalBytes) .build(); // Two regions x two days -> four partitions, written by a single state instance. Instant today = Instant.now(); Instant yesterday = today.minus(1, ChronoUnit.DAYS); - FixedBatchSpout spout = new FixedBatchSpout(new Fields("id", "region", "event_time"), 3, - new Values(1L, "eu-west", yesterday), new Values(2L, "us-east", yesterday), - new Values(3L, "eu-west", today), new Values(4L, "us-east", today), - new Values(5L, "eu-west", today), new Values(6L, "us-east", yesterday)); - spout.setCycle(false); + BoundedTupleSpout spout = new BoundedTupleSpout(totalTuples, BATCH_SIZE, + new Fields("id", "region", "event_time"), + index -> new Values(index, + REGIONS[(int) (index % REGIONS.length)], + index % 4 < 2 ? yesterday : today)); TridentTopology topology = new TridentTopology(); topology.newStream("events", spout) + // IBatchSpout has no notion of partition index: parallelizing the spout itself would + // make every task call emitBatch() for the same txid, each emitting the full batch. + // shuffle() repartitions after the (single-task) spout, so the persist stage below can + // run at a different parallelism. The hint has to be set on the TridentState returned + // by partitionPersist(), not on the Stream before it: partitionPersist() creates its + // own node that does not inherit a hint set upstream. + .shuffle() .partitionPersist(new IcebergStateFactory(options), - new Fields("id", "region", "event_time"), new IcebergStateUpdater()); + new Fields("id", "region", "event_time"), new IcebergStateUpdater()) + .parallelismHint(STATE_PARALLELISM); Config conf = new Config(); - conf.setMaxSpoutPending(3); + // In batches, not tuples: up to 8 Trident batches (400k tuples with BATCH_SIZE = 50_000) + // may be in flight, overlapping this batch's writes with the previous one's commit. + // Commits still land in txid order regardless of this setting. + conf.setMaxSpoutPending(8); StormSubmitter.submitTopology(topologyName, conf, topology.build()); } } diff --git a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergTridentExampleTopology.java b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergTridentExampleTopology.java index ecd45dafb7b..e5ef077c47f 100644 --- a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergTridentExampleTopology.java +++ b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergTridentExampleTopology.java @@ -31,25 +31,56 @@ import org.apache.storm.iceberg.trident.IcebergStateFactory; import org.apache.storm.iceberg.trident.IcebergStateUpdater; import org.apache.storm.trident.TridentTopology; -import org.apache.storm.trident.testing.FixedBatchSpout; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Values; /** - * Writes a small fixed stream into an Iceberg table on a local Hadoop catalog. + * Ingests 10 million generated rows into an Iceberg table on a local Hadoop catalog, committing + * roughly every 1 MB written instead of once per Trident batch. + * + *

Commit batching weakens the delivery guarantee. Trident treats a batch as + * delivered when {@code commit()} returns, so the batches buffered towards the next commit are + * lost if the worker dies. This example opts into that trade to keep the snapshot count sane: at + * one commit per batch, 10M rows in batches of {@value #BATCH_SIZE} would produce + * {@code 10_000_000 / BATCH_SIZE} snapshots. + * + *

For the same reason a final partial window is never flushed: no further batch arrives to + * cross a threshold. Whether any rows are left behind depends on where the last batch falls + * relative to the threshold, so the committed count may be at or slightly below the row count + * ingested. That is the documented behaviour of commit batching, not a defect of the example. + * + *

The state runs at {@value #STATE_PARALLELISM}-way parallelism: each of those partitions + * buffers and commits independently, sees roughly a quarter of the stream, and therefore takes + * about four times as many rows to cross the 1 MB threshold on its own. Expect roughly + * {@value #STATE_PARALLELISM} snapshots per flush round, not one. + * + *

The row count and the commit threshold can be overridden on the command line, which is the + * quick way to try the topology out: + * {@code }. * *

Run locally with: - * {@code storm local storm-iceberg-examples-*.jar org.apache.storm.iceberg.examples.IcebergTridentExampleTopology} + * {@code storm local storm-iceberg-examples-*.jar + * org.apache.storm.iceberg.examples.IcebergTridentExampleTopology} * then inspect the warehouse directory (default {@code file:///tmp/storm-iceberg-warehouse}). */ public final class IcebergTridentExampleTopology { + static final int BATCH_SIZE = 50_000; + private static final long TOTAL_TUPLES = 10_000_000L; + private static final long COMMIT_INTERVAL_BYTES = 1024L * 1024; + private static final int STATE_PARALLELISM = 4; + private static final String[] WORDS = + {"storm", "iceberg", "trident", "lakehouse", "parquet", "snapshot"}; + private IcebergTridentExampleTopology() { } public static void main(String[] args) throws Exception { String warehouse = args.length > 0 ? args[0] : "file:///tmp/storm-iceberg-warehouse"; String topologyName = args.length > 1 ? args[1] : "iceberg-example"; + long totalTuples = args.length > 2 ? Long.parseLong(args[2]) : TOTAL_TUPLES; + long commitIntervalBytes = + args.length > 3 ? Long.parseLong(args[3]) : COMMIT_INTERVAL_BYTES; Schema schema = new Schema( Types.NestedField.required(1, "id", Types.LongType.get()), @@ -63,19 +94,34 @@ public static void main(String[] args) throws Exception { .withCatalogProperties(catalogProps) .withTable("example.words") .withAutoCreate(schema, PartitionSpec.unpartitioned()) + .withCommitIntervalBytes(commitIntervalBytes) .build(); - FixedBatchSpout spout = new FixedBatchSpout(new Fields("id", "word"), 3, - new Values(1L, "storm"), new Values(2L, "iceberg"), new Values(3L, "trident"), - new Values(4L, "lakehouse"), new Values(5L, "parquet"), new Values(6L, "snapshot")); - spout.setCycle(false); + // A distinct word per row on purpose: a handful of repeated values would be dictionary + // encoded down to about a byte per row, and no realistic size threshold would ever be + // reached. High cardinality keeps the example representative of real ingestion. + BoundedTupleSpout spout = new BoundedTupleSpout(totalTuples, BATCH_SIZE, + new Fields("id", "word"), + index -> new Values(index, WORDS[(int) (index % WORDS.length)] + "-" + index)); TridentTopology topology = new TridentTopology(); topology.newStream("words", spout) - .partitionPersist(new IcebergStateFactory(options), new Fields("id", "word"), new IcebergStateUpdater()); + // IBatchSpout has no notion of partition index: parallelizing the spout itself would + // make every task call emitBatch() for the same txid, each emitting the full batch. + // shuffle() repartitions after the (single-task) spout, so the persist stage below can + // run at a different parallelism. The hint has to be set on the TridentState returned + // by partitionPersist(), not on the Stream before it: partitionPersist() creates its + // own node that does not inherit a hint set upstream. + .shuffle() + .partitionPersist(new IcebergStateFactory(options), + new Fields("id", "word"), new IcebergStateUpdater()) + .parallelismHint(STATE_PARALLELISM); Config conf = new Config(); - conf.setMaxSpoutPending(3); + // In batches, not tuples: up to 8 Trident batches (400k tuples with BATCH_SIZE = 50_000) + // may be in flight, overlapping this batch's writes with the previous one's commit. + // Commits still land in txid order regardless of this setting. + conf.setMaxSpoutPending(8); StormSubmitter.submitTopology(topologyName, conf, topology.build()); } } From ac94076fb03dd231d2f7fc2392929f45b15fd9e7 Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Sun, 26 Jul 2026 12:17:34 +0200 Subject: [PATCH 4/4] Add BoundedTupleSpout --- .../iceberg/examples/BoundedTupleSpout.java | 97 +++++++++++++++++++ 1 file changed, 97 insertions(+) create mode 100644 examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/BoundedTupleSpout.java diff --git a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/BoundedTupleSpout.java b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/BoundedTupleSpout.java new file mode 100644 index 00000000000..f359b524ead --- /dev/null +++ b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/BoundedTupleSpout.java @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.examples; + +import java.io.Serial; +import java.io.Serializable; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.trident.operation.TridentCollector; +import org.apache.storm.trident.spout.IBatchSpout; +import org.apache.storm.trident.topology.MasterBatchCoordinator; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.Values; + +/** + * Emits a bounded, deterministic sequence of generated tuples and then stops. + * + *

Each tuple's content is derived purely from its index, and the index range of a batch is + * derived purely from the batch id, so a replayed batch always carries the same tuples. That is + * what makes this spout usable with the exactly-once semantics of {@code IcebergState}: a + * cycling {@code FixedBatchSpout} would not be, and a fixed list cannot reach interesting volumes. + */ +public class BoundedTupleSpout implements IBatchSpout { + + @Serial + private static final long serialVersionUID = 1L; + + private final long totalTuples; + private final int batchSize; + private final Fields outputFields; + private final ValuesFactory valuesFactory; + + public BoundedTupleSpout(long totalTuples, int batchSize, Fields outputFields, + ValuesFactory valuesFactory) { + this.totalTuples = totalTuples; + this.batchSize = batchSize; + this.outputFields = outputFields; + this.valuesFactory = valuesFactory; + } + + @Override + public void open(Map conf, TopologyContext context) { + } + + @Override + public void emitBatch(long batchId, TridentCollector collector) { + // Trident transaction ids start at MasterBatchCoordinator.INIT_TXID (1). + long start = (batchId - MasterBatchCoordinator.INIT_TXID) * batchSize; + if (start < 0 || start >= totalTuples) { + return; + } + long end = Math.min(start + batchSize, totalTuples); + for (long index = start; index < end; index++) { + collector.emit(valuesFactory.create(index)); + } + } + + @Override + public void ack(long batchId) { + } + + @Override + public void close() { + } + + @Override + public Map getComponentConfiguration() { + return new HashMap<>(); + } + + @Override + public Fields getOutputFields() { + return outputFields; + } + + /** Builds the tuple at a given position in the sequence. */ + public interface ValuesFactory extends Serializable { + Values create(long index); + } +}