feat: Deltalake to iceberg transfer (#1058)

* feat: Deltalake to iceberg transfer

* fix: add gcs deltalake to iceberg test case

* fix: syntax error

* fix: format and typo mistakes

* fix: change input location to align with other templates

* fix: improve code based on suggestions

* fix: SaveMode based on enum value provided by spark

* fix: SaveMode

* fix: using already available metastore

* fix: merge schema setting for iceberg table

Author: rajc242

java/.ci/dataproc-serverless-integration-tests.yaml

@@ -746,7 +746,7 @@ steps:
     script: |
       #!/usr/bin/env bash
       
-      echo "Running PubSub TO GCS avro"
+      echo "Running PubSub TO BigQuery avro"
       gcloud pubsub topics publish pubsub-to-bq --message='{"Template": "PUBSUBTOBQ", "Branch": "$BRANCH_NAME", "Commit": "$SHORT_SHA"}' --project=$GCP_PROJECT      
       cd java
       bin/start.sh \
@@ -758,6 +758,30 @@ steps:
         --templateProperty pubsub.bq.output.table=pubsubtobq
     waitFor: ['build-and-upload']
 
+  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk'
+    id: pubsub-to-bigtable
+    env:
+      - 'GCP_PROJECT=${_GCP_PROJECT}'
+      - 'GCS_STAGING_LOCATION=${_GCS_STAGING_LOCATION_BASE}'
+      - 'REGION=${_REGION}'
+      - 'SKIP_BUILD=true'
+      - 'SERVICE_ACCOUNT_NAME=${_SERVICE_ACCOUNT}'
+    script: |
+      #!/usr/bin/env bash
+      
+      echo "Running PubSub TO BigTable"
+      gcloud pubsub topics publish pubsub-to-bt --message='{"key": "1000", "name": "dataproc_pubsub_bt_test", "address": "GCP Cloud", "emp_no": "10"}' --project=$GCP_PROJECT      
+      cd java
+      bin/start.sh \
+        -- --template PUBSUBTOBIGTABLE \
+        --templateProperty pubsub.input.subscription=subs-pubsub-to-bt  \
+        --templateProperty pubsub.input.project.id=$GCP_PROJECT \
+        --templateProperty pubsub.bigtable.output.project.id=$GCP_PROJECT \
+        --templateProperty pubsub.bigtable.output.instance.id=bt-int-test \
+        --templateProperty pubsub.bigtable.output.table=employee \
+        --templateProperty pubsub.bigtable.catalog.location="gs://dataproc-templates_cloudbuild/integration-testing/bigtable/employeecatalog.json"
+    waitFor: ['build-and-upload']
+
   - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk'
     id: kafka-to-bigquery
     env:
@@ -883,6 +907,31 @@ steps:
         --templateProperty kafka.gcs.await.termination.timeout.ms=60000
     waitFor: ['build-and-upload']
 
+  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk'
+    id: gcs-deltalake-to-iceberg
+    env:
+      - 'GCP_PROJECT=${_GCP_PROJECT}'
+      - 'GCS_STAGING_LOCATION=${_GCS_STAGING_LOCATION_BASE}'
+      - 'REGION=${_REGION}'
+      - 'SKIP_BUILD=true'
+      - 'SERVICE_ACCOUNT_NAME=${_SERVICE_ACCOUNT}'
+      - 'DATAPROC_METASTORE_SERVICE=${_ENV_DATAPROC_METASTORE_SERVICE}'
+    script: |
+      #!/usr/bin/env bash
+      
+      echo "Running GCS Deltalake To Iceberg"      
+      cd java
+      bin/start.sh \
+        --metastore-service=$DATAPROC_METASTORE_SERVICE \
+        -- --template GCSDELTALAKETOICEBERG \
+        --templateProperty project.id=$GCP_PROJECT \
+        --templateProperty gcsdeltalaketoiceberg.input.path="gs://dataproc-templates_cloudbuild/integration-testing/GCSDELTALAKETOICEBERG/hive_warehouse/delta-table-trn-data" \
+        --templateProperty deltalake.version.as_of=0 \
+        --templateProperty iceberg.table.name="spark_catalog.default.iceberg_tbl_trn_data" \
+        --templateProperty iceberg.table.partition.columns="dt" \
+        --templateProperty iceberg.gcs.output.mode="overwrite"
+    waitFor: ['build-and-upload']
+
 options:
   logging: CLOUD_LOGGING_ONLY
   pubsubTopic: projects/dataproc-templates/topics/dpt-build

java/bin/start.sh

@@ -68,9 +68,18 @@ OPT_SPARK_VERSION="--version=1.2"
 OPT_PROJECT="--project=${GCP_PROJECT}"
 OPT_REGION="--region=${REGION}"
 OPT_JARS="--jars=file:///usr/lib/spark/connector/spark-avro.jar,${GCS_STAGING_LOCATION}/${JAR_FILE}"
-if [[ $OPT_SPARK_VERSION == *"=1.1"* ]]; then
-  echo "Dataproc Serverless Runtime 1.1 or CLUSTER Job Type Detected"
-	OPT_JARS="--jars=file:///usr/lib/spark/external/spark-avro.jar,${GCS_STAGING_LOCATION}/${JAR_FILE}"
+if [[ $JOB_TYPE == "SERVERLESS" ]]; then
+  if [[ $OPT_SPARK_VERSION == *"=1.1"* ]]; then
+    echo "Dataproc Serverless Runtime 1.1 Job Type Detected"
+  	OPT_JARS="--jars=file:///usr/lib/spark/external/spark-avro.jar,${GCS_STAGING_LOCATION}/${JAR_FILE}"
+  else
+    # We added Deltalake and Iceberg jars to align with serverless dataproc version compatibility
+    # Please check below internal document to make sure you aligned with what dataproc team provides i.e. jar version, path etc.
+    # https://docs.google.com/document/d/1VCv9sewWdFVu_2gzmZHFmikKBGxX7_zefhk0yJ-i0OA/edit?tab=t.0#heading=h.6ogrcxwzv7et
+    # Our java templates have scope provided for these dependencies to avoid version conflicting with dataproc
+    # We are doing this only for serverless dataproc because cluster dataproc is providing optional components which make sures dependencies available within cluster
+    OPT_JARS="--jars=file:///usr/lib/spark/connector/spark-avro.jar,gs://spark-lib/delta/delta-spark_2.12-3.2.0.jar,gs://spark-lib/delta/delta-storage-3.2.0.jar,gs://spark-lib/iceberg/iceberg-spark-runtime-3.5_2.12-1.6.1.jar,${GCS_STAGING_LOCATION}/${JAR_FILE}"
+  fi
 fi
 if [[ $JOB_TYPE == "CLUSTER" ]]; then
   if [[ -n "${CLUSTER}" ]]; then

java/pom.xml

@@ -44,6 +44,7 @@
     <maven.compiler.target>${java.version}</maven.compiler.target>
 
     <!-- Spark Version Aligned with Dataproc Serverless Spark Runtime 1.2 -->
+    <spark.major.version>3.5</spark.major.version>
     <spark.version>3.5.1</spark.version>
 
     <!-- GCP Dependencies. Only change when it is aligned with Spark -->
@@ -99,6 +100,12 @@
     <hbase.spark.connector.version>1.0.1</hbase.spark.connector.version>
     <hbase.client.version>2.6.0-hadoop3</hbase.client.version>
 
+    <!-- Iceberg Dependencies. Compatible with dataproc and spark major version. -->
+    <iceberg.version>1.6.1</iceberg.version>
+
+    <!-- Deltalake Dependencies. Compatible with spark major version. -->
+    <deltalake.version>3.2.0</deltalake.version>
+
     <!-- jakarta.validation -->
     <hibernate.validator.version>7.0.1.Final</hibernate.validator.version>
     <jakarta.el.api.version>4.0.0</jakarta.el.api.version>
@@ -405,6 +412,22 @@
       <scope>provided</scope>
     </dependency>
 
+    <!-- Iceberg Dependencies -->
+    <dependency>
+      <groupId>org.apache.iceberg</groupId>
+      <artifactId>iceberg-spark-runtime-${spark.major.version}_${scala.binary.version}</artifactId>
+      <version>${iceberg.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <!-- Deltalake Dependencies -->
+    <dependency>
+      <groupId>io.delta</groupId>
+      <artifactId>delta-spark_${scala.binary.version}</artifactId>
+      <version>${deltalake.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
     <!-- jakarta.validation -->
     <dependency>
       <groupId>org.hibernate.validator</groupId>

java/src/main/java/com/google/cloud/dataproc/templates/BaseTemplate.java

@@ -37,6 +37,7 @@ enum TemplateName {
     GCSTOBIGQUERY,
     GCSTOGCS,
     GCSTOMONGO,
+    GCSDELTALAKETOICEBERG,
     JDBCTOBIGQUERY,
     JDBCTOGCS,
     BIGQUERYTOGCS,

java/src/main/java/com/google/cloud/dataproc/templates/gcs/GCSDLtoIBConfig.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright (C) 2025 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.dataproc.templates.gcs;
+
+import static com.google.cloud.dataproc.templates.util.TemplateConstants.*;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.validation.constraints.AssertTrue;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.Pattern;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import org.apache.commons.lang3.StringUtils;
+
+public class GCSDLtoIBConfig {
+
+  static final ObjectMapper mapper =
+      new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+  @JsonProperty(value = PROJECT_ID_PROP)
+  @NotEmpty
+  private String projectId;
+
+  @JsonProperty(value = DELTALAKE_INPUT_LOCATION)
+  @NotEmpty
+  private String inputFileLocation;
+
+  @JsonProperty(value = DELTALAKE_VERSION_AS_OF)
+  @Min(value = 0)
+  private int versionAsOf;
+
+  @JsonProperty(value = DELTALAKE_TIMESTAMP_AS_OF)
+  private String timestampAsOf;
+
+  @JsonProperty(value = ICEBERG_TABLE_NAME)
+  @NotEmpty
+  private String icebergTableName;
+
+  @JsonProperty(value = ICEBERG_TABLE_PARTITION_COLUMNS)
+  private String icebergTablePartitionColumns;
+
+  @JsonProperty(value = ICEBERG_GCS_OUTPUT_MODE)
+  @Pattern(regexp = "(?i)(Append|Overwrite)")
+  private String icebergTableWriteMode;
+
+  @JsonProperty(value = ICEBERG_TABLE_MERGE_SCHEMA)
+  private String icebergTableMergeSchema;
+
+  @JsonProperty(value = SPARK_LOG_LEVEL)
+  @Pattern(regexp = "ALL|DEBUG|ERROR|FATAL|INFO|OFF|TRACE|WARN")
+  private String sparkLogLevel;
+
+  public @NotEmpty String getProjectId() {
+    return projectId;
+  }
+
+  public @NotEmpty String getInputFileLocation() {
+    return inputFileLocation;
+  }
+
+  public int getVersionAsOf() {
+    return versionAsOf;
+  }
+
+  public String getTimestampAsOf() {
+    return timestampAsOf;
+  }
+
+  public @NotEmpty String getIcebergTableName() {
+    return icebergTableName;
+  }
+
+  public List<String> getIcebergTablePartitionColumns() {
+
+    if (icebergTablePartitionColumns != null && !icebergTablePartitionColumns.isEmpty()) {
+      return new ArrayList<>(Arrays.asList(icebergTablePartitionColumns.split(",")));
+    }
+    return new ArrayList<>();
+  }
+
+  public String getIcebergTableWriteMode() {
+    return icebergTableWriteMode;
+  }
+
+  public String getIcebergTableMergeSchema() {
+    return icebergTableMergeSchema;
+  }
+
+  public @NotEmpty String getSparkLogLevel() {
+    return sparkLogLevel;
+  }
+
+  @Override
+  public String toString() {
+    return "GCSDLtoIBConfig{"
+        + "projectId='"
+        + projectId
+        + '\''
+        + ", inputFileLocation='"
+        + inputFileLocation
+        + '\''
+        + ", versionAsOf="
+        + versionAsOf
+        + ", timestampAsOf='"
+        + timestampAsOf
+        + '\''
+        + ", icebergTableName='"
+        + icebergTableName
+        + '\''
+        + ", icebergTablePartitionColumns='"
+        + icebergTablePartitionColumns
+        + '\''
+        + ", icebergTableWriteMode='"
+        + icebergTableWriteMode
+        + '\''
+        + ", icebergTableMergeSchema='"
+        + icebergTableMergeSchema
+        + '\''
+        + ", sparkLogLevel='"
+        + sparkLogLevel
+        + '\''
+        + '}';
+  }
+
+  public static GCSDLtoIBConfig fromProperties(Properties properties) {
+    return mapper.convertValue(properties, GCSDLtoIBConfig.class);
+  }
+
+  @AssertTrue(
+      message =
+          "Required parameters for GCSDeltalakeToIceberg not passed. Refer to gcs/README.md for more instructions.")
+  private boolean isInputValid() {
+    return StringUtils.isNotBlank(projectId)
+        && StringUtils.isNotBlank(inputFileLocation)
+        && StringUtils.isNotBlank(icebergTableName);
+  }
+}