You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2022/01/28 19:41:12 UTC
[iceberg] branch master updated: Spark: Add option to skip overwrites in streaming (#3517)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new a7bbe4d Spark: Add option to skip overwrites in streaming (#3517)
a7bbe4d is described below
commit a7bbe4d5674ff667dcb0619e6fa7b5845abd9036
Author: Sreeram Garlapati <Sr...@users.noreply.github.com>
AuthorDate: Fri Jan 28 11:40:52 2022 -0800
Spark: Add option to skip overwrites in streaming (#3517)
Co-authored-by: Rajarshi Sarkar <sr...@amazon.com>
---
.../org/apache/iceberg/spark/SparkReadConf.java | 7 ++++++
.../org/apache/iceberg/spark/SparkReadOptions.java | 4 +++
.../spark/source/SparkMicroBatchStream.java | 29 ++++++++++++++++------
.../spark/source/TestStructuredStreamingRead3.java | 25 +++++++++++++++++++
4 files changed, 58 insertions(+), 7 deletions(-)
diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
index a4748df..c7c0175 100644
--- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
+++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java
@@ -116,6 +116,13 @@ public class SparkReadConf {
.parse();
}
+ public boolean streamingSkipOverwriteSnapshots() {
+ return confParser.booleanConf()
+ .option(SparkReadOptions.STREAMING_SKIP_OVERWRITE_SNAPSHOTS)
+ .defaultValue(SparkReadOptions.STREAMING_SKIP_OVERWRITE_SNAPSHOTS_DEFAULT)
+ .parse();
+ }
+
public boolean parquetVectorizationEnabled() {
return confParser.booleanConf()
.option(SparkReadOptions.VECTORIZATION_ENABLED)
diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadOptions.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadOptions.java
index 2d1c8cd..edcc230 100644
--- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadOptions.java
+++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadOptions.java
@@ -61,6 +61,10 @@ public class SparkReadOptions {
public static final String STREAMING_SKIP_DELETE_SNAPSHOTS = "streaming-skip-delete-snapshots";
public static final boolean STREAMING_SKIP_DELETE_SNAPSHOTS_DEFAULT = false;
+ // skip snapshots of type overwrite while reading stream out of iceberg table
+ public static final String STREAMING_SKIP_OVERWRITE_SNAPSHOTS = "streaming-skip-overwrite-snapshots";
+ public static final boolean STREAMING_SKIP_OVERWRITE_SNAPSHOTS_DEFAULT = false;
+
// Controls whether to allow reading timestamps without zone info
public static final String HANDLE_TIMESTAMP_WITHOUT_TIMEZONE = "handle-timestamp-without-timezone";
diff --git a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
index 6311c48..dd68022 100644
--- a/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
+++ b/spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
@@ -77,6 +77,7 @@ public class SparkMicroBatchStream implements MicroBatchStream {
private final boolean localityPreferred;
private final StreamingOffset initialOffset;
private final boolean skipDelete;
+ private final boolean skipOverwrite;
private final Long fromTimestamp;
SparkMicroBatchStream(JavaSparkContext sparkContext, Table table, SparkReadConf readConf, boolean caseSensitive,
@@ -95,6 +96,7 @@ public class SparkMicroBatchStream implements MicroBatchStream {
this.initialOffset = initialOffsetStore.initialOffset();
this.skipDelete = readConf.streamingSkipDeleteSnapshots();
+ this.skipOverwrite = readConf.streamingSkipOverwriteSnapshots();
}
@Override
@@ -200,13 +202,26 @@ public class SparkMicroBatchStream implements MicroBatchStream {
private boolean shouldProcess(Snapshot snapshot) {
String op = snapshot.operation();
- Preconditions.checkState(!op.equals(DataOperations.DELETE) || skipDelete,
- "Cannot process delete snapshot: %s, to ignore deletes, set %s=true.",
- snapshot.snapshotId(), SparkReadOptions.STREAMING_SKIP_DELETE_SNAPSHOTS);
- Preconditions.checkState(
- op.equals(DataOperations.DELETE) || op.equals(DataOperations.APPEND) || op.equals(DataOperations.REPLACE),
- "Cannot process %s snapshot: %s", op.toLowerCase(Locale.ROOT), snapshot.snapshotId());
- return op.equals(DataOperations.APPEND);
+ switch (op) {
+ case DataOperations.APPEND:
+ return true;
+ case DataOperations.REPLACE:
+ return false;
+ case DataOperations.DELETE:
+ Preconditions.checkState(skipDelete,
+ "Cannot process delete snapshot: %s, to ignore deletes, set %s=true",
+ snapshot.snapshotId(), SparkReadOptions.STREAMING_SKIP_DELETE_SNAPSHOTS);
+ return false;
+ case DataOperations.OVERWRITE:
+ Preconditions.checkState(skipOverwrite,
+ "Cannot process overwrite snapshot: %s, to ignore overwrites, set %s=true",
+ snapshot.snapshotId(), SparkReadOptions.STREAMING_SKIP_OVERWRITE_SNAPSHOTS);
+ return false;
+ default:
+ throw new IllegalStateException(String.format(
+ "Cannot process unknown snapshot operation: %s (snapshot id %s)",
+ op.toLowerCase(Locale.ROOT), snapshot.snapshotId()));
+ }
}
private static StreamingOffset determineStartingOffset(Table table, Long fromTimestamp) {
diff --git a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
index 145cf78..fb0b903 100644
--- a/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
+++ b/spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreamingRead3.java
@@ -481,6 +481,31 @@ public final class TestStructuredStreamingRead3 extends SparkCatalogTestBase {
.containsExactlyInAnyOrderElementsOf(Iterables.concat(dataAcrossSnapshots));
}
+ @Test
+ public void testReadStreamWithSnapshotTypeDeleteAndSkipOverwriteOption() throws Exception {
+ table.updateSpec()
+ .removeField("id_bucket")
+ .addField(ref("id"))
+ .commit();
+
+ // fill table with some data
+ List<List<SimpleRecord>> dataAcrossSnapshots = TEST_DATA_MULTIPLE_SNAPSHOTS;
+ appendDataAsMultipleSnapshots(dataAcrossSnapshots);
+
+ // this should create a snapshot with type overwrite.
+ table.newOverwrite()
+ .overwriteByRowFilter(Expressions.greaterThan("id", 4))
+ .commit();
+
+ // check pre-condition - that the above delete operation on table resulted in Snapshot of Type OVERWRITE.
+ table.refresh();
+ Assert.assertEquals(DataOperations.OVERWRITE, table.currentSnapshot().operation());
+
+ StreamingQuery query = startStream(SparkReadOptions.STREAMING_SKIP_OVERWRITE_SNAPSHOTS, "true");
+ Assertions.assertThat(rowsAvailable(query))
+ .containsExactlyInAnyOrderElementsOf(Iterables.concat(dataAcrossSnapshots));
+ }
+
/**
* appends each list as a Snapshot on the iceberg table at the given location.
* accepts a list of lists - each list representing data per snapshot.