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/read_from_delta_lake.py b/python/src/main/python/job-builder-util-transforms/read_from_delta_lake.py index 7fcdfa974b..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-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.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 new file mode 100644 index 0000000000..ba753dac54 --- /dev/null +++ b/python/src/main/python/job-builder-util-transforms/write_to_lakehouse.py @@ -0,0 +1,103 @@ +# 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.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 + + +class WriteToLakehouse(PTransform): + """A PTransform that writes data to a Lakehouse table. + + Currently, it wraps the Apache Iceberg sink using the unified expansion service. + """ + + 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.""" + 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 new file mode 100644 index 0000000000..a5dcd8cb05 --- /dev/null +++ b/python/src/test/python/job-builder-util-transforms/write_to_lakehouse_test.py @@ -0,0 +1,83 @@ +# 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.managed.Write") + def test_write_to_lakehouse(self, mock_managed_write): + mock_transform = MagicMock() + mock_managed_write.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() + pcoll.pipeline.options.view_as.return_value.beam_services = {} + transform.expand(pcoll) + + 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__": + unittest.main() diff --git a/yaml/README_DeltaLake_To_Lakehouse_Yaml.md b/yaml/README_DeltaLake_To_Lakehouse_Yaml.md new file mode 100644 index 0000000000..4a64d6c35b --- /dev/null +++ b/yaml/README_DeltaLake_To_Lakehouse_Yaml.md @@ -0,0 +1,254 @@ + +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 a Lakehouse 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`. +* **lakehouseTable**: A fully-qualified table identifier, e.g., my_dataset.my_table. For example, `my_dataset.my_table`. +* **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 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 Lakehouse 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/DeltaLakeToLakehouseYaml.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_Lakehouse_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_Lakehouse_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_Lakehouse_Yaml" + +### Required +export DELTA_LAKE_TABLE= +export LAKEHOUSE_TABLE= +export LAKEHOUSE_CATALOG_NAME= +export LAKEHOUSE_CATALOG_PROPERTIES= + +### Optional +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-lakehouse-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 "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: +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 LAKEHOUSE_TABLE= +export LAKEHOUSE_CATALOG_NAME= +export LAKEHOUSE_CATALOG_PROPERTIES= + +### Optional +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 \ +-DprojectId="$PROJECT" \ +-DbucketName="$BUCKET_NAME" \ +-Dregion="$REGION" \ +-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 +``` + +## 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_Lakehouse_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_lakehouse_yaml" { + + provider = google-beta + 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 = "" + 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/pom.xml b/yaml/pom.xml index d1590476af..01fe0fa925 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/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java new file mode 100644 index 0000000000..ba29b2b4ae --- /dev/null +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYaml.java @@ -0,0 +1,162 @@ +/* + * 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.Default; +import org.apache.beam.sdk.options.Validation; + +@Template( + name = "DeltaLake_To_Lakehouse_Yaml", + 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 a Lakehouse table.", + flexContainerName = "pipeline-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.", + "The Output Lakehouse table must exist or be created, and the warehouse must be accessible." + }, + streaming = false, + hidden = false) +public interface DeltaLakeToLakehouseYaml { + + @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\"}") + @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 = "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") + @Validation.Required + String getLakehouseTable(); + + @TemplateParameter.Text( + order = 4, + name = "lakehouseCatalogName", + 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") + @Validation.Required + String getLakehouseCatalogName(); + + @TemplateParameter.Text( + order = 5, + name = "lakehouseCatalogProperties", + 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\"}") + @Validation.Required + String getLakehouseCatalogProperties(); + + @TemplateParameter.Text( + order = 6, + 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 getLakehouseConfigProperties(); + + @TemplateParameter.Text( + order = 7, + 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 getLakehouseDrop(); + + @TemplateParameter.Text( + order = 8, + 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 Lakehouse table.", + example = "age > 18") + String getLakehouseFilter(); + + @TemplateParameter.Text( + order = 9, + 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 getLakehouseKeep(); + + @TemplateParameter.Text( + order = 10, + 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 getLakehouseOnly(); + + @TemplateParameter.Text( + order = 11, + 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 getLakehousePartitionFields(); + + @TemplateParameter.Text( + order = 12, + name = "lakehouseTableProperties", + 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\"}") + String getLakehouseTableProperties(); +} 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(); diff --git a/yaml/src/main/python/generate_yaml_java_templates.py b/yaml/src/main/python/generate_yaml_java_templates.py index 71082760be..12d1c328cb 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,8 +161,10 @@ 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' + escaped_default = str(param["default"]).replace('"', '\\"') + param_code += f' @Default.String("{escaped_default}")\n' else: param_code += f" @Default.{java_type}({param['default']})\n" @@ -170,6 +173,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 +216,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}", 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..4099a4f0f7 --- /dev/null +++ b/yaml/src/main/python/options/deltalake_options.yaml @@ -0,0 +1,19 @@ +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 + 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/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 new file mode 100644 index 0000000000..9edab848c6 --- /dev/null +++ b/yaml/src/main/yaml/DeltaLakeToLakehouse.yaml @@ -0,0 +1,76 @@ +template: + name: "DeltaLake_To_Lakehouse_Yaml" + category: "BATCH" + type: "YAML" + 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 a Lakehouse table. + flex_container_name: "pipeline-yaml" + yamlTemplateFile: "DeltaLakeToLakehouse.yaml" + filesToCopy: > + {"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.", + "The Output Lakehouse table must exist or be created, and the warehouse must be accessible." + } + streaming: false + hidden: false + + options_file: + - "deltalake_options" + - "lakehouse_options" + + parameters: + - deltalake_read_options + - lakehouse_common_options + - lakehouse_write_options + +pipeline: + type: chain + transforms: + - type: ReadFromDeltaLake + name: ReadFromDeltaLake + config: + table: "{{ deltaLakeTable }}" + {% if deltaLakeHadoopConfig %} + hadoop_config: {{ deltaLakeHadoopConfig }} + {% endif %} + + - type: WriteToLakehouse + name: WriteToLakehouse + config: + table: "{{ lakehouseTable }}" + catalog_name: "{{ lakehouseCatalogName }}" + catalog_properties: {{ lakehouseCatalogProperties }} + {% if lakehouseConfigProperties %} + config_properties: {{ lakehouseConfigProperties }} + {% endif %} + {% if lakehouseDrop %} + drop: {{ lakehouseDrop }} + {% endif %} + {% if lakehouseKeep %} + keep: {{ lakehouseKeep }} + {% endif %} + {% if lakehouseOnly %} + only: {{ lakehouseOnly }} + {% endif %} + {% if lakehousePartitionFields %} + partition_fields: {{ lakehousePartitionFields }} + {% endif %} + {% if lakehouseTableProperties %} + table_properties: {{ lakehouseTableProperties }} + {% endif %} + +providers: + - type: pythonPackage + config: + packages: + - 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" + +options: + streaming: false 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 new file mode 100644 index 0000000000..f1438c434e --- /dev/null +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/DeltaLakeToLakehouseYamlIT.java @@ -0,0 +1,304 @@ +/* + * 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 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.hadoop.conf.Configuration; +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 DeltaLakeToLakehouseYaml} template. */ +@Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) +@TemplateIntegrationTest(DeltaLakeToLakehouseYaml.class) +@RunWith(JUnit4.class) +public class DeltaLakeToLakehouseYamlIT extends TemplateTestBase { + + private IcebergResourceManager icebergResourceManager; + + private static final String CATALOG_NAME = "hadoop_catalog"; + private final String namespace = + "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 { + 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 testDeltaLakeToLakehouse() throws Exception { + // 1. Arrange: Create Delta Lake source table in GCS + String deltaTableDir = "delta-table"; + 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 = + 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(lakehouseTableIdentifier, 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("lakehouseTable", lakehouseTableIdentifier) + .addParameter("lakehouseCatalogName", CATALOG_NAME) + .addParameter( + "lakehouseCatalogProperties", + 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(lakehouseTableIdentifier); + 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/v1/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); + } +}