diff --git a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java index c511a7380dc1..d9c928ef6590 100644 --- a/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java +++ b/sdks/java/io/delta/src/main/java/org/apache/beam/sdk/io/delta/DeltaIO.java @@ -50,12 +50,40 @@ /** * A connector that reads from Delta Lake tables. * - *

This is work in progress. For more details and to track progress, please see Issue 21100. + *

{@link DeltaIO} is offered as a Managed transform. This class is subject to change and should + * not be used directly. Instead, use it like so: + * + *

{@code
+ * Map config = Map.of(
+ *         "table", "gs://my-bucket/delta-table",
+ *         "hadoop_config", Map.of(
+ *                 "fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
+ *                 "fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS",
+ *                 "fs.gs.project.id", "my-project-id"));
+ *
+ * pipeline
+ *     .apply(Managed.read(Managed.DELTA_LAKE).withConfig(config))
+ *     .getSinglePCollection()
+ *     .apply(ParDo.of(...));
+ * }
+ * + *

Configuration Options

+ * + * Please check the Managed IO + * configuration page for more details. + * + *

This is work in progress and is subject to change. For more details and to track progress, see + * Issue 21100. */ @Internal public class DeltaIO { + /** + * Reads rows from a Delta Lake table. + * + *

Normally, it is recommended to use {@link org.apache.beam.sdk.managed.Managed#read(String)} + * with {@code Managed.DELTA_LAKE} instead of directly using this transform. + */ public static ReadRows readRows() { return new AutoValue_DeltaIO_ReadRows.Builder().build(); } @@ -108,6 +136,15 @@ public PCollection expand(PBegin input) { if (path == null) { throw new IllegalArgumentException("Table path must be set."); } + if (getTimestamp() != null) { + throw new UnsupportedOperationException( + "Reading from a specific timestamp is not supported yet"); + } + + if (getVersion() != null) { + throw new UnsupportedOperationException( + "Reading from a specific version is not supported yet"); + } Configuration conf = new Configuration(); Map hadoopConfig = getHadoopConfig(); diff --git a/website/www/site/content/en/documentation/io/managed-io.md b/website/www/site/content/en/documentation/io/managed-io.md index 925f55aa52a5..7239832937a7 100644 --- a/website/www/site/content/en/documentation/io/managed-io.md +++ b/website/www/site/content/en/documentation/io/managed-io.md @@ -58,6 +58,18 @@ and Beam SQL is invoked via the Managed API under the hood. Read Configuration Write Configuration + + DELTA + + table (str)
+ hadoop_config (map[str, str])
+ timestamp (str)
+ version (int64)
+ + + Unavailable + + ICEBERG @@ -238,6 +250,62 @@ and Beam SQL is invoked via the Managed API under the hood. ## Configuration Details +### `DELTA` Read + +

+ + + + + + + + + + + + + + + + + + + + + + + + + + +
ConfigurationTypeDescription
+ table + + str + + Identifier of the Delta Lake table. +
+ hadoop_config + + map[str, str] + + Properties passed to the Hadoop Configuration. +
+ timestamp + + str + + Timestamp of the Delta Lake table to read. +
+ version + + int64 + + Version of the Delta Lake table to read. +
+
+ ### `ICEBERG` Read