From 66fc208682c844cb5ba6262a0b78e25b9d0f6dbb Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Mon, 3 Aug 2026 21:01:44 +0000 Subject: [PATCH 01/26] add delta lake to iceberg IT --- .../yaml/DeltaLakeToIcebergYaml.java | 164 ++++++++++++++++ .../python/options/deltalake_options.yaml | 18 ++ .../yaml/DeltaLakeToIcebergYamlIT.java | 180 ++++++++++++++++++ 3 files changed, 362 insertions(+) create mode 100644 yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java create mode 100644 yaml/src/main/python/options/deltalake_options.yaml create mode 100644 yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java new file mode 100644 index 0000000000..1649c5a6e5 --- /dev/null +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java @@ -0,0 +1,164 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.templates.yaml; + +import com.google.cloud.teleport.metadata.Template; +import com.google.cloud.teleport.metadata.TemplateCategory; +import com.google.cloud.teleport.metadata.TemplateParameter; +import org.apache.beam.sdk.options.Validation; + +@Template( + name = "DeltaLake_To_Iceberg_Yaml", + category = TemplateCategory.BATCH, + type = Template.TemplateType.YAML, + displayName = "Delta Lake to Iceberg (YAML)", + description = "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", + flexContainerName = "pipeline-yaml", + yamlTemplateFile = "DeltaLakeToIceberg.yaml", + filesToCopy = {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/iceberg_options.yaml"}, + documentation = "", + contactInformation = "https://cloud.google.com/support", + requirements = {"The Input Delta Lake table must exist and be accessible.", + "The Output Iceberg table must exist or be created, and the warehouse must be accessible." + }, + streaming = false, + hidden = false) +public interface DeltaLakeToIcebergYaml { + + @TemplateParameter.Text( + order = 1, + name = "deltaLakeTable", + optional = false, + description = "A GCS path to the Delta Lake table.", + helpText = "The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table.", + example = "gs://your-bucket/path/to/table" + ) + @Validation.Required + String getDeltaLakeTable(); + + @TemplateParameter.Text( + order = 2, + name = "deltaLakeHadoopConfig", + optional = true, + description = "Properties passed to Hadoop Configuration.", + helpText = "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs.", + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" + ) + String getDeltaLakeHadoopConfig(); + + @TemplateParameter.Text( + order = 3, + name = "table", + optional = false, + description = "A fully-qualified table identifier.", + helpText = "A fully-qualified table identifier, e.g., my_dataset.my_table.", + example = "my_dataset.my_table" + ) + @Validation.Required + String getTable(); + + @TemplateParameter.Text( + order = 4, + name = "catalogName", + optional = false, + description = "Name of the catalog containing the table.", + helpText = "The name of the Iceberg catalog that contains the table.", + example = "my_hadoop_catalog" + ) + @Validation.Required + String getCatalogName(); + + @TemplateParameter.Text( + order = 5, + name = "catalogProperties", + optional = false, + description = "Properties used to set up the Iceberg catalog.", + helpText = "A map of properties for setting up the Iceberg catalog.", + example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}" + ) + @Validation.Required + String getCatalogProperties(); + + @TemplateParameter.Text( + order = 6, + name = "configProperties", + optional = true, + description = "Properties passed to the Hadoop Configuration.", + helpText = "A map of properties to pass to the Hadoop Configuration.", + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" + ) + String getConfigProperties(); + + @TemplateParameter.Text( + order = 7, + name = "drop", + optional = true, + description = "A list of field names to drop from the input record before writing.", + helpText = "A list of field names to drop. Mutually exclusive with 'keep' and 'only'.", + example = "[\"field_to_drop_1\", \"field_to_drop_2\"]" + ) + String getDrop(); + + @TemplateParameter.Text( + order = 8, + name = "filter", + optional = true, + description = "An optional filter expression to apply to the input records.", + helpText = "A filter expression to apply to records from the Iceberg table.", + example = "age > 18" + ) + String getFilter(); + + @TemplateParameter.Text( + order = 9, + name = "keep", + optional = true, + description = "A list of field names to keep in the input record.", + helpText = "A list of field names to keep. Mutually exclusive with 'drop' and 'only'.", + example = "[\"field_to_keep_1\", \"field_to_keep_2\"]" + ) + String getKeep(); + + @TemplateParameter.Text( + order = 10, + name = "only", + optional = true, + description = "The name of a single record field that should be written.", + helpText = "The name of a single field to write. Mutually exclusive with 'keep' and 'drop'.", + example = "my_record_field" + ) + String getOnly(); + + @TemplateParameter.Text( + order = 11, + name = "partitionFields", + optional = true, + description = "Fields used to create a partition spec for new tables.", + helpText = "A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category'].", + example = "[\"day(ts)\", \"bucket(id, 4)\"]" + ) + String getPartitionFields(); + + @TemplateParameter.Text( + order = 12, + name = "tableProperties", + optional = true, + description = "Iceberg table properties to be set on table creation.", + helpText = "A map of Iceberg table properties to set when the table is created.", + example = "{\"commit.retry.num-retries\": \"2\"}" + ) + String getTableProperties(); +} diff --git a/yaml/src/main/python/options/deltalake_options.yaml b/yaml/src/main/python/options/deltalake_options.yaml new file mode 100644 index 0000000000..26940eb9b6 --- /dev/null +++ b/yaml/src/main/python/options/deltalake_options.yaml @@ -0,0 +1,18 @@ +options: + - name: "deltalake_read_options" + parameters: + - order: 1 + name: "deltaLakeTable" + description: "A GCS path to the Delta Lake table." + help: "The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table." + example: "gs://your-bucket/path/to/table" + required: true + type: text + - order: 2 + name: "deltaLakeHadoopConfig" + description: "Properties passed to Hadoop Configuration." + help: "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs." + example: '{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}' + required: false + type: map + diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java new file mode 100644 index 0000000000..684961ea4d --- /dev/null +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java @@ -0,0 +1,180 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.templates.yaml; + +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; +import static org.junit.Assert.assertEquals; + +import com.google.cloud.teleport.it.iceberg.IcebergResourceManager; +import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; +import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import org.apache.beam.it.common.PipelineLauncher.LaunchConfig; +import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; +import org.apache.beam.it.common.PipelineOperator; +import org.apache.beam.it.common.utils.ResourceManagerUtils; +import org.apache.beam.it.gcp.TemplateTestBase; +import org.apache.beam.it.gcp.artifacts.utils.ParquetTestUtil; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.types.Types; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Integration test for {@link DeltaLakeToIcebergYaml} template. */ +@Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) +@TemplateIntegrationTest(DeltaLakeToIcebergYaml.class) +@RunWith(JUnit4.class) +public class DeltaLakeToIcebergYamlIT extends TemplateTestBase { + + private IcebergResourceManager icebergResourceManager; + + private static final String CATALOG_NAME = "hadoop_catalog"; + private final String namespace = + "deltalake_iceberg_ns_" + UUID.randomUUID().toString().replace("-", ""); + private static final String ICEBERG_TABLE_NAME = "iceberg_table"; + private final String icebergTableIdentifier = namespace + "." + ICEBERG_TABLE_NAME; + + @Before + public void setUp() throws IOException { + gcsClient.registerTempDir(namespace); + + // Initialize Iceberg resource manager + icebergResourceManager = + IcebergResourceManager.builder(testName) + .setCatalogName(CATALOG_NAME) + .setCatalogProperties(getCatalogProperties()) + .build(); + } + + @After + public void tearDown() { + ResourceManagerUtils.cleanResources(icebergResourceManager); + } + + @Test + public void testDeltaLakeToIceberg() throws IOException { + // 1. Arrange: Create Delta Lake source table in GCS + String deltaTableDir = "delta-table"; + org.apache.avro.Schema avroSchema = + new org.apache.avro.Schema.Parser() + .parse( + "{\"type\":\"record\",\"name\":\"test_record\",\"fields\":[" + + "{\"name\":\"id\",\"type\":\"string\"}," + + "{\"name\":\"state\",\"type\":\"string\"}," + + "{\"name\":\"price\",\"type\":\"double\"}" + + "]}"); + org.apache.avro.generic.GenericRecord avroRecord = + new org.apache.avro.generic.GenericData.Record(avroSchema); + avroRecord.put("id", "007"); + avroRecord.put("state", "CA"); + avroRecord.put("price", 26.23); + byte[] parquetBytes = + ParquetTestUtil.createParquetFile(avroSchema, List.of(avroRecord)); + + // Upload data Parquet file + gcsClient.createArtifact(deltaTableDir + "/part-00000.parquet", parquetBytes); + + // Create and upload Delta Lake transaction log + String commitContent = + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" + + "{\"metaData\":{\"id\":\"test-id\",\"format\":{\"provider\":\"parquet\",\"options\":{}}," + + "\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"state\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"price\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + + "]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":123456789}}\n" + + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":774," + + "\"modificationTime\":123456789,\"dataChange\":true}}"; + + gcsClient.createArtifact( + deltaTableDir + "/_delta_log/00000000000000000000.json", commitContent); + + String deltaTableGcsPath = getGcsPath(deltaTableDir); + + // 2. Arrange: Create destination Iceberg table + icebergResourceManager.createNamespace(namespace); + Schema icebergSchema = + new Schema( + Types.NestedField.required(1, "id", Types.StringType.get()), + Types.NestedField.required(2, "state", Types.StringType.get()), + Types.NestedField.required(3, "price", Types.DoubleType.get())); + icebergResourceManager.createTable(icebergTableIdentifier, icebergSchema); + + // 3. Act: Configure options and launch template + LaunchConfig.Builder options = + LaunchConfig.builder(testName, specPath) + .addParameter("deltaLakeTable", deltaTableGcsPath) + .addParameter( + "deltaLakeHadoopConfig", + new org.json.JSONObject(getGcsHadoopConfig()).toString()) + .addParameter("table", icebergTableIdentifier) + .addParameter("catalogName", CATALOG_NAME) + .addParameter( + "catalogProperties", new org.json.JSONObject(getCatalogProperties()).toString()); + + LaunchInfo info = launchTemplate(options); + assertThatPipeline(info).isRunning(); + + PipelineOperator.Result result = pipelineOperator().waitUntilDone(createConfig(info)); + + // 4. Assert + assertThatResult(result).isLaunchFinished(); + + List records = icebergResourceManager.read(icebergTableIdentifier); + assertEquals(1, records.size()); + + Record record = records.get(0); + assertEquals("007", record.getField("id")); + assertEquals("CA", record.getField("state")); + assertEquals(26.23, record.getField("price")); + } + + @Override + protected PipelineOperator.Config createConfig(LaunchInfo info) { + return PipelineOperator.Config.builder() + .setJobId(info.jobId()) + .setProject(PROJECT) + .setRegion(REGION) + .build(); + } + + private Map getCatalogProperties() { + return Map.of( + "type", "rest", + "uri", "https://biglake.googleapis.com/iceberg/v1beta/restcatalog", + "warehouse", "gs://" + gcsClient.getBucket(), + "header.x-goog-user-project", PROJECT, + "rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager", + "rest-metrics-reporting-enabled", "false"); + } + + private Map getGcsHadoopConfig() { + return 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.auth.type", "APPLICATION_DEFAULT", + "fs.gs.project.id", PROJECT); + } +} From 92aadfaa0ccbeeb9a4354fed49bf1e88b1706837 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Mon, 3 Aug 2026 21:14:38 +0000 Subject: [PATCH 02/26] fix spotless and rerun auto-generation --- .../yaml/DeltaLakeToIcebergYaml.java | 51 +++++++++---------- .../yaml/DeltaLakeToIcebergYamlIT.java | 6 +-- 2 files changed, 25 insertions(+), 32 deletions(-) diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java index 1649c5a6e5..6052128284 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java @@ -24,14 +24,21 @@ name = "DeltaLake_To_Iceberg_Yaml", category = TemplateCategory.BATCH, type = Template.TemplateType.YAML, - displayName = "Delta Lake to Iceberg (YAML)", - description = "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", + displayName = "Delta Lake to Lakehouse", + description = + "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", flexContainerName = "pipeline-yaml", yamlTemplateFile = "DeltaLakeToIceberg.yaml", - filesToCopy = {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/iceberg_options.yaml"}, + filesToCopy = { + "main.py", + "requirements.txt", + "options/deltalake_options.yaml", + "options/iceberg_options.yaml" + }, documentation = "", contactInformation = "https://cloud.google.com/support", - requirements = {"The Input Delta Lake table must exist and be accessible.", + requirements = { + "The Input Delta Lake table must exist and be accessible.", "The Output Iceberg table must exist or be created, and the warehouse must be accessible." }, streaming = false, @@ -44,8 +51,7 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "A GCS path to the Delta Lake table.", helpText = "The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table.", - example = "gs://your-bucket/path/to/table" - ) + example = "gs://your-bucket/path/to/table") @Validation.Required String getDeltaLakeTable(); @@ -55,8 +61,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Properties passed to Hadoop Configuration.", helpText = "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs.", - example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" - ) + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}") String getDeltaLakeHadoopConfig(); @TemplateParameter.Text( @@ -65,8 +70,7 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "A fully-qualified table identifier.", helpText = "A fully-qualified table identifier, e.g., my_dataset.my_table.", - example = "my_dataset.my_table" - ) + example = "my_dataset.my_table") @Validation.Required String getTable(); @@ -76,8 +80,7 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "Name of the catalog containing the table.", helpText = "The name of the Iceberg catalog that contains the table.", - example = "my_hadoop_catalog" - ) + example = "my_hadoop_catalog") @Validation.Required String getCatalogName(); @@ -87,8 +90,7 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "Properties used to set up the Iceberg catalog.", helpText = "A map of properties for setting up the Iceberg catalog.", - example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}" - ) + example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}") @Validation.Required String getCatalogProperties(); @@ -98,8 +100,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Properties passed to the Hadoop Configuration.", helpText = "A map of properties to pass to the Hadoop Configuration.", - example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" - ) + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}") String getConfigProperties(); @TemplateParameter.Text( @@ -108,8 +109,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "A list of field names to drop from the input record before writing.", helpText = "A list of field names to drop. Mutually exclusive with 'keep' and 'only'.", - example = "[\"field_to_drop_1\", \"field_to_drop_2\"]" - ) + example = "[\"field_to_drop_1\", \"field_to_drop_2\"]") String getDrop(); @TemplateParameter.Text( @@ -118,8 +118,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "An optional filter expression to apply to the input records.", helpText = "A filter expression to apply to records from the Iceberg table.", - example = "age > 18" - ) + example = "age > 18") String getFilter(); @TemplateParameter.Text( @@ -128,8 +127,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "A list of field names to keep in the input record.", helpText = "A list of field names to keep. Mutually exclusive with 'drop' and 'only'.", - example = "[\"field_to_keep_1\", \"field_to_keep_2\"]" - ) + example = "[\"field_to_keep_1\", \"field_to_keep_2\"]") String getKeep(); @TemplateParameter.Text( @@ -138,8 +136,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "The name of a single record field that should be written.", helpText = "The name of a single field to write. Mutually exclusive with 'keep' and 'drop'.", - example = "my_record_field" - ) + example = "my_record_field") String getOnly(); @TemplateParameter.Text( @@ -148,8 +145,7 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Fields used to create a partition spec for new tables.", helpText = "A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category'].", - example = "[\"day(ts)\", \"bucket(id, 4)\"]" - ) + example = "[\"day(ts)\", \"bucket(id, 4)\"]") String getPartitionFields(); @TemplateParameter.Text( @@ -158,7 +154,6 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Iceberg table properties to be set on table creation.", helpText = "A map of Iceberg table properties to set when the table is created.", - example = "{\"commit.retry.num-retries\": \"2\"}" - ) + example = "{\"commit.retry.num-retries\": \"2\"}") String getTableProperties(); } diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java index 684961ea4d..d7e8e80e2e 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java @@ -90,8 +90,7 @@ public void testDeltaLakeToIceberg() throws IOException { avroRecord.put("id", "007"); avroRecord.put("state", "CA"); avroRecord.put("price", 26.23); - byte[] parquetBytes = - ParquetTestUtil.createParquetFile(avroSchema, List.of(avroRecord)); + byte[] parquetBytes = ParquetTestUtil.createParquetFile(avroSchema, List.of(avroRecord)); // Upload data Parquet file gcsClient.createArtifact(deltaTableDir + "/part-00000.parquet", parquetBytes); @@ -127,8 +126,7 @@ public void testDeltaLakeToIceberg() throws IOException { LaunchConfig.builder(testName, specPath) .addParameter("deltaLakeTable", deltaTableGcsPath) .addParameter( - "deltaLakeHadoopConfig", - new org.json.JSONObject(getGcsHadoopConfig()).toString()) + "deltaLakeHadoopConfig", new org.json.JSONObject(getGcsHadoopConfig()).toString()) .addParameter("table", icebergTableIdentifier) .addParameter("catalogName", CATALOG_NAME) .addParameter( From 9caddc275da39de4ed356b1176c67ad9be01063b Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Wed, 5 Aug 2026 15:42:56 +0000 Subject: [PATCH 03/26] switch to interrupt call --- .../org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java index efce55e16a..8041dc7a80 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java +++ b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java @@ -265,7 +265,7 @@ public void cancel() { currentJob.setCurrentState(JobState.CANCELLED.toString()); try { - this.stop(); + this.interrupt(); } catch (Exception e) { LOG.warn("Error cancelling job", e); } From ece2e53a09e5a033d11e7d4c7a4fe4f46bde7054 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Mon, 17 Aug 2026 16:42:42 +0000 Subject: [PATCH 04/26] change to a dynamic import to fix error --- .../src/main/python/generate_yaml_java_templates.py | 13 +++++++++++++ yaml/src/main/python/java.tmpl | 6 +----- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/yaml/src/main/python/generate_yaml_java_templates.py b/yaml/src/main/python/generate_yaml_java_templates.py index 71082760be..0e8b23b4fc 100644 --- a/yaml/src/main/python/generate_yaml_java_templates.py +++ b/yaml/src/main/python/generate_yaml_java_templates.py @@ -135,6 +135,7 @@ def generate_java_interface(yaml_path, java_path): # Build the parameters code parameters_code = [] + has_defaults = False for i, param in enumerate(flat_parameters): param_name = param['name'] java_type = JAVA_TYPE_BY_YAML_TYPE.get(param.get('type', 'text'), 'String') @@ -160,6 +161,7 @@ def generate_java_interface(yaml_path, java_path): # default param if 'default' in param: + has_defaults = True if java_type == 'String': param_code += f' @Default.String("{param["default"]}")\n' else: @@ -170,6 +172,16 @@ def generate_java_interface(yaml_path, java_path): parameters_code.append(param_code) + imports = [ + "import com.google.cloud.teleport.metadata.Template;", + "import com.google.cloud.teleport.metadata.TemplateCategory;", + "import com.google.cloud.teleport.metadata.TemplateParameter;", + ] + if has_defaults: + imports.append("import org.apache.beam.sdk.options.Default;") + imports.append("import org.apache.beam.sdk.options.Validation;") + imports_code = "\n".join(imports) + # Format requirements for Java array reqs = template_info.get('requirements', []) reqs_formatted = "{}" @@ -203,6 +215,7 @@ def generate_java_interface(yaml_path, java_path): template_info_streaming=str(template_info.get('streaming', False)).lower(), template_info_hidden=str(template_info.get('hidden', False)).lower(), class_name=class_name, + imports=imports_code, parameters='\n'.join(parameters_code), ) diff --git a/yaml/src/main/python/java.tmpl b/yaml/src/main/python/java.tmpl index 9e2810de9d..a8210a5231 100644 --- a/yaml/src/main/python/java.tmpl +++ b/yaml/src/main/python/java.tmpl @@ -15,11 +15,7 @@ */ package com.google.cloud.teleport.templates.yaml; -import com.google.cloud.teleport.metadata.Template; -import com.google.cloud.teleport.metadata.TemplateCategory; -import com.google.cloud.teleport.metadata.TemplateParameter; -import org.apache.beam.sdk.options.Default; -import org.apache.beam.sdk.options.Validation; +{imports} @Template( name = "{template_info_name}", From 58cc7ccbafb872b747cf417632e8a84dd07c1c94 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Mon, 17 Aug 2026 18:59:09 +0000 Subject: [PATCH 05/26] add forgotten yaml definition file --- yaml/src/main/yaml/DeltaLakeToIceberg.yaml | 75 ++++++++++++++++++++++ 1 file changed, 75 insertions(+) create mode 100644 yaml/src/main/yaml/DeltaLakeToIceberg.yaml diff --git a/yaml/src/main/yaml/DeltaLakeToIceberg.yaml b/yaml/src/main/yaml/DeltaLakeToIceberg.yaml new file mode 100644 index 0000000000..c4651dd2f2 --- /dev/null +++ b/yaml/src/main/yaml/DeltaLakeToIceberg.yaml @@ -0,0 +1,75 @@ +template: + name: "DeltaLake_To_Iceberg_Yaml" + category: "BATCH" + type: "YAML" + display_name: "Delta Lake to Lakehouse" + description: > + The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table + and outputs the records to an Apache Iceberg table. + flex_container_name: "pipeline-yaml" + yamlTemplateFile: "DeltaLakeToIceberg.yaml" + filesToCopy: > + {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/iceberg_options.yaml"} + contactInformation: "https://cloud.google.com/support" + requirements: { + "The Input Delta Lake table must exist and be accessible.", + "The Output Iceberg table must exist or be created, and the warehouse must be accessible." + } + streaming: false + hidden: false + + options_file: + - "deltalake_options" + - "iceberg_options" + + parameters: + - deltalake_read_options + - iceberg_common_options + - iceberg_write_options + +pipeline: + type: chain + transforms: + - type: ReadFromDeltaLake + name: ReadFromDeltaLake + config: + table: "{{ deltaLakeTable }}" + {% if deltaLakeHadoopConfig %} + hadoop_config: {{ deltaLakeHadoopConfig }} + {% endif %} + + - type: WriteToIceberg + name: WriteToIceberg + config: + table: "{{ table }}" + catalog_name: "{{ catalogName }}" + catalog_properties: {{ catalogProperties }} + {% if configProperties %} + config_properties: {{ configProperties }} + {% endif %} + {% if drop %} + drop: {{ drop }} + {% endif %} + {% if keep %} + keep: {{ keep }} + {% endif %} + {% if only %} + only: {{ only }} + {% endif %} + {% if partitionFields %} + partition_fields: {{ partitionFields }} + {% endif %} + {% if tableProperties %} + table_properties: {{ tableProperties }} + {% endif %} + +providers: + - type: pythonPackage + config: + packages: + - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-07-20/job_builder_util_transforms-0.2.0.tar.gz + transforms: + ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" + +options: + streaming: false From 6f0eb7d873dee911cf4020bd101da14d7c6442c4 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Mon, 17 Aug 2026 19:09:26 +0000 Subject: [PATCH 06/26] add generated README for DeltaLakeToIceberg template --- yaml/README_DeltaLake_To_Iceberg_Yaml.md | 254 +++++++++++++++++++++++ 1 file changed, 254 insertions(+) create mode 100644 yaml/README_DeltaLake_To_Iceberg_Yaml.md diff --git a/yaml/README_DeltaLake_To_Iceberg_Yaml.md b/yaml/README_DeltaLake_To_Iceberg_Yaml.md new file mode 100644 index 0000000000..6cae1b0840 --- /dev/null +++ b/yaml/README_DeltaLake_To_Iceberg_Yaml.md @@ -0,0 +1,254 @@ + +Delta Lake to Lakehouse template +--- +The Delta Lake to Iceberg template is a batch pipeline that reads data from a +Delta Lake table and outputs the records to an Apache Iceberg table. + + + +:bulb: This is a generated documentation based +on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/contributor-docs/code-contributions.md#metadata-annotations) +. Do not change this file directly. + +## Parameters + +### Required parameters + +* **deltaLakeTable**: The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table. For example, `gs://your-bucket/path/to/table`. +* **table**: A fully-qualified table identifier, e.g., my_dataset.my_table. For example, `my_dataset.my_table`. +* **catalogName**: The name of the Iceberg catalog that contains the table. For example, `my_hadoop_catalog`. +* **catalogProperties**: A map of properties for setting up the Iceberg catalog. For example, `{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}`. + +### Optional parameters + +* **deltaLakeHadoopConfig**: A map of properties to pass to Hadoop Configuration, e.g. key-value pairs. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. +* **configProperties**: A map of properties to pass to the Hadoop Configuration. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. +* **drop**: A list of field names to drop. Mutually exclusive with 'keep' and 'only'. For example, `["field_to_drop_1", "field_to_drop_2"]`. +* **filter**: A filter expression to apply to records from the Iceberg table. For example, `age > 18`. +* **keep**: A list of field names to keep. Mutually exclusive with 'drop' and 'only'. For example, `["field_to_keep_1", "field_to_keep_2"]`. +* **only**: The name of a single field to write. Mutually exclusive with 'keep' and 'drop'. For example, `my_record_field`. +* **partitionFields**: A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category']. For example, `["day(ts)", "bucket(id, 4)"]`. +* **tableProperties**: A map of Iceberg table properties to set when the table is created. For example, `{"commit.retry.num-retries": "2"}`. + + + +## Getting Started + +### Requirements + +* Java 17 +* Maven +* [gcloud CLI](https://cloud.google.com/sdk/gcloud), and execution of the + following commands: + * `gcloud auth login` + * `gcloud auth application-default login` + +:star2: Those dependencies are pre-installed if you use Google Cloud Shell! + +[![Open in Cloud Shell](http://gstatic.com/cloudssh/images/open-btn.svg)](https://console.cloud.google.com/cloudshell/editor?cloudshell_git_repo=https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2FDataflowTemplates.git&cloudshell_open_in_editor=yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java) + +### Templates Plugin + +This README provides instructions using +the [Templates Plugin](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/contributor-docs/code-contributions.md#templates-plugin). + +#### Validating the Template + +This template has a validation command that is used to check code quality. + +```shell +mvn clean install -PtemplatesValidate \ +-DskipTests -am \ +-pl yaml +``` + +### Building Template + +This template is a Flex Template, meaning that the pipeline code will be +containerized and the container will be executed on Dataflow. Please +check [Use Flex Templates](https://cloud.google.com/dataflow/docs/guides/templates/using-flex-templates) +and [Configure Flex Templates](https://cloud.google.com/dataflow/docs/guides/templates/configuring-flex-templates) +for more information. + +#### Staging the Template + +If the plan is to just stage the template (i.e., make it available to use) by +the `gcloud` command or Dataflow "Create job from template" UI, +the `-PtemplatesStage` profile should be used: + +```shell +export PROJECT= +export BUCKET_NAME= +export ARTIFACT_REGISTRY_REPO=-docker.pkg.dev/$PROJECT/ + +mvn clean package -PtemplatesStage \ +-DskipTests \ +-DprojectId="$PROJECT" \ +-DbucketName="$BUCKET_NAME" \ +-DartifactRegistry="$ARTIFACT_REGISTRY_REPO" \ +-DstagePrefix="templates" \ +-DtemplateName="DeltaLake_To_Iceberg_Yaml" \ +-f yaml +``` + +The `-DartifactRegistry` parameter can be specified to set the artifact registry repository of the Flex Templates image. +If not provided, it defaults to `gcr.io/`. + +The command should build and save the template to Google Cloud, and then print +the complete location on Cloud Storage: + +``` +Flex Template was staged! gs:///templates/flex/DeltaLake_To_Iceberg_Yaml +``` + +The specific path should be copied as it will be used in the following steps. + +#### Running the Template + +**Using the staged template**: + +You can use the path above run the template (or share with others for execution). + +To start a job with the template at any time using `gcloud`, you are going to +need valid resources for the required parameters. + +Provided that, the following command line can be used: + +```shell +export PROJECT= +export BUCKET_NAME= +export REGION=us-central1 +export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/DeltaLake_To_Iceberg_Yaml" + +### Required +export DELTA_LAKE_TABLE= +export TABLE= +export CATALOG_NAME= +export CATALOG_PROPERTIES= + +### Optional +export DELTA_LAKE_HADOOP_CONFIG= +export CONFIG_PROPERTIES= +export DROP= +export FILTER= +export KEEP= +export ONLY= +export PARTITION_FIELDS= +export TABLE_PROPERTIES= + +gcloud dataflow flex-template run "deltalake-to-iceberg-yaml-job" \ + --project "$PROJECT" \ + --region "$REGION" \ + --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ + --parameters "deltaLakeTable=$DELTA_LAKE_TABLE" \ + --parameters "deltaLakeHadoopConfig=$DELTA_LAKE_HADOOP_CONFIG" \ + --parameters "table=$TABLE" \ + --parameters "catalogName=$CATALOG_NAME" \ + --parameters "catalogProperties=$CATALOG_PROPERTIES" \ + --parameters "configProperties=$CONFIG_PROPERTIES" \ + --parameters "drop=$DROP" \ + --parameters "filter=$FILTER" \ + --parameters "keep=$KEEP" \ + --parameters "only=$ONLY" \ + --parameters "partitionFields=$PARTITION_FIELDS" \ + --parameters "tableProperties=$TABLE_PROPERTIES" +``` + +For more information about the command, please check: +https://cloud.google.com/sdk/gcloud/reference/dataflow/flex-template/run + + +**Using the plugin**: + +Instead of just generating the template in the folder, it is possible to stage +and run the template in a single command. This may be useful for testing when +changing the templates. + +```shell +export PROJECT= +export BUCKET_NAME= +export REGION=us-central1 + +### Required +export DELTA_LAKE_TABLE= +export TABLE=
+export CATALOG_NAME= +export CATALOG_PROPERTIES= + +### Optional +export DELTA_LAKE_HADOOP_CONFIG= +export CONFIG_PROPERTIES= +export DROP= +export FILTER= +export KEEP= +export ONLY= +export PARTITION_FIELDS= +export TABLE_PROPERTIES= + +mvn clean package -PtemplatesRun \ +-DskipTests \ +-DprojectId="$PROJECT" \ +-DbucketName="$BUCKET_NAME" \ +-Dregion="$REGION" \ +-DjobName="deltalake-to-iceberg-yaml-job" \ +-DtemplateName="DeltaLake_To_Iceberg_Yaml" \ +-Dparameters="deltaLakeTable=$DELTA_LAKE_TABLE,deltaLakeHadoopConfig=$DELTA_LAKE_HADOOP_CONFIG,table=$TABLE,catalogName=$CATALOG_NAME,catalogProperties=$CATALOG_PROPERTIES,configProperties=$CONFIG_PROPERTIES,drop=$DROP,filter=$FILTER,keep=$KEEP,only=$ONLY,partitionFields=$PARTITION_FIELDS,tableProperties=$TABLE_PROPERTIES" \ +-f yaml +``` + +## Terraform + +Dataflow supports the utilization of Terraform to manage template jobs, +see [dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job). + +Terraform modules have been generated for most templates in this repository. This includes the relevant parameters +specific to the template. If available, they may be used instead of +[dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job) +directly. + +To use the autogenerated module, execute the standard +[terraform workflow](https://developer.hashicorp.com/terraform/intro/core-workflow): + +```shell +cd yaml/terraform/DeltaLake_To_Iceberg_Yaml +terraform init +terraform apply +``` + +To use +[dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job) +directly: + +```terraform +provider "google-beta" { + project = var.project +} +variable "project" { + default = "" +} +variable "region" { + default = "us-central1" +} + +resource "google_dataflow_flex_template_job" "deltalake_to_iceberg_yaml" { + + provider = google-beta + container_spec_gcs_path = "gs://dataflow-templates-${var.region}/latest/flex/DeltaLake_To_Iceberg_Yaml" + name = "deltalake-to-iceberg-yaml" + region = var.region + parameters = { + deltaLakeTable = "" + table = "
" + catalogName = "" + catalogProperties = "" + # deltaLakeHadoopConfig = "" + # configProperties = "" + # drop = "" + # filter = "" + # keep = "" + # only = "" + # partitionFields = "" + # tableProperties = "" + } +} +``` From 353123c097253f85d28ff1e0a2afe1a8603c0a0c Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 18 Aug 2026 16:36:55 +0000 Subject: [PATCH 07/26] switch back to stop method for direct runner --- .../org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java index 8041dc7a80..efce55e16a 100644 --- a/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java +++ b/it/google-cloud-platform/src/main/java/org/apache/beam/it/gcp/dataflow/DirectRunnerClient.java @@ -265,7 +265,7 @@ public void cancel() { currentJob.setCurrentState(JobState.CANCELLED.toString()); try { - this.interrupt(); + this.stop(); } catch (Exception e) { LOG.warn("Error cancelling job", e); } From 0b6fa4a80e400e5568a37850ddaea130a5b08d43 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 18 Aug 2026 18:20:33 +0000 Subject: [PATCH 08/26] address gemini comment --- .../cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java index d7e8e80e2e..8d9e1502fe 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java @@ -104,7 +104,7 @@ public void testDeltaLakeToIceberg() throws IOException { + "{\\\"name\\\":\\\"state\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + "{\\\"name\\\":\\\"price\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + "]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":123456789}}\n" - + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":774," + + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" + parquetBytes.length + "," + "\"modificationTime\":123456789,\"dataChange\":true}}"; gcsClient.createArtifact( From a9b0bcf532820767ce62449586561945589e7320 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 18 Aug 2026 18:31:32 +0000 Subject: [PATCH 09/26] spotless --- .../teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java index 8d9e1502fe..499bc3b93d 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java @@ -104,7 +104,9 @@ public void testDeltaLakeToIceberg() throws IOException { + "{\\\"name\\\":\\\"state\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," + "{\\\"name\\\":\\\"price\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" + "]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":123456789}}\n" - + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" + parquetBytes.length + "," + + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" + + parquetBytes.length + + "," + "\"modificationTime\":123456789,\"dataChange\":true}}"; gcsClient.createArtifact( From 9fdced5dbbc817e3c2c4dde2cfec895f4a0356c6 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 25 Aug 2026 14:25:18 +0000 Subject: [PATCH 10/26] initial comments fixes --- yaml/README_DeltaLake_To_Iceberg_Yaml.md | 110 +++++++++--------- .../yaml/DeltaLakeToIcebergYaml.java | 91 ++++++++------- .../python/generate_yaml_java_templates.py | 3 +- .../python/options/deltalake_options.yaml | 1 + yaml/src/main/yaml/DeltaLakeToIceberg.yaml | 38 +++--- .../yaml/DeltaLakeToIcebergYamlIT.java | 6 +- 6 files changed, 129 insertions(+), 120 deletions(-) diff --git a/yaml/README_DeltaLake_To_Iceberg_Yaml.md b/yaml/README_DeltaLake_To_Iceberg_Yaml.md index 6cae1b0840..1715d73ba6 100644 --- a/yaml/README_DeltaLake_To_Iceberg_Yaml.md +++ b/yaml/README_DeltaLake_To_Iceberg_Yaml.md @@ -15,20 +15,20 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Required parameters * **deltaLakeTable**: The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table. For example, `gs://your-bucket/path/to/table`. -* **table**: A fully-qualified table identifier, e.g., my_dataset.my_table. For example, `my_dataset.my_table`. -* **catalogName**: The name of the Iceberg catalog that contains the table. For example, `my_hadoop_catalog`. -* **catalogProperties**: A map of properties for setting up the Iceberg catalog. For example, `{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}`. +* **lakehouseTable**: A fully-qualified table identifier, e.g., my_dataset.my_table. For example, `my_dataset.my_table`. +* **lakehouseCatalogName**: The name of the Iceberg catalog that contains the table. For example, `my_hadoop_catalog`. +* **lakehouseCatalogProperties**: A map of properties for setting up the Iceberg catalog. For example, `{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}`. ### Optional parameters -* **deltaLakeHadoopConfig**: A map of properties to pass to Hadoop Configuration, e.g. key-value pairs. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. -* **configProperties**: A map of properties to pass to the Hadoop Configuration. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. -* **drop**: A list of field names to drop. Mutually exclusive with 'keep' and 'only'. For example, `["field_to_drop_1", "field_to_drop_2"]`. -* **filter**: A filter expression to apply to records from the Iceberg table. For example, `age > 18`. -* **keep**: A list of field names to keep. Mutually exclusive with 'drop' and 'only'. For example, `["field_to_keep_1", "field_to_keep_2"]`. -* **only**: The name of a single field to write. Mutually exclusive with 'keep' and 'drop'. For example, `my_record_field`. -* **partitionFields**: A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category']. For example, `["day(ts)", "bucket(id, 4)"]`. -* **tableProperties**: A map of Iceberg table properties to set when the table is created. For example, `{"commit.retry.num-retries": "2"}`. +* **deltaLakeHadoopConfig**: A map of properties to pass to Hadoop Configuration, e.g. key-value pairs. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. Defaults to: {"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", "fs.AbstractFileSystem.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS", "fs.gs.auth.type": "APPLICATION_DEFAULT", "fs.gs.project.id": ""}. +* **lakehouseConfigProperties**: A map of properties to pass to the Hadoop Configuration. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. +* **lakehouseDrop**: A list of field names to drop. Mutually exclusive with 'keep' and 'only'. For example, `["field_to_drop_1", "field_to_drop_2"]`. +* **lakehouseFilter**: A filter expression to apply to records from the Iceberg table. For example, `age > 18`. +* **lakehouseKeep**: A list of field names to keep. Mutually exclusive with 'drop' and 'only'. For example, `["field_to_keep_1", "field_to_keep_2"]`. +* **lakehouseOnly**: The name of a single field to write. Mutually exclusive with 'keep' and 'drop'. For example, `my_record_field`. +* **lakehousePartitionFields**: A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category']. For example, `["day(ts)", "bucket(id, 4)"]`. +* **lakehouseTableProperties**: A map of Iceberg table properties to set when the table is created. For example, `{"commit.retry.num-retries": "2"}`. @@ -122,19 +122,19 @@ export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/DeltaLake_To_Iceb ### Required export DELTA_LAKE_TABLE= -export TABLE=
-export CATALOG_NAME= -export CATALOG_PROPERTIES= +export LAKEHOUSE_TABLE= +export LAKEHOUSE_CATALOG_NAME= +export LAKEHOUSE_CATALOG_PROPERTIES= ### Optional -export DELTA_LAKE_HADOOP_CONFIG= -export CONFIG_PROPERTIES= -export DROP= -export FILTER= -export KEEP= -export ONLY= -export PARTITION_FIELDS= -export TABLE_PROPERTIES= +export DELTA_LAKE_HADOOP_CONFIG="{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", "fs.AbstractFileSystem.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS", "fs.gs.auth.type": "APPLICATION_DEFAULT", "fs.gs.project.id": ""}" +export LAKEHOUSE_CONFIG_PROPERTIES= +export LAKEHOUSE_DROP= +export LAKEHOUSE_FILTER= +export LAKEHOUSE_KEEP= +export LAKEHOUSE_ONLY= +export LAKEHOUSE_PARTITION_FIELDS= +export LAKEHOUSE_TABLE_PROPERTIES= gcloud dataflow flex-template run "deltalake-to-iceberg-yaml-job" \ --project "$PROJECT" \ @@ -142,16 +142,16 @@ gcloud dataflow flex-template run "deltalake-to-iceberg-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "deltaLakeTable=$DELTA_LAKE_TABLE" \ --parameters "deltaLakeHadoopConfig=$DELTA_LAKE_HADOOP_CONFIG" \ - --parameters "table=$TABLE" \ - --parameters "catalogName=$CATALOG_NAME" \ - --parameters "catalogProperties=$CATALOG_PROPERTIES" \ - --parameters "configProperties=$CONFIG_PROPERTIES" \ - --parameters "drop=$DROP" \ - --parameters "filter=$FILTER" \ - --parameters "keep=$KEEP" \ - --parameters "only=$ONLY" \ - --parameters "partitionFields=$PARTITION_FIELDS" \ - --parameters "tableProperties=$TABLE_PROPERTIES" + --parameters "lakehouseTable=$LAKEHOUSE_TABLE" \ + --parameters "lakehouseCatalogName=$LAKEHOUSE_CATALOG_NAME" \ + --parameters "lakehouseCatalogProperties=$LAKEHOUSE_CATALOG_PROPERTIES" \ + --parameters "lakehouseConfigProperties=$LAKEHOUSE_CONFIG_PROPERTIES" \ + --parameters "lakehouseDrop=$LAKEHOUSE_DROP" \ + --parameters "lakehouseFilter=$LAKEHOUSE_FILTER" \ + --parameters "lakehouseKeep=$LAKEHOUSE_KEEP" \ + --parameters "lakehouseOnly=$LAKEHOUSE_ONLY" \ + --parameters "lakehousePartitionFields=$LAKEHOUSE_PARTITION_FIELDS" \ + --parameters "lakehouseTableProperties=$LAKEHOUSE_TABLE_PROPERTIES" ``` For more information about the command, please check: @@ -171,19 +171,19 @@ export REGION=us-central1 ### Required export DELTA_LAKE_TABLE= -export TABLE=
-export CATALOG_NAME= -export CATALOG_PROPERTIES= +export LAKEHOUSE_TABLE= +export LAKEHOUSE_CATALOG_NAME= +export LAKEHOUSE_CATALOG_PROPERTIES= ### Optional -export DELTA_LAKE_HADOOP_CONFIG= -export CONFIG_PROPERTIES= -export DROP= -export FILTER= -export KEEP= -export ONLY= -export PARTITION_FIELDS= -export TABLE_PROPERTIES= +export DELTA_LAKE_HADOOP_CONFIG="{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", "fs.AbstractFileSystem.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS", "fs.gs.auth.type": "APPLICATION_DEFAULT", "fs.gs.project.id": ""}" +export LAKEHOUSE_CONFIG_PROPERTIES= +export LAKEHOUSE_DROP= +export LAKEHOUSE_FILTER= +export LAKEHOUSE_KEEP= +export LAKEHOUSE_ONLY= +export LAKEHOUSE_PARTITION_FIELDS= +export LAKEHOUSE_TABLE_PROPERTIES= mvn clean package -PtemplatesRun \ -DskipTests \ @@ -192,7 +192,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="deltalake-to-iceberg-yaml-job" \ -DtemplateName="DeltaLake_To_Iceberg_Yaml" \ --Dparameters="deltaLakeTable=$DELTA_LAKE_TABLE,deltaLakeHadoopConfig=$DELTA_LAKE_HADOOP_CONFIG,table=$TABLE,catalogName=$CATALOG_NAME,catalogProperties=$CATALOG_PROPERTIES,configProperties=$CONFIG_PROPERTIES,drop=$DROP,filter=$FILTER,keep=$KEEP,only=$ONLY,partitionFields=$PARTITION_FIELDS,tableProperties=$TABLE_PROPERTIES" \ +-Dparameters="deltaLakeTable=$DELTA_LAKE_TABLE,deltaLakeHadoopConfig=$DELTA_LAKE_HADOOP_CONFIG,lakehouseTable=$LAKEHOUSE_TABLE,lakehouseCatalogName=$LAKEHOUSE_CATALOG_NAME,lakehouseCatalogProperties=$LAKEHOUSE_CATALOG_PROPERTIES,lakehouseConfigProperties=$LAKEHOUSE_CONFIG_PROPERTIES,lakehouseDrop=$LAKEHOUSE_DROP,lakehouseFilter=$LAKEHOUSE_FILTER,lakehouseKeep=$LAKEHOUSE_KEEP,lakehouseOnly=$LAKEHOUSE_ONLY,lakehousePartitionFields=$LAKEHOUSE_PARTITION_FIELDS,lakehouseTableProperties=$LAKEHOUSE_TABLE_PROPERTIES" \ -f yaml ``` @@ -238,17 +238,17 @@ resource "google_dataflow_flex_template_job" "deltalake_to_iceberg_yaml" { region = var.region parameters = { deltaLakeTable = "" - table = "
" - catalogName = "" - catalogProperties = "" - # deltaLakeHadoopConfig = "" - # configProperties = "" - # drop = "" - # filter = "" - # keep = "" - # only = "" - # partitionFields = "" - # tableProperties = "" + lakehouseTable = "" + lakehouseCatalogName = "" + lakehouseCatalogProperties = "" + # deltaLakeHadoopConfig = ""{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", "fs.AbstractFileSystem.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS", "fs.gs.auth.type": "APPLICATION_DEFAULT", "fs.gs.project.id": }"" + # lakehouseConfigProperties = "" + # lakehouseDrop = "" + # lakehouseFilter = "" + # lakehouseKeep = "" + # lakehouseOnly = "" + # lakehousePartitionFields = "" + # lakehouseTableProperties = "" } } ``` diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java index 6052128284..6d0e70f302 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java @@ -18,6 +18,7 @@ import com.google.cloud.teleport.metadata.Template; import com.google.cloud.teleport.metadata.TemplateCategory; import com.google.cloud.teleport.metadata.TemplateParameter; +import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.Validation; @Template( @@ -25,20 +26,13 @@ category = TemplateCategory.BATCH, type = Template.TemplateType.YAML, displayName = "Delta Lake to Lakehouse", - description = - "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", + description = "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", flexContainerName = "pipeline-yaml", yamlTemplateFile = "DeltaLakeToIceberg.yaml", - filesToCopy = { - "main.py", - "requirements.txt", - "options/deltalake_options.yaml", - "options/iceberg_options.yaml" - }, + filesToCopy = {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/iceberg_options.yaml"}, documentation = "", contactInformation = "https://cloud.google.com/support", - requirements = { - "The Input Delta Lake table must exist and be accessible.", + requirements = {"The Input Delta Lake table must exist and be accessible.", "The Output Iceberg table must exist or be created, and the warehouse must be accessible." }, streaming = false, @@ -51,7 +45,8 @@ public interface DeltaLakeToIcebergYaml { optional = false, description = "A GCS path to the Delta Lake table.", helpText = "The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table.", - example = "gs://your-bucket/path/to/table") + example = "gs://your-bucket/path/to/table" + ) @Validation.Required String getDeltaLakeTable(); @@ -61,99 +56,111 @@ public interface DeltaLakeToIcebergYaml { optional = true, description = "Properties passed to Hadoop Configuration.", helpText = "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs.", - example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}") + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" + ) + @Default.String("{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\", \"fs.AbstractFileSystem.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS\", \"fs.gs.auth.type\": \"APPLICATION_DEFAULT\", \"fs.gs.project.id\": \"\"}") String getDeltaLakeHadoopConfig(); @TemplateParameter.Text( order = 3, - name = "table", + name = "lakehouseTable", optional = false, description = "A fully-qualified table identifier.", helpText = "A fully-qualified table identifier, e.g., my_dataset.my_table.", - example = "my_dataset.my_table") + example = "my_dataset.my_table" + ) @Validation.Required - String getTable(); + String getLakehouseTable(); @TemplateParameter.Text( order = 4, - name = "catalogName", + name = "lakehouseCatalogName", optional = false, description = "Name of the catalog containing the table.", helpText = "The name of the Iceberg catalog that contains the table.", - example = "my_hadoop_catalog") + example = "my_hadoop_catalog" + ) @Validation.Required - String getCatalogName(); + String getLakehouseCatalogName(); @TemplateParameter.Text( order = 5, - name = "catalogProperties", + name = "lakehouseCatalogProperties", optional = false, description = "Properties used to set up the Iceberg catalog.", helpText = "A map of properties for setting up the Iceberg catalog.", - example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}") + example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}" + ) @Validation.Required - String getCatalogProperties(); + String getLakehouseCatalogProperties(); @TemplateParameter.Text( order = 6, - name = "configProperties", + name = "lakehouseConfigProperties", optional = true, description = "Properties passed to the Hadoop Configuration.", helpText = "A map of properties to pass to the Hadoop Configuration.", - example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}") - String getConfigProperties(); + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" + ) + String getLakehouseConfigProperties(); @TemplateParameter.Text( order = 7, - name = "drop", + name = "lakehouseDrop", optional = true, description = "A list of field names to drop from the input record before writing.", helpText = "A list of field names to drop. Mutually exclusive with 'keep' and 'only'.", - example = "[\"field_to_drop_1\", \"field_to_drop_2\"]") - String getDrop(); + example = "[\"field_to_drop_1\", \"field_to_drop_2\"]" + ) + String getLakehouseDrop(); @TemplateParameter.Text( order = 8, - name = "filter", + name = "lakehouseFilter", optional = true, description = "An optional filter expression to apply to the input records.", helpText = "A filter expression to apply to records from the Iceberg table.", - example = "age > 18") - String getFilter(); + example = "age > 18" + ) + String getLakehouseFilter(); @TemplateParameter.Text( order = 9, - name = "keep", + name = "lakehouseKeep", optional = true, description = "A list of field names to keep in the input record.", helpText = "A list of field names to keep. Mutually exclusive with 'drop' and 'only'.", - example = "[\"field_to_keep_1\", \"field_to_keep_2\"]") - String getKeep(); + example = "[\"field_to_keep_1\", \"field_to_keep_2\"]" + ) + String getLakehouseKeep(); @TemplateParameter.Text( order = 10, - name = "only", + name = "lakehouseOnly", optional = true, description = "The name of a single record field that should be written.", helpText = "The name of a single field to write. Mutually exclusive with 'keep' and 'drop'.", - example = "my_record_field") - String getOnly(); + example = "my_record_field" + ) + String getLakehouseOnly(); @TemplateParameter.Text( order = 11, - name = "partitionFields", + name = "lakehousePartitionFields", optional = true, description = "Fields used to create a partition spec for new tables.", helpText = "A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category'].", - example = "[\"day(ts)\", \"bucket(id, 4)\"]") - String getPartitionFields(); + example = "[\"day(ts)\", \"bucket(id, 4)\"]" + ) + String getLakehousePartitionFields(); @TemplateParameter.Text( order = 12, - name = "tableProperties", + name = "lakehouseTableProperties", optional = true, description = "Iceberg table properties to be set on table creation.", helpText = "A map of Iceberg table properties to set when the table is created.", - example = "{\"commit.retry.num-retries\": \"2\"}") - String getTableProperties(); + example = "{\"commit.retry.num-retries\": \"2\"}" + ) + String getLakehouseTableProperties(); } diff --git a/yaml/src/main/python/generate_yaml_java_templates.py b/yaml/src/main/python/generate_yaml_java_templates.py index 0e8b23b4fc..12d1c328cb 100644 --- a/yaml/src/main/python/generate_yaml_java_templates.py +++ b/yaml/src/main/python/generate_yaml_java_templates.py @@ -163,7 +163,8 @@ def generate_java_interface(yaml_path, java_path): if 'default' in param: has_defaults = True if java_type == 'String': - param_code += f' @Default.String("{param["default"]}")\n' + escaped_default = str(param["default"]).replace('"', '\\"') + param_code += f' @Default.String("{escaped_default}")\n' else: param_code += f" @Default.{java_type}({param['default']})\n" diff --git a/yaml/src/main/python/options/deltalake_options.yaml b/yaml/src/main/python/options/deltalake_options.yaml index 26940eb9b6..4099a4f0f7 100644 --- a/yaml/src/main/python/options/deltalake_options.yaml +++ b/yaml/src/main/python/options/deltalake_options.yaml @@ -14,5 +14,6 @@ options: help: "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs." example: '{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}' required: false + default: '{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", "fs.AbstractFileSystem.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS", "fs.gs.auth.type": "APPLICATION_DEFAULT", "fs.gs.project.id": ""}' type: map diff --git a/yaml/src/main/yaml/DeltaLakeToIceberg.yaml b/yaml/src/main/yaml/DeltaLakeToIceberg.yaml index c4651dd2f2..46b53fb0d4 100644 --- a/yaml/src/main/yaml/DeltaLakeToIceberg.yaml +++ b/yaml/src/main/yaml/DeltaLakeToIceberg.yaml @@ -20,12 +20,12 @@ template: options_file: - "deltalake_options" - - "iceberg_options" + - "lakehouse_options" parameters: - deltalake_read_options - - iceberg_common_options - - iceberg_write_options + - lakehouse_common_options + - lakehouse_write_options pipeline: type: chain @@ -39,28 +39,28 @@ pipeline: {% endif %} - type: WriteToIceberg - name: WriteToIceberg + name: WriteToLakehouse config: - table: "{{ table }}" - catalog_name: "{{ catalogName }}" - catalog_properties: {{ catalogProperties }} - {% if configProperties %} - config_properties: {{ configProperties }} + table: "{{ lakehouseTable }}" + catalog_name: "{{ lakehouseCatalogName }}" + catalog_properties: {{ lakehouseCatalogProperties }} + {% if lakehouseConfigProperties %} + config_properties: {{ lakehouseConfigProperties }} {% endif %} - {% if drop %} - drop: {{ drop }} + {% if lakehouseDrop %} + drop: {{ lakehouseDrop }} {% endif %} - {% if keep %} - keep: {{ keep }} + {% if lakehouseKeep %} + keep: {{ lakehouseKeep }} {% endif %} - {% if only %} - only: {{ only }} + {% if lakehouseOnly %} + only: {{ lakehouseOnly }} {% endif %} - {% if partitionFields %} - partition_fields: {{ partitionFields }} + {% if lakehousePartitionFields %} + partition_fields: {{ lakehousePartitionFields }} {% endif %} - {% if tableProperties %} - table_properties: {{ tableProperties }} + {% if lakehouseTableProperties %} + table_properties: {{ lakehouseTableProperties }} {% endif %} providers: diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java index 499bc3b93d..f749888248 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java @@ -129,10 +129,10 @@ public void testDeltaLakeToIceberg() throws IOException { .addParameter("deltaLakeTable", deltaTableGcsPath) .addParameter( "deltaLakeHadoopConfig", new org.json.JSONObject(getGcsHadoopConfig()).toString()) - .addParameter("table", icebergTableIdentifier) - .addParameter("catalogName", CATALOG_NAME) + .addParameter("lakehouseTable", icebergTableIdentifier) + .addParameter("lakehouseCatalogName", CATALOG_NAME) .addParameter( - "catalogProperties", new org.json.JSONObject(getCatalogProperties()).toString()); + "lakehouseCatalogProperties", new org.json.JSONObject(getCatalogProperties()).toString()); LaunchInfo info = launchTemplate(options); assertThatPipeline(info).isRunning(); From ee83e6e6f3b091ed91ccb7d08d7f985f8004964f Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 25 Aug 2026 14:27:18 +0000 Subject: [PATCH 11/26] change iceberg to lakehouse --- ...e_To_Iceberg_Yaml.md => README_DeltaLake_To_Lakehouse_Yaml.md} | 0 ...{DeltaLakeToIcebergYaml.java => DeltaLakeToLakehouseYaml.java} | 0 .../yaml/{DeltaLakeToIceberg.yaml => DeltaLakeToLakehouse.yaml} | 0 3 files changed, 0 insertions(+), 0 deletions(-) rename yaml/{README_DeltaLake_To_Iceberg_Yaml.md => README_DeltaLake_To_Lakehouse_Yaml.md} (100%) rename yaml/src/main/java/com/google/cloud/teleport/templates/yaml/{DeltaLakeToIcebergYaml.java => DeltaLakeToLakehouseYaml.java} (100%) rename yaml/src/main/yaml/{DeltaLakeToIceberg.yaml => DeltaLakeToLakehouse.yaml} (100%) diff --git a/yaml/README_DeltaLake_To_Iceberg_Yaml.md b/yaml/README_DeltaLake_To_Lakehouse_Yaml.md similarity index 100% rename from yaml/README_DeltaLake_To_Iceberg_Yaml.md rename to yaml/README_DeltaLake_To_Lakehouse_Yaml.md diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java similarity index 100% rename from yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java rename to yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java diff --git a/yaml/src/main/yaml/DeltaLakeToIceberg.yaml b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml similarity index 100% rename from yaml/src/main/yaml/DeltaLakeToIceberg.yaml rename to yaml/src/main/yaml/DeltaLakeToLakehouse.yaml From 99d9dd53a4edfc072ddb5a561fcd4e3dd1c88610 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 25 Aug 2026 14:57:12 +0000 Subject: [PATCH 12/26] change to a wrapped writetoiceberg via writetolakehouse --- .../provider_listing_template.yaml | 3 +- yaml/README_DeltaLake_To_Lakehouse_Yaml.md | 34 ++++----- .../yaml/DeltaLakeToLakehouseYaml.java | 22 +++--- .../python/options/lakehouse_options.yaml | 76 +++++++++++++++++++ yaml/src/main/yaml/DeltaLakeToLakehouse.yaml | 13 ++-- ...T.java => DeltaLakeToLakehouseYamlIT.java} | 22 +++--- 6 files changed, 124 insertions(+), 46 deletions(-) create mode 100644 yaml/src/main/python/options/lakehouse_options.yaml rename yaml/src/test/java/com/google/cloud/teleport/templates/yaml/{DeltaLakeToIcebergYamlIT.java => DeltaLakeToLakehouseYamlIT.java} (89%) diff --git a/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml b/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml index 4f8042d3da..16cc5244e0 100644 --- a/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml +++ b/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml @@ -4,4 +4,5 @@ - https://storage.googleapis.com/ transforms: CopyFilesToGCS: "copy_files_to_gcs.CopyFilesToGCS" - ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" \ No newline at end of file + ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" + WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse" \ No newline at end of file diff --git a/yaml/README_DeltaLake_To_Lakehouse_Yaml.md b/yaml/README_DeltaLake_To_Lakehouse_Yaml.md index 1715d73ba6..a9519793ad 100644 --- a/yaml/README_DeltaLake_To_Lakehouse_Yaml.md +++ b/yaml/README_DeltaLake_To_Lakehouse_Yaml.md @@ -1,8 +1,8 @@ Delta Lake to Lakehouse template --- -The Delta Lake to Iceberg template is a batch pipeline that reads data from a -Delta Lake table and outputs the records to an Apache Iceberg table. +The Delta Lake to Lakehouse template is a batch pipeline that reads data from a +Delta Lake table and outputs the records to an Lakehouse table. @@ -16,19 +16,19 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat * **deltaLakeTable**: The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table. For example, `gs://your-bucket/path/to/table`. * **lakehouseTable**: A fully-qualified table identifier, e.g., my_dataset.my_table. For example, `my_dataset.my_table`. -* **lakehouseCatalogName**: The name of the Iceberg catalog that contains the table. For example, `my_hadoop_catalog`. -* **lakehouseCatalogProperties**: A map of properties for setting up the Iceberg catalog. For example, `{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}`. +* **lakehouseCatalogName**: The name of the Lakehouse catalog that contains the table. For example, `my_hadoop_catalog`. +* **lakehouseCatalogProperties**: A map of properties for setting up the Lakehouse catalog. For example, `{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}`. ### Optional parameters * **deltaLakeHadoopConfig**: A map of properties to pass to Hadoop Configuration, e.g. key-value pairs. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. Defaults to: {"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem", "fs.AbstractFileSystem.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS", "fs.gs.auth.type": "APPLICATION_DEFAULT", "fs.gs.project.id": ""}. * **lakehouseConfigProperties**: A map of properties to pass to the Hadoop Configuration. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. * **lakehouseDrop**: A list of field names to drop. Mutually exclusive with 'keep' and 'only'. For example, `["field_to_drop_1", "field_to_drop_2"]`. -* **lakehouseFilter**: A filter expression to apply to records from the Iceberg table. For example, `age > 18`. +* **lakehouseFilter**: A filter expression to apply to records from the Lakehouse table. For example, `age > 18`. * **lakehouseKeep**: A list of field names to keep. Mutually exclusive with 'drop' and 'only'. For example, `["field_to_keep_1", "field_to_keep_2"]`. * **lakehouseOnly**: The name of a single field to write. Mutually exclusive with 'keep' and 'drop'. For example, `my_record_field`. * **lakehousePartitionFields**: A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category']. For example, `["day(ts)", "bucket(id, 4)"]`. -* **lakehouseTableProperties**: A map of Iceberg table properties to set when the table is created. For example, `{"commit.retry.num-retries": "2"}`. +* **lakehouseTableProperties**: A map of Lakehouse table properties to set when the table is created. For example, `{"commit.retry.num-retries": "2"}`. @@ -45,7 +45,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat :star2: Those dependencies are pre-installed if you use Google Cloud Shell! -[![Open in Cloud Shell](http://gstatic.com/cloudssh/images/open-btn.svg)](https://console.cloud.google.com/cloudshell/editor?cloudshell_git_repo=https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2FDataflowTemplates.git&cloudshell_open_in_editor=yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYaml.java) +[![Open in Cloud Shell](http://gstatic.com/cloudssh/images/open-btn.svg)](https://console.cloud.google.com/cloudshell/editor?cloudshell_git_repo=https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2FDataflowTemplates.git&cloudshell_open_in_editor=yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java) ### Templates Plugin @@ -87,7 +87,7 @@ mvn clean package -PtemplatesStage \ -DbucketName="$BUCKET_NAME" \ -DartifactRegistry="$ARTIFACT_REGISTRY_REPO" \ -DstagePrefix="templates" \ --DtemplateName="DeltaLake_To_Iceberg_Yaml" \ +-DtemplateName="DeltaLake_To_Lakehouse_Yaml" \ -f yaml ``` @@ -98,7 +98,7 @@ The command should build and save the template to Google Cloud, and then print the complete location on Cloud Storage: ``` -Flex Template was staged! gs:///templates/flex/DeltaLake_To_Iceberg_Yaml +Flex Template was staged! gs:///templates/flex/DeltaLake_To_Lakehouse_Yaml ``` The specific path should be copied as it will be used in the following steps. @@ -118,7 +118,7 @@ Provided that, the following command line can be used: export PROJECT= export BUCKET_NAME= export REGION=us-central1 -export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/DeltaLake_To_Iceberg_Yaml" +export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/DeltaLake_To_Lakehouse_Yaml" ### Required export DELTA_LAKE_TABLE= @@ -136,7 +136,7 @@ export LAKEHOUSE_ONLY= export LAKEHOUSE_PARTITION_FIELDS= export LAKEHOUSE_TABLE_PROPERTIES= -gcloud dataflow flex-template run "deltalake-to-iceberg-yaml-job" \ +gcloud dataflow flex-template run "deltalake-to-lakehouse-yaml-job" \ --project "$PROJECT" \ --region "$REGION" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ @@ -190,8 +190,8 @@ mvn clean package -PtemplatesRun \ -DprojectId="$PROJECT" \ -DbucketName="$BUCKET_NAME" \ -Dregion="$REGION" \ --DjobName="deltalake-to-iceberg-yaml-job" \ --DtemplateName="DeltaLake_To_Iceberg_Yaml" \ +-DjobName="deltalake-to-lakehouse-yaml-job" \ +-DtemplateName="DeltaLake_To_Lakehouse_Yaml" \ -Dparameters="deltaLakeTable=$DELTA_LAKE_TABLE,deltaLakeHadoopConfig=$DELTA_LAKE_HADOOP_CONFIG,lakehouseTable=$LAKEHOUSE_TABLE,lakehouseCatalogName=$LAKEHOUSE_CATALOG_NAME,lakehouseCatalogProperties=$LAKEHOUSE_CATALOG_PROPERTIES,lakehouseConfigProperties=$LAKEHOUSE_CONFIG_PROPERTIES,lakehouseDrop=$LAKEHOUSE_DROP,lakehouseFilter=$LAKEHOUSE_FILTER,lakehouseKeep=$LAKEHOUSE_KEEP,lakehouseOnly=$LAKEHOUSE_ONLY,lakehousePartitionFields=$LAKEHOUSE_PARTITION_FIELDS,lakehouseTableProperties=$LAKEHOUSE_TABLE_PROPERTIES" \ -f yaml ``` @@ -210,7 +210,7 @@ To use the autogenerated module, execute the standard [terraform workflow](https://developer.hashicorp.com/terraform/intro/core-workflow): ```shell -cd yaml/terraform/DeltaLake_To_Iceberg_Yaml +cd yaml/terraform/DeltaLake_To_Lakehouse_Yaml terraform init terraform apply ``` @@ -230,11 +230,11 @@ variable "region" { default = "us-central1" } -resource "google_dataflow_flex_template_job" "deltalake_to_iceberg_yaml" { +resource "google_dataflow_flex_template_job" "deltalake_to_lakehouse_yaml" { provider = google-beta - container_spec_gcs_path = "gs://dataflow-templates-${var.region}/latest/flex/DeltaLake_To_Iceberg_Yaml" - name = "deltalake-to-iceberg-yaml" + container_spec_gcs_path = "gs://dataflow-templates-${var.region}/latest/flex/DeltaLake_To_Lakehouse_Yaml" + name = "deltalake-to-lakehouse-yaml" region = var.region parameters = { deltaLakeTable = "" diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java index 6d0e70f302..0b565efa32 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java @@ -22,14 +22,14 @@ import org.apache.beam.sdk.options.Validation; @Template( - name = "DeltaLake_To_Iceberg_Yaml", + name = "DeltaLake_To_Lakehouse_Yaml", category = TemplateCategory.BATCH, type = Template.TemplateType.YAML, displayName = "Delta Lake to Lakehouse", - description = "The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Apache Iceberg table.", + description = "The Delta Lake to Lakehouse template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Lakehouse table.", flexContainerName = "pipeline-yaml", - yamlTemplateFile = "DeltaLakeToIceberg.yaml", - filesToCopy = {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/iceberg_options.yaml"}, + yamlTemplateFile = "DeltaLakeToLakehouse.yaml", + filesToCopy = {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/lakehouse_options.yaml"}, documentation = "", contactInformation = "https://cloud.google.com/support", requirements = {"The Input Delta Lake table must exist and be accessible.", @@ -37,7 +37,7 @@ }, streaming = false, hidden = false) -public interface DeltaLakeToIcebergYaml { +public interface DeltaLakeToLakehouseYaml { @TemplateParameter.Text( order = 1, @@ -77,7 +77,7 @@ public interface DeltaLakeToIcebergYaml { name = "lakehouseCatalogName", optional = false, description = "Name of the catalog containing the table.", - helpText = "The name of the Iceberg catalog that contains the table.", + helpText = "The name of the Lakehouse catalog that contains the table.", example = "my_hadoop_catalog" ) @Validation.Required @@ -87,8 +87,8 @@ public interface DeltaLakeToIcebergYaml { order = 5, name = "lakehouseCatalogProperties", optional = false, - description = "Properties used to set up the Iceberg catalog.", - helpText = "A map of properties for setting up the Iceberg catalog.", + description = "Properties used to set up the Lakehouse catalog.", + helpText = "A map of properties for setting up the Lakehouse catalog.", example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}" ) @Validation.Required @@ -119,7 +119,7 @@ public interface DeltaLakeToIcebergYaml { name = "lakehouseFilter", optional = true, description = "An optional filter expression to apply to the input records.", - helpText = "A filter expression to apply to records from the Iceberg table.", + helpText = "A filter expression to apply to records from the Lakehouse table.", example = "age > 18" ) String getLakehouseFilter(); @@ -158,8 +158,8 @@ public interface DeltaLakeToIcebergYaml { order = 12, name = "lakehouseTableProperties", optional = true, - description = "Iceberg table properties to be set on table creation.", - helpText = "A map of Iceberg table properties to set when the table is created.", + description = "Lakehouse table properties to be set on table creation.", + helpText = "A map of Lakehouse table properties to set when the table is created.", example = "{\"commit.retry.num-retries\": \"2\"}" ) String getLakehouseTableProperties(); diff --git a/yaml/src/main/python/options/lakehouse_options.yaml b/yaml/src/main/python/options/lakehouse_options.yaml new file mode 100644 index 0000000000..8eba3e2f81 --- /dev/null +++ b/yaml/src/main/python/options/lakehouse_options.yaml @@ -0,0 +1,76 @@ +options: + - name: "lakehouse_common_options" + parameters: + - order: 1 + name: "lakehouseTable" + description: "A fully-qualified table identifier." + help: "A fully-qualified table identifier, e.g., my_dataset.my_table." + example: "my_dataset.my_table" + required: true + type: text + - order: 2 + name: "lakehouseCatalogName" + description: "Name of the catalog containing the table." + help: "The name of the Lakehouse catalog that contains the table." + example: "my_hadoop_catalog" + required: true + type: text + - order: 3 + name: "lakehouseCatalogProperties" + description: "Properties used to set up the Lakehouse catalog." + help: "A map of properties for setting up the Lakehouse catalog." + example: '{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}' + required: true + type: text + - order: 4 + name: "lakehouseConfigProperties" + description: "Properties passed to the Hadoop Configuration." + help: "A map of properties to pass to the Hadoop Configuration." + example: '{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}' + required: false + type: map + - order: 5 + name: "lakehouseDrop" + description: "A list of field names to drop from the input record before writing." + help: "A list of field names to drop. Mutually exclusive with 'keep' and 'only'." + example: '["field_to_drop_1", "field_to_drop_2"]' + required: false + type: text + - order: 6 + name: "lakehouseFilter" + description: "An optional filter expression to apply to the input records." + help: "A filter expression to apply to records from the Lakehouse table." + example: "age > 18" + required: false + type: text + - order: 7 + name: "lakehouseKeep" + description: "A list of field names to keep in the input record." + help: "A list of field names to keep. Mutually exclusive with 'drop' and 'only'." + example: '["field_to_keep_1", "field_to_keep_2"]' + required: false + type: text + + - name: "lakehouse_write_options" + parameters: + - order: 1 + name: "lakehouseOnly" + description: "The name of a single record field that should be written." + help: "The name of a single field to write. Mutually exclusive with 'keep' and 'drop'." + example: "my_record_field" + required: false + type: text + - order: 2 + name: "lakehousePartitionFields" + description: "Fields used to create a partition spec for new tables." + help: "A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category']." + example: '["day(ts)", "bucket(id, 4)"]' + required: false + type: text + - order: 3 + name: "lakehouseTableProperties" + description: "Lakehouse table properties to be set on table creation." + help: "A map of Lakehouse table properties to set when the table is created." + example: '{"commit.retry.num-retries": "2"}' + required: false + type: text diff --git a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml index 46b53fb0d4..6987d6148f 100644 --- a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml +++ b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml @@ -1,15 +1,15 @@ template: - name: "DeltaLake_To_Iceberg_Yaml" + name: "DeltaLake_To_Lakehouse_Yaml" category: "BATCH" type: "YAML" display_name: "Delta Lake to Lakehouse" description: > - The Delta Lake to Iceberg template is a batch pipeline that reads data from a Delta Lake table - and outputs the records to an Apache Iceberg table. + The Delta Lake to Lakehouse template is a batch pipeline that reads data from a Delta Lake table + and outputs the records to an Lakehouse table. flex_container_name: "pipeline-yaml" - yamlTemplateFile: "DeltaLakeToIceberg.yaml" + yamlTemplateFile: "DeltaLakeToLakehouse.yaml" filesToCopy: > - {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/iceberg_options.yaml"} + {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/lakehouse_options.yaml"} contactInformation: "https://cloud.google.com/support" requirements: { "The Input Delta Lake table must exist and be accessible.", @@ -38,7 +38,7 @@ pipeline: hadoop_config: {{ deltaLakeHadoopConfig }} {% endif %} - - type: WriteToIceberg + - type: WriteToLakehouse name: WriteToLakehouse config: table: "{{ lakehouseTable }}" @@ -70,6 +70,7 @@ providers: - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-07-20/job_builder_util_transforms-0.2.0.tar.gz transforms: ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" + WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse" options: streaming: false diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java similarity index 89% rename from yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java rename to yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java index f749888248..0c6456bd8b 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToIcebergYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java @@ -42,19 +42,19 @@ import org.junit.runner.RunWith; import org.junit.runners.JUnit4; -/** Integration test for {@link DeltaLakeToIcebergYaml} template. */ +/** Integration test for {@link DeltaLakeToLakehouseYaml} template. */ @Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) -@TemplateIntegrationTest(DeltaLakeToIcebergYaml.class) +@TemplateIntegrationTest(DeltaLakeToLakehouseYaml.class) @RunWith(JUnit4.class) -public class DeltaLakeToIcebergYamlIT extends TemplateTestBase { +public class DeltaLakeToLakehouseYamlIT extends TemplateTestBase { private IcebergResourceManager icebergResourceManager; private static final String CATALOG_NAME = "hadoop_catalog"; private final String namespace = - "deltalake_iceberg_ns_" + UUID.randomUUID().toString().replace("-", ""); - private static final String ICEBERG_TABLE_NAME = "iceberg_table"; - private final String icebergTableIdentifier = namespace + "." + ICEBERG_TABLE_NAME; + "deltalake_lakehouse_ns_" + UUID.randomUUID().toString().replace("-", ""); + private static final String LAKEHOUSE_TABLE_NAME = "lakehouse_table"; + private final String lakehouseTableIdentifier = namespace + "." + LAKEHOUSE_TABLE_NAME; @Before public void setUp() throws IOException { @@ -74,7 +74,7 @@ public void tearDown() { } @Test - public void testDeltaLakeToIceberg() throws IOException { + public void testDeltaLakeToLakehouse() throws IOException { // 1. Arrange: Create Delta Lake source table in GCS String deltaTableDir = "delta-table"; org.apache.avro.Schema avroSchema = @@ -114,14 +114,14 @@ public void testDeltaLakeToIceberg() throws IOException { String deltaTableGcsPath = getGcsPath(deltaTableDir); - // 2. Arrange: Create destination Iceberg table + // 2. Arrange: Create destination Lakehouse table icebergResourceManager.createNamespace(namespace); Schema icebergSchema = new Schema( Types.NestedField.required(1, "id", Types.StringType.get()), Types.NestedField.required(2, "state", Types.StringType.get()), Types.NestedField.required(3, "price", Types.DoubleType.get())); - icebergResourceManager.createTable(icebergTableIdentifier, icebergSchema); + icebergResourceManager.createTable(lakehouseTableIdentifier, icebergSchema); // 3. Act: Configure options and launch template LaunchConfig.Builder options = @@ -129,7 +129,7 @@ public void testDeltaLakeToIceberg() throws IOException { .addParameter("deltaLakeTable", deltaTableGcsPath) .addParameter( "deltaLakeHadoopConfig", new org.json.JSONObject(getGcsHadoopConfig()).toString()) - .addParameter("lakehouseTable", icebergTableIdentifier) + .addParameter("lakehouseTable", lakehouseTableIdentifier) .addParameter("lakehouseCatalogName", CATALOG_NAME) .addParameter( "lakehouseCatalogProperties", new org.json.JSONObject(getCatalogProperties()).toString()); @@ -142,7 +142,7 @@ public void testDeltaLakeToIceberg() throws IOException { // 4. Assert assertThatResult(result).isLaunchFinished(); - List records = icebergResourceManager.read(icebergTableIdentifier); + List records = icebergResourceManager.read(lakehouseTableIdentifier); assertEquals(1, records.size()); Record record = records.get(0); From 0dc18f3aa4e5301b49df01a56ba3220a52156d9a Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 25 Aug 2026 16:23:06 +0000 Subject: [PATCH 13/26] update yaml file with wrapper transform --- yaml/src/main/yaml/DeltaLakeToLakehouse.yaml | 29 +++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml index 6987d6148f..c51a441320 100644 --- a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml +++ b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml @@ -70,7 +70,34 @@ providers: - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-07-20/job_builder_util_transforms-0.2.0.tar.gz transforms: ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" - WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse" + + - type: yaml + transforms: + WriteToLakehouse: + body: | + type: WriteToIceberg + config: + table: "{{table}}" + catalog_name: "{{catalog_name}}" + catalog_properties: {{catalog_properties}} + {% if config_properties %} + config_properties: {{config_properties}} + {% endif %} + {% if drop %} + drop: {{drop}} + {% endif %} + {% if keep %} + keep: {{keep}} + {% endif %} + {% if only %} + only: {{only}} + {% endif %} + {% if partition_fields %} + partition_fields: {{partition_fields}} + {% endif %} + {% if table_properties %} + table_properties: {{table_properties}} + {% endif %} options: streaming: false From e35c1d207a5d5554ff27175ca07f2e4162cdcb20 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 25 Aug 2026 16:25:11 +0000 Subject: [PATCH 14/26] remove old writetolakehouse transform idea --- .../job-builder-util-transforms/provider_listing_template.yaml | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml b/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml index 16cc5244e0..4f8042d3da 100644 --- a/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml +++ b/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml @@ -4,5 +4,4 @@ - https://storage.googleapis.com/ transforms: CopyFilesToGCS: "copy_files_to_gcs.CopyFilesToGCS" - ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" - WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse" \ No newline at end of file + ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" \ No newline at end of file From a0cf3b445f584e6fe628881d9b8dd90d7c9f02b7 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 25 Aug 2026 16:32:21 +0000 Subject: [PATCH 15/26] spotless --- .../yaml/DeltaLakeToLakehouseYaml.java | 52 +++++++++---------- .../yaml/DeltaLakeToLakehouseYamlIT.java | 3 +- 2 files changed, 26 insertions(+), 29 deletions(-) diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java index 0b565efa32..1c105dbe44 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java @@ -26,13 +26,20 @@ category = TemplateCategory.BATCH, type = Template.TemplateType.YAML, displayName = "Delta Lake to Lakehouse", - description = "The Delta Lake to Lakehouse template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Lakehouse table.", + description = + "The Delta Lake to Lakehouse template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Lakehouse table.", flexContainerName = "pipeline-yaml", yamlTemplateFile = "DeltaLakeToLakehouse.yaml", - filesToCopy = {"main.py", "requirements.txt", "options/deltalake_options.yaml", "options/lakehouse_options.yaml"}, + filesToCopy = { + "main.py", + "requirements.txt", + "options/deltalake_options.yaml", + "options/lakehouse_options.yaml" + }, documentation = "", contactInformation = "https://cloud.google.com/support", - requirements = {"The Input Delta Lake table must exist and be accessible.", + requirements = { + "The Input Delta Lake table must exist and be accessible.", "The Output Iceberg table must exist or be created, and the warehouse must be accessible." }, streaming = false, @@ -45,8 +52,7 @@ public interface DeltaLakeToLakehouseYaml { optional = false, description = "A GCS path to the Delta Lake table.", helpText = "The GCS path to the Delta Lake table, e.g., gs://your-bucket/path/to/table.", - example = "gs://your-bucket/path/to/table" - ) + example = "gs://your-bucket/path/to/table") @Validation.Required String getDeltaLakeTable(); @@ -56,9 +62,9 @@ public interface DeltaLakeToLakehouseYaml { optional = true, description = "Properties passed to Hadoop Configuration.", helpText = "A map of properties to pass to Hadoop Configuration, e.g. key-value pairs.", - example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" - ) - @Default.String("{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\", \"fs.AbstractFileSystem.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS\", \"fs.gs.auth.type\": \"APPLICATION_DEFAULT\", \"fs.gs.project.id\": \"\"}") + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}") + @Default.String( + "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\", \"fs.AbstractFileSystem.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS\", \"fs.gs.auth.type\": \"APPLICATION_DEFAULT\", \"fs.gs.project.id\": \"\"}") String getDeltaLakeHadoopConfig(); @TemplateParameter.Text( @@ -67,8 +73,7 @@ public interface DeltaLakeToLakehouseYaml { optional = false, description = "A fully-qualified table identifier.", helpText = "A fully-qualified table identifier, e.g., my_dataset.my_table.", - example = "my_dataset.my_table" - ) + example = "my_dataset.my_table") @Validation.Required String getLakehouseTable(); @@ -78,8 +83,7 @@ public interface DeltaLakeToLakehouseYaml { optional = false, description = "Name of the catalog containing the table.", helpText = "The name of the Lakehouse catalog that contains the table.", - example = "my_hadoop_catalog" - ) + example = "my_hadoop_catalog") @Validation.Required String getLakehouseCatalogName(); @@ -89,8 +93,7 @@ public interface DeltaLakeToLakehouseYaml { optional = false, description = "Properties used to set up the Lakehouse catalog.", helpText = "A map of properties for setting up the Lakehouse catalog.", - example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}" - ) + example = "{\"type\": \"hadoop\", \"warehouse\": \"gs://your-bucket/warehouse\"}") @Validation.Required String getLakehouseCatalogProperties(); @@ -100,8 +103,7 @@ public interface DeltaLakeToLakehouseYaml { optional = true, description = "Properties passed to the Hadoop Configuration.", helpText = "A map of properties to pass to the Hadoop Configuration.", - example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}" - ) + example = "{\"fs.gs.impl\": \"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem\"}") String getLakehouseConfigProperties(); @TemplateParameter.Text( @@ -110,8 +112,7 @@ public interface DeltaLakeToLakehouseYaml { optional = true, description = "A list of field names to drop from the input record before writing.", helpText = "A list of field names to drop. Mutually exclusive with 'keep' and 'only'.", - example = "[\"field_to_drop_1\", \"field_to_drop_2\"]" - ) + example = "[\"field_to_drop_1\", \"field_to_drop_2\"]") String getLakehouseDrop(); @TemplateParameter.Text( @@ -120,8 +121,7 @@ public interface DeltaLakeToLakehouseYaml { optional = true, description = "An optional filter expression to apply to the input records.", helpText = "A filter expression to apply to records from the Lakehouse table.", - example = "age > 18" - ) + example = "age > 18") String getLakehouseFilter(); @TemplateParameter.Text( @@ -130,8 +130,7 @@ public interface DeltaLakeToLakehouseYaml { optional = true, description = "A list of field names to keep in the input record.", helpText = "A list of field names to keep. Mutually exclusive with 'drop' and 'only'.", - example = "[\"field_to_keep_1\", \"field_to_keep_2\"]" - ) + example = "[\"field_to_keep_1\", \"field_to_keep_2\"]") String getLakehouseKeep(); @TemplateParameter.Text( @@ -140,8 +139,7 @@ public interface DeltaLakeToLakehouseYaml { optional = true, description = "The name of a single record field that should be written.", helpText = "The name of a single field to write. Mutually exclusive with 'keep' and 'drop'.", - example = "my_record_field" - ) + example = "my_record_field") String getLakehouseOnly(); @TemplateParameter.Text( @@ -150,8 +148,7 @@ public interface DeltaLakeToLakehouseYaml { optional = true, description = "Fields used to create a partition spec for new tables.", helpText = "A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category'].", - example = "[\"day(ts)\", \"bucket(id, 4)\"]" - ) + example = "[\"day(ts)\", \"bucket(id, 4)\"]") String getLakehousePartitionFields(); @TemplateParameter.Text( @@ -160,7 +157,6 @@ public interface DeltaLakeToLakehouseYaml { optional = true, description = "Lakehouse table properties to be set on table creation.", helpText = "A map of Lakehouse table properties to set when the table is created.", - example = "{\"commit.retry.num-retries\": \"2\"}" - ) + example = "{\"commit.retry.num-retries\": \"2\"}") String getLakehouseTableProperties(); } diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java index 0c6456bd8b..7fbafb18e6 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java @@ -132,7 +132,8 @@ public void testDeltaLakeToLakehouse() throws IOException { .addParameter("lakehouseTable", lakehouseTableIdentifier) .addParameter("lakehouseCatalogName", CATALOG_NAME) .addParameter( - "lakehouseCatalogProperties", new org.json.JSONObject(getCatalogProperties()).toString()); + "lakehouseCatalogProperties", + new org.json.JSONObject(getCatalogProperties()).toString()); LaunchInfo info = launchTemplate(options); assertThatPipeline(info).isRunning(); From a323e9463d1f2d92e02934726788c62e2496881c Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 25 Aug 2026 19:42:43 +0000 Subject: [PATCH 16/26] change to renamed transform --- yaml/src/main/yaml/DeltaLakeToLakehouse.yaml | 44 ++++++++------------ 1 file changed, 18 insertions(+), 26 deletions(-) diff --git a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml index c51a441320..dd6ba9cfe3 100644 --- a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml +++ b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml @@ -71,33 +71,25 @@ providers: transforms: ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" - - type: yaml + - type: renaming transforms: - WriteToLakehouse: - body: | - type: WriteToIceberg - config: - table: "{{table}}" - catalog_name: "{{catalog_name}}" - catalog_properties: {{catalog_properties}} - {% if config_properties %} - config_properties: {{config_properties}} - {% endif %} - {% if drop %} - drop: {{drop}} - {% endif %} - {% if keep %} - keep: {{keep}} - {% endif %} - {% if only %} - only: {{only}} - {% endif %} - {% if partition_fields %} - partition_fields: {{partition_fields}} - {% endif %} - {% if table_properties %} - table_properties: {{table_properties}} - {% endif %} + WriteToLakehouse: WriteToIceberg + config: + mappings: + WriteToLakehouse: + table: table + catalog_name: catalog_name + catalog_properties: catalog_properties + config_properties: config_properties + drop: drop + keep: keep + only: only + partition_fields: partition_fields + table_properties: table_properties + underlying_provider: + type: python + transforms: + WriteToIceberg: 'apache_beam.yaml.yaml_io.write_to_iceberg' options: streaming: false From bf3e10cbba33718642f79039f7c3c3c0c6e5fb81 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Fri, 28 Aug 2026 14:15:02 +0000 Subject: [PATCH 17/26] address Delta client comment --- yaml/pom.xml | 1 - .../yaml/DeltaLakeToLakehouseYamlIT.java | 196 ++++++++++++++---- 2 files changed, 159 insertions(+), 38 deletions(-) diff --git a/yaml/pom.xml b/yaml/pom.xml index 74c0de580e..942fd49fef 100644 --- a/yaml/pom.xml +++ b/yaml/pom.xml @@ -131,7 +131,6 @@ org.apache.beam beam-sdks-java-io-delta ${beam.version} - runtime com.github.jbellis diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java index 7fbafb18e6..62cf599914 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java @@ -22,16 +22,38 @@ import com.google.cloud.teleport.it.iceberg.IcebergResourceManager; import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import io.delta.kernel.DataWriteContext; +import io.delta.kernel.Operation; +import io.delta.kernel.Table; +import io.delta.kernel.Transaction; +import io.delta.kernel.TransactionBuilder; +import io.delta.kernel.TransactionCommitResult; +import io.delta.kernel.data.ColumnVector; +import io.delta.kernel.data.ColumnarBatch; +import io.delta.kernel.data.FilteredColumnarBatch; +import io.delta.kernel.defaults.engine.DefaultEngine; +import io.delta.kernel.defaults.internal.data.DefaultColumnarBatch; +import io.delta.kernel.engine.Engine; +import io.delta.kernel.types.DataType; +import io.delta.kernel.types.DoubleType; +import io.delta.kernel.types.StringType; +import io.delta.kernel.types.StructType; +import io.delta.kernel.utils.CloseableIterable; +import io.delta.kernel.utils.CloseableIterator; +import io.delta.kernel.utils.DataFileStatus; import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.UUID; import org.apache.beam.it.common.PipelineLauncher.LaunchConfig; import org.apache.beam.it.common.PipelineLauncher.LaunchInfo; import org.apache.beam.it.common.PipelineOperator; import org.apache.beam.it.common.utils.ResourceManagerUtils; import org.apache.beam.it.gcp.TemplateTestBase; -import org.apache.beam.it.gcp.artifacts.utils.ParquetTestUtil; +import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.Schema; import org.apache.iceberg.data.Record; import org.apache.iceberg.types.Types; @@ -74,46 +96,146 @@ public void tearDown() { } @Test - public void testDeltaLakeToLakehouse() throws IOException { + public void testDeltaLakeToLakehouse() throws Exception { // 1. Arrange: Create Delta Lake source table in GCS String deltaTableDir = "delta-table"; - org.apache.avro.Schema avroSchema = - new org.apache.avro.Schema.Parser() - .parse( - "{\"type\":\"record\",\"name\":\"test_record\",\"fields\":[" - + "{\"name\":\"id\",\"type\":\"string\"}," - + "{\"name\":\"state\",\"type\":\"string\"}," - + "{\"name\":\"price\",\"type\":\"double\"}" - + "]}"); - org.apache.avro.generic.GenericRecord avroRecord = - new org.apache.avro.generic.GenericData.Record(avroSchema); - avroRecord.put("id", "007"); - avroRecord.put("state", "CA"); - avroRecord.put("price", 26.23); - byte[] parquetBytes = ParquetTestUtil.createParquetFile(avroSchema, List.of(avroRecord)); - - // Upload data Parquet file - gcsClient.createArtifact(deltaTableDir + "/part-00000.parquet", parquetBytes); - - // Create and upload Delta Lake transaction log - String commitContent = - "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n" - + "{\"metaData\":{\"id\":\"test-id\",\"format\":{\"provider\":\"parquet\",\"options\":{}}," - + "\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" - + "{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," - + "{\\\"name\\\":\\\"state\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}," - + "{\\\"name\\\":\\\"price\\\",\\\"type\\\":\\\"double\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}" - + "]}\",\"partitionColumns\":[],\"configuration\":{},\"createdAt\":123456789}}\n" - + "{\"add\":{\"path\":\"part-00000.parquet\",\"partitionValues\":{},\"size\":" - + parquetBytes.length - + "," - + "\"modificationTime\":123456789,\"dataChange\":true}}"; - - gcsClient.createArtifact( - deltaTableDir + "/_delta_log/00000000000000000000.json", commitContent); - String deltaTableGcsPath = getGcsPath(deltaTableDir); + Configuration configuration = new Configuration(); + getGcsHadoopConfig().forEach(configuration::set); + Engine engine = DefaultEngine.create(configuration); + Table table = Table.forPath(engine, deltaTableGcsPath); + + StructType deltaSchema = + new StructType() + .add("id", StringType.STRING) + .add("state", StringType.STRING) + .add("price", DoubleType.DOUBLE); + + TransactionBuilder txnBuilder = + table.createTransactionBuilder(engine, "DeltaLakeToLakehouseYamlIT", Operation.CREATE_TABLE); + txnBuilder = txnBuilder.withSchema(engine, deltaSchema); + Transaction txn = txnBuilder.build(engine); + io.delta.kernel.data.Row txnState = txn.getTransactionState(engine); + + ColumnVector idVector = + new ColumnVector() { + @Override + public DataType getDataType() { + return StringType.STRING; + } + + @Override + public int getSize() { + return 1; + } + + @Override + public void close() {} + + @Override + public boolean isNullAt(int rowId) { + return false; + } + + @Override + public String getString(int rowId) { + return "007"; + } + }; + + ColumnVector stateVector = + new ColumnVector() { + @Override + public DataType getDataType() { + return StringType.STRING; + } + + @Override + public int getSize() { + return 1; + } + + @Override + public void close() {} + + @Override + public boolean isNullAt(int rowId) { + return false; + } + + @Override + public String getString(int rowId) { + return "CA"; + } + }; + + ColumnVector priceVector = + new ColumnVector() { + @Override + public DataType getDataType() { + return DoubleType.DOUBLE; + } + + @Override + public int getSize() { + return 1; + } + + @Override + public void close() {} + + @Override + public boolean isNullAt(int rowId) { + return false; + } + + @Override + public double getDouble(int rowId) { + return 26.23; + } + }; + + ColumnVector[] vectors = new ColumnVector[] {idVector, stateVector, priceVector}; + ColumnarBatch columnarBatch = new DefaultColumnarBatch(1, deltaSchema, vectors); + FilteredColumnarBatch filteredBatch = + new FilteredColumnarBatch(columnarBatch, Optional.empty()); + + CloseableIterator data = + io.delta.kernel.internal.util.Utils.toCloseableIterator( + Collections.singletonList(filteredBatch).iterator()); + + CloseableIterator physicalData = + Transaction.transformLogicalData(engine, txnState, data, Collections.emptyMap()); + + DataWriteContext writeContext = + Transaction.getWriteContext(engine, txnState, Collections.emptyMap()); + + CloseableIterator dataFiles = + engine + .getParquetHandler() + .writeParquetFiles( + writeContext.getTargetDirectory(), + physicalData, + writeContext.getStatisticsColumns()); + + CloseableIterator dataActions = + Transaction.generateAppendActions(engine, txnState, dataFiles, writeContext); + + List addActionsList = new ArrayList<>(); + while (dataActions.hasNext()) { + addActionsList.add(dataActions.next()); + } + + CloseableIterable dataActionsIterable = + CloseableIterable.inMemoryIterable( + io.delta.kernel.internal.util.Utils.toCloseableIterator(addActionsList.iterator())); + + TransactionCommitResult commitResult = txn.commit(engine, dataActionsIterable); + if (commitResult.getVersion() < 0) { + throw new RuntimeException("Table creation/write failed"); + } + // 2. Arrange: Create destination Lakehouse table icebergResourceManager.createNamespace(namespace); Schema icebergSchema = From f1580c4960b2e1fa2769710fb2a4e084c3a4f24f Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Fri, 28 Aug 2026 14:30:24 +0000 Subject: [PATCH 18/26] spotless --- .../teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java index 62cf599914..eec45127e4 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java @@ -113,7 +113,8 @@ public void testDeltaLakeToLakehouse() throws Exception { .add("price", DoubleType.DOUBLE); TransactionBuilder txnBuilder = - table.createTransactionBuilder(engine, "DeltaLakeToLakehouseYamlIT", Operation.CREATE_TABLE); + table.createTransactionBuilder( + engine, "DeltaLakeToLakehouseYamlIT", Operation.CREATE_TABLE); txnBuilder = txnBuilder.withSchema(engine, deltaSchema); Transaction txn = txnBuilder.build(engine); io.delta.kernel.data.Row txnState = txn.getTransactionState(engine); From 3cfd16cc983ee79acd8404d2a9c8c771b565dd46 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Fri, 28 Aug 2026 20:34:37 +0000 Subject: [PATCH 19/26] fix generated file --- .../cloud/teleport/templates/yaml/PubSubToBigQueryYaml.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/PubSubToBigQueryYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/PubSubToBigQueryYaml.java index 4d916e1ff1..61a7f83c36 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/PubSubToBigQueryYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/PubSubToBigQueryYaml.java @@ -138,7 +138,7 @@ public interface PubSubToBigQueryYaml { optional = false, description = "BigQuery table", helpText = - "BigQuery table location to write the output to or read from. The name should be in the format :.`. For write, the table's schema must match input objects.", + "BigQuery table location to write the output to or read from. The name should be in the format :.. For write, the table's schema must match input objects.", example = "") @Validation.Required String getTable(); From 12f799e905513b48ba93f6be70ed917a86741258 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Fri, 28 Aug 2026 23:20:01 +0000 Subject: [PATCH 20/26] create write to lakehouse --- .../provider_listing_template.yaml | 3 +- .../pyproject.toml | 1 + .../write_to_lakehouse.py | 72 ++++++++++++++++++ .../write_to_lakehouse_test.py | 75 +++++++++++++++++++ yaml/README_DeltaLake_To_Lakehouse_Yaml.md | 2 +- .../yaml/DeltaLakeToLakehouseYaml.java | 4 +- yaml/src/main/yaml/DeltaLakeToLakehouse.yaml | 25 +------ 7 files changed, 156 insertions(+), 26 deletions(-) create mode 100644 python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py create mode 100644 python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py diff --git a/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml b/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml index 4f8042d3da..16cc5244e0 100644 --- a/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml +++ b/python/src/main/python/job-builder-util-transforms/provider_listing_template.yaml @@ -4,4 +4,5 @@ - https://storage.googleapis.com/ transforms: CopyFilesToGCS: "copy_files_to_gcs.CopyFilesToGCS" - ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" \ No newline at end of file + ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" + WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse" \ No newline at end of file diff --git a/python/src/main/python/job-builder-util-transforms/pyproject.toml b/python/src/main/python/job-builder-util-transforms/pyproject.toml index e427abfe4d..feb9b12c2d 100644 --- a/python/src/main/python/job-builder-util-transforms/pyproject.toml +++ b/python/src/main/python/job-builder-util-transforms/pyproject.toml @@ -6,6 +6,7 @@ authors = ["Google Cloud Platform"] packages = [ { include = "copy_files_to_gcs.py" }, { include = "read_from_delta_lake.py" }, + { include = "write_to_lakehouse.py" }, ] [tool.poetry.dependencies] diff --git a/python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py b/python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py new file mode 100644 index 0000000000..77612b215b --- /dev/null +++ b/python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py @@ -0,0 +1,72 @@ +# Copyright 2026 Google LLC +# +# Licensed 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. + +"""Module containing transforms to write data to Lakehouse tables.""" + +from typing import Iterable, Mapping, Optional +from apache_beam.transforms import PTransform +from apache_beam.yaml.yaml_io import write_to_iceberg + + +class WriteToLakehouse(PTransform): + """A PTransform that writes data to a Lakehouse table. + + Currently, it wraps the Apache Iceberg sink. + """ + + def __init__( + self, + table: str, + catalog_name: Optional[str] = None, + catalog_properties: Optional[Mapping[str, str]] = None, + config_properties: Optional[Mapping[str, str]] = None, + partition_fields: Optional[Iterable[str]] = None, + table_properties: Optional[Mapping[str, str]] = None, + triggering_frequency_seconds: Optional[int] = None, + keep: Optional[Iterable[str]] = None, + drop: Optional[Iterable[str]] = None, + only: Optional[str] = None, + distribution_mode: Optional[str] = None, + autosharding: Optional[bool] = None, + ): + super().__init__() + self.table = table + self.catalog_name = catalog_name + self.catalog_properties = catalog_properties + self.config_properties = config_properties + self.partition_fields = partition_fields + self.table_properties = table_properties + self.triggering_frequency_seconds = triggering_frequency_seconds + self.keep = keep + self.drop = drop + self.only = only + self.distribution_mode = distribution_mode + self.autosharding = autosharding + + def expand(self, pcoll): + """Expands the WriteToLakehouse transform.""" + return pcoll | write_to_iceberg( + table=self.table, + catalog_name=self.catalog_name, + catalog_properties=self.catalog_properties, + config_properties=self.config_properties, + partition_fields=self.partition_fields, + table_properties=self.table_properties, + triggering_frequency_seconds=self.triggering_frequency_seconds, + keep=self.keep, + drop=self.drop, + only=self.only, + distribution_mode=self.distribution_mode, + autosharding=self.autosharding, + ) diff --git a/python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py b/python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py new file mode 100644 index 0000000000..36cb474f0d --- /dev/null +++ b/python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py @@ -0,0 +1,75 @@ +# Copyright 2026 Google LLC +# +# Licensed 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. + +import unittest +from unittest.mock import MagicMock, patch +from write_to_lakehouse import WriteToLakehouse + + +class WriteToLakehouseTest(unittest.TestCase): + + @patch("write_to_lakehouse.write_to_iceberg") + def test_write_to_lakehouse(self, mock_write_to_iceberg): + mock_transform = MagicMock() + mock_write_to_iceberg.return_value = mock_transform + + table = "lakehouse_catalog.dataset.table" + catalog_name = "lakehouse_catalog" + catalog_properties = {"type": "hadoop", "warehouse": "gs://bucket/warehouse"} + config_properties = {"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"} + partition_fields = ["day(ts)", "category"] + table_properties = {"commit.retry.num-retries": "2"} + triggering_frequency_seconds = 60 + keep = ["field1", "field2"] + drop = ["field3"] + only = "field4" + distribution_mode = "hash" + autosharding = True + + transform = WriteToLakehouse( + table=table, + catalog_name=catalog_name, + catalog_properties=catalog_properties, + config_properties=config_properties, + partition_fields=partition_fields, + table_properties=table_properties, + triggering_frequency_seconds=triggering_frequency_seconds, + keep=keep, + drop=drop, + only=only, + distribution_mode=distribution_mode, + autosharding=autosharding, + ) + + pcoll = MagicMock() + transform.expand(pcoll) + + mock_write_to_iceberg.assert_called_once_with( + table=table, + catalog_name=catalog_name, + catalog_properties=catalog_properties, + config_properties=config_properties, + partition_fields=partition_fields, + table_properties=table_properties, + triggering_frequency_seconds=triggering_frequency_seconds, + keep=keep, + drop=drop, + only=only, + distribution_mode=distribution_mode, + autosharding=autosharding, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/yaml/README_DeltaLake_To_Lakehouse_Yaml.md b/yaml/README_DeltaLake_To_Lakehouse_Yaml.md index a9519793ad..4a64d6c35b 100644 --- a/yaml/README_DeltaLake_To_Lakehouse_Yaml.md +++ b/yaml/README_DeltaLake_To_Lakehouse_Yaml.md @@ -2,7 +2,7 @@ Delta Lake to Lakehouse template --- The Delta Lake to Lakehouse template is a batch pipeline that reads data from a -Delta Lake table and outputs the records to an Lakehouse table. +Delta Lake table and outputs the records to a Lakehouse table. diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java index 1c105dbe44..ba29b2b4ae 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java @@ -27,7 +27,7 @@ type = Template.TemplateType.YAML, displayName = "Delta Lake to Lakehouse", description = - "The Delta Lake to Lakehouse template is a batch pipeline that reads data from a Delta Lake table and outputs the records to an Lakehouse table.", + "The Delta Lake to Lakehouse template is a batch pipeline that reads data from a Delta Lake table and outputs the records to a Lakehouse table.", flexContainerName = "pipeline-yaml", yamlTemplateFile = "DeltaLakeToLakehouse.yaml", filesToCopy = { @@ -40,7 +40,7 @@ contactInformation = "https://cloud.google.com/support", requirements = { "The Input Delta Lake table must exist and be accessible.", - "The Output Iceberg table must exist or be created, and the warehouse must be accessible." + "The Output Lakehouse table must exist or be created, and the warehouse must be accessible." }, streaming = false, hidden = false) diff --git a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml index dd6ba9cfe3..0ee7b2316b 100644 --- a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml +++ b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml @@ -5,7 +5,7 @@ template: display_name: "Delta Lake to Lakehouse" description: > The Delta Lake to Lakehouse template is a batch pipeline that reads data from a Delta Lake table - and outputs the records to an Lakehouse table. + and outputs the records to a Lakehouse table. flex_container_name: "pipeline-yaml" yamlTemplateFile: "DeltaLakeToLakehouse.yaml" filesToCopy: > @@ -13,7 +13,7 @@ template: contactInformation: "https://cloud.google.com/support" requirements: { "The Input Delta Lake table must exist and be accessible.", - "The Output Iceberg table must exist or be created, and the warehouse must be accessible." + "The Output Lakehouse table must exist or be created, and the warehouse must be accessible." } streaming: false hidden: false @@ -70,26 +70,7 @@ providers: - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-07-20/job_builder_util_transforms-0.2.0.tar.gz transforms: ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" - - - type: renaming - transforms: - WriteToLakehouse: WriteToIceberg - config: - mappings: - WriteToLakehouse: - table: table - catalog_name: catalog_name - catalog_properties: catalog_properties - config_properties: config_properties - drop: drop - keep: keep - only: only - partition_fields: partition_fields - table_properties: table_properties - underlying_provider: - type: python - transforms: - WriteToIceberg: 'apache_beam.yaml.yaml_io.write_to_iceberg' + WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse" options: streaming: false From aa73a256c094002e0f86498377364790088edb75 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sat, 29 Aug 2026 01:55:43 +0000 Subject: [PATCH 21/26] change package --- yaml/src/main/yaml/DeltaLakeToLakehouse.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml index 0ee7b2316b..4721906458 100644 --- a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml +++ b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml @@ -67,7 +67,7 @@ providers: - type: pythonPackage config: packages: - - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-07-20/job_builder_util_transforms-0.2.0.tar.gz + - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-28/job_builder_util_transforms-0.3.0.tar.gz transforms: ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse" From a6e7bad0deaac196d05b0213221f601a3fe63038 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sat, 29 Aug 2026 14:11:58 +0000 Subject: [PATCH 22/26] oops - used wrong assumed date --- yaml/src/main/yaml/DeltaLakeToLakehouse.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml index 4721906458..3e73d0f39f 100644 --- a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml +++ b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml @@ -67,7 +67,7 @@ providers: - type: pythonPackage config: packages: - - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-28/job_builder_util_transforms-0.3.0.tar.gz + - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-29/job_builder_util_transforms-0.3.0.tar.gz transforms: ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse" From f0e93b37f93c59e78c399868a434444624004045 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sat, 29 Aug 2026 16:00:44 +0000 Subject: [PATCH 23/26] fix uri catalog version --- .../teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java index eec45127e4..f1438c434e 100644 --- a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java @@ -287,7 +287,7 @@ protected PipelineOperator.Config createConfig(LaunchInfo info) { private Map getCatalogProperties() { return Map.of( "type", "rest", - "uri", "https://biglake.googleapis.com/iceberg/v1beta/restcatalog", + "uri", "https://biglake.googleapis.com/iceberg/v1/restcatalog", "warehouse", "gs://" + gcsClient.getBucket(), "header.x-goog-user-project", PROJECT, "rest.auth.type", "org.apache.iceberg.gcp.auth.GoogleAuthManager", From 1300f828d5ebaf3c949b6de406974ac57d73eb14 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sat, 29 Aug 2026 17:19:44 +0000 Subject: [PATCH 24/26] update jar to to be the same --- .../python/job-builder-util-transforms/read_from_delta_lake.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py b/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py index 7fcdfa974b..10bbd2eb5d 100644 --- a/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py +++ b/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py @@ -61,7 +61,7 @@ def expand(self, pbegin): ) else: expansion_service = JavaJarExpansionService( - 'https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-07-20/expansion-service-custom-0.2.0.jar' + 'https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-29/expansion-service-custom-0.3.0.jar' ) return pbegin | SchemaAwareExternalTransform( From 6333399a55332eab83961f0d9e4743028ba9e86d Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sat, 29 Aug 2026 18:41:10 +0000 Subject: [PATCH 25/26] change to managed and mimic deltalake --- .../read_from_delta_lake.py | 2 +- .../write_to_lakehouse.py | 61 ++++++++++++++----- .../write_to_lakehouse_test.py | 40 +++++++----- 3 files changed, 71 insertions(+), 32 deletions(-) diff --git a/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py b/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py index 10bbd2eb5d..0255268bff 100644 --- a/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py +++ b/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py @@ -61,7 +61,7 @@ def expand(self, pbegin): ) else: expansion_service = JavaJarExpansionService( - 'https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-29/expansion-service-custom-0.3.0.jar' + 'https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-29/expansion-service-custom-0.3.1.jar' ) return pbegin | SchemaAwareExternalTransform( diff --git a/python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py b/python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py index 77612b215b..ba753dac54 100644 --- a/python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py +++ b/python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py @@ -15,14 +15,17 @@ """Module containing transforms to write data to Lakehouse tables.""" from typing import Iterable, Mapping, Optional +from apache_beam.options.pipeline_options import CrossLanguageOptions +from apache_beam.transforms import managed +from apache_beam.transforms.external import BeamJarExpansionService +from apache_beam.transforms.external import JavaJarExpansionService from apache_beam.transforms import PTransform -from apache_beam.yaml.yaml_io import write_to_iceberg class WriteToLakehouse(PTransform): """A PTransform that writes data to a Lakehouse table. - Currently, it wraps the Apache Iceberg sink. + Currently, it wraps the Apache Iceberg sink using the unified expansion service. """ def __init__( @@ -56,17 +59,45 @@ def __init__( def expand(self, pcoll): """Expands the WriteToLakehouse transform.""" - return pcoll | write_to_iceberg( - table=self.table, - catalog_name=self.catalog_name, - catalog_properties=self.catalog_properties, - config_properties=self.config_properties, - partition_fields=self.partition_fields, - table_properties=self.table_properties, - triggering_frequency_seconds=self.triggering_frequency_seconds, - keep=self.keep, - drop=self.drop, - only=self.only, - distribution_mode=self.distribution_mode, - autosharding=self.autosharding, + options = pcoll.pipeline.options + beam_services = options.view_as(CrossLanguageOptions).beam_services or {} + if 'sdks:java:io:expansion-service:shadowJar' in beam_services: + expansion_service = BeamJarExpansionService( + 'sdks:java:io:expansion-service:shadowJar' + ) + else: + expansion_service = JavaJarExpansionService( + 'https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-29/expansion-service-custom-0.3.1.jar' + ) + + config = { + 'table': self.table, + } + if self.catalog_name is not None: + config['catalog_name'] = self.catalog_name + if self.catalog_properties is not None: + config['catalog_properties'] = dict(self.catalog_properties) + if self.config_properties is not None: + config['config_properties'] = dict(self.config_properties) + if self.partition_fields is not None: + config['partition_fields'] = list(self.partition_fields) + if self.table_properties is not None: + config['table_properties'] = dict(self.table_properties) + if self.triggering_frequency_seconds is not None: + config['triggering_frequency_seconds'] = self.triggering_frequency_seconds + if self.keep is not None: + config['keep'] = list(self.keep) + if self.drop is not None: + config['drop'] = list(self.drop) + if self.only is not None: + config['only'] = self.only + if self.distribution_mode is not None: + config['distribution_mode'] = self.distribution_mode + if self.autosharding is not None: + config['autosharding'] = self.autosharding + + return pcoll | managed.Write( + "iceberg", + config=config, + expansion_service=expansion_service, ) diff --git a/python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py b/python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py index 36cb474f0d..a5dcd8cb05 100644 --- a/python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py +++ b/python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py @@ -19,10 +19,10 @@ class WriteToLakehouseTest(unittest.TestCase): - @patch("write_to_lakehouse.write_to_iceberg") - def test_write_to_lakehouse(self, mock_write_to_iceberg): + @patch("write_to_lakehouse.managed.Write") + def test_write_to_lakehouse(self, mock_managed_write): mock_transform = MagicMock() - mock_write_to_iceberg.return_value = mock_transform + mock_managed_write.return_value = mock_transform table = "lakehouse_catalog.dataset.table" catalog_name = "lakehouse_catalog" @@ -53,22 +53,30 @@ def test_write_to_lakehouse(self, mock_write_to_iceberg): ) pcoll = MagicMock() + pcoll.pipeline.options.view_as.return_value.beam_services = {} transform.expand(pcoll) - mock_write_to_iceberg.assert_called_once_with( - table=table, - catalog_name=catalog_name, - catalog_properties=catalog_properties, - config_properties=config_properties, - partition_fields=partition_fields, - table_properties=table_properties, - triggering_frequency_seconds=triggering_frequency_seconds, - keep=keep, - drop=drop, - only=only, - distribution_mode=distribution_mode, - autosharding=autosharding, + mock_managed_write.assert_called_once() + args, kwargs = mock_managed_write.call_args + self.assertEqual(args[0], "iceberg") + self.assertEqual( + kwargs["config"], + { + "table": table, + "catalog_name": catalog_name, + "catalog_properties": catalog_properties, + "config_properties": config_properties, + "partition_fields": partition_fields, + "table_properties": table_properties, + "triggering_frequency_seconds": triggering_frequency_seconds, + "keep": keep, + "drop": drop, + "only": only, + "distribution_mode": distribution_mode, + "autosharding": autosharding, + }, ) + self.assertIsNotNone(kwargs["expansion_service"]) if __name__ == "__main__": From ef2df8ee5e3619bf605f69fff074582c3694e944 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sat, 29 Aug 2026 18:53:42 +0000 Subject: [PATCH 26/26] update one more version jar --- yaml/src/main/yaml/DeltaLakeToLakehouse.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml index 3e73d0f39f..9edab848c6 100644 --- a/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml +++ b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml @@ -67,7 +67,7 @@ providers: - type: pythonPackage config: packages: - - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-29/job_builder_util_transforms-0.3.0.tar.gz + - https://storage.googleapis.com/dataflow-templates/extra-python-packages/2026-08-29/job_builder_util_transforms-0.3.1.tar.gz transforms: ReadFromDeltaLake: "read_from_delta_lake.ReadFromDeltaLake" WriteToLakehouse: "write_to_lakehouse.WriteToLakehouse"