-
Notifications
You must be signed in to change notification settings - Fork 2.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Delta: Support Snapshot Delta Lake Table to Iceberg Table (#6449)
Co-authored-by: ericlgoodman <erigood@amazon.com>
- Loading branch information
1 parent
10df74a
commit eeb055a
Showing
15 changed files
with
1,936 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,115 @@ | ||
| # | ||
| # Licensed to the Apache Software Foundation (ASF) under one | ||
| # or more contributor license agreements. See the NOTICE file | ||
| # distributed with this work for additional information | ||
| # regarding copyright ownership. The ASF licenses this file | ||
| # to you 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. | ||
| # | ||
|
|
||
| name: "Delta Conversion CI" | ||
| on: | ||
| push: | ||
| branches: | ||
| - 'master' | ||
| - '0.**' | ||
| tags: | ||
| - 'apache-iceberg-**' | ||
| pull_request: | ||
| paths-ignore: | ||
| - '.github/ISSUE_TEMPLATE/iceberg_bug_report.yml' | ||
| - '.github/workflows/python-ci.yml' | ||
| - '.github/workflows/flink-ci.yml' | ||
| - '.github/workflows/hive-ci.yml' | ||
| - '.gitignore' | ||
| - '.asf.yml' | ||
| - 'dev/**' | ||
| - 'mr/**' | ||
| - 'hive3/**' | ||
| - 'hive3-orc-bundle/**' | ||
| - 'hive-runtime/**' | ||
| - 'flink/**' | ||
| - 'pig/**' | ||
| - 'python/**' | ||
| - 'python_legacy/**' | ||
| - 'docs/**' | ||
| - 'open-api/**' | ||
| - 'format/**' | ||
| - '.gitattributes' | ||
| - 'README.md' | ||
| - 'CONTRIBUTING.md' | ||
| - 'LICENSE' | ||
| - 'NOTICE' | ||
|
|
||
| concurrency: | ||
| group: ${{ github.workflow }}-${{ github.ref }} | ||
| cancel-in-progress: ${{ github.event_name == 'pull_request' }} | ||
|
|
||
| jobs: | ||
| delta-conversion-scala-2-12-tests: | ||
| runs-on: ubuntu-20.04 | ||
| strategy: | ||
| matrix: | ||
| jvm: [8, 11] | ||
| env: | ||
| SPARK_LOCAL_IP: localhost | ||
| steps: | ||
| - uses: actions/checkout@v3 | ||
| - uses: actions/setup-java@v3 | ||
| with: | ||
| distribution: zulu | ||
| java-version: ${{ matrix.jvm }} | ||
| - uses: actions/cache@v3 | ||
| with: | ||
| path: | | ||
| ~/.gradle/caches | ||
| ~/.gradle/wrapper | ||
| key: ${{ runner.os }}-gradle-${{ hashFiles('**/*.gradle*', '**/gradle-wrapper.properties') }} | ||
| restore-keys: ${{ runner.os }}-gradle- | ||
| - run: echo -e "$(ip addr show eth0 | grep "inet\b" | awk '{print $2}' | cut -d/ -f1)\t$(hostname -f) $(hostname -s)" | sudo tee -a /etc/hosts | ||
| - run: ./gradlew -DsparkVersions=3.3 -DscalaVersion=2.12 -DhiveVersions= -DflinkVersions= :iceberg-delta-lake:check -Pquick=true -x javadoc | ||
| - uses: actions/upload-artifact@v3 | ||
| if: failure() | ||
| with: | ||
| name: test logs | ||
| path: | | ||
| **/build/testlogs | ||
| delta-conversion-scala-2-13-tests: | ||
| runs-on: ubuntu-20.04 | ||
| strategy: | ||
| matrix: | ||
| jvm: [ 8, 11 ] | ||
| env: | ||
| SPARK_LOCAL_IP: localhost | ||
| steps: | ||
| - uses: actions/checkout@v3 | ||
| - uses: actions/setup-java@v3 | ||
| with: | ||
| distribution: zulu | ||
| java-version: ${{ matrix.jvm }} | ||
| - uses: actions/cache@v3 | ||
| with: | ||
| path: | | ||
| ~/.gradle/caches | ||
| ~/.gradle/wrapper | ||
| key: ${{ runner.os }}-gradle-${{ hashFiles('**/*.gradle*', '**/gradle-wrapper.properties') }} | ||
| restore-keys: ${{ runner.os }}-gradle- | ||
| - run: echo -e "$(ip addr show eth0 | grep "inet\b" | awk '{print $2}' | cut -d/ -f1)\t$(hostname -f) $(hostname -s)" | sudo tee -a /etc/hosts | ||
| - run: ./gradlew -DsparkVersions=3.3 -DscalaVersion=2.13 -DhiveVersions= -DflinkVersions= :iceberg-delta-lake:check -Pquick=true -x javadoc | ||
| - uses: actions/upload-artifact@v3 | ||
| if: failure() | ||
| with: | ||
| name: test logs | ||
| path: | | ||
| **/build/testlogs |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
63 changes: 63 additions & 0 deletions
63
...ntegration/java/org/apache/iceberg/delta/DeltaLakeToIcebergMigrationSparkIntegration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you 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 org.apache.iceberg.delta; | ||
|
|
||
| import org.apache.iceberg.catalog.TableIdentifier; | ||
| import org.apache.iceberg.relocated.com.google.common.base.Preconditions; | ||
| import org.apache.iceberg.spark.Spark3Util; | ||
| import org.apache.spark.sql.SparkSession; | ||
| import org.apache.spark.sql.connector.catalog.CatalogPlugin; | ||
|
|
||
| /** An example class shows how to use the delta lake migration actions in SparkContext. */ | ||
| class DeltaLakeToIcebergMigrationSparkIntegration { | ||
|
|
||
| private DeltaLakeToIcebergMigrationSparkIntegration() {} | ||
|
|
||
| /** | ||
| * Example of how to use a {@link SparkSession}, a table identifier and a delta table location to | ||
| * construct an action for snapshotting the delta table to an iceberg table. | ||
| * | ||
| * @param spark a SparkSession with iceberg catalog configured. | ||
| * @param newTableIdentifier can be both 2 parts and 3 parts identifier, if it is 2 parts, the | ||
| * default spark catalog will be used | ||
| * @param deltaTableLocation the location of the delta table | ||
| * @return an instance of snapshot delta lake table action. | ||
| */ | ||
| static SnapshotDeltaLakeTable snapshotDeltaLakeTable( | ||
| SparkSession spark, String newTableIdentifier, String deltaTableLocation) { | ||
| Preconditions.checkArgument( | ||
| spark != null, "The SparkSession cannot be null, please provide a valid SparkSession"); | ||
| Preconditions.checkArgument( | ||
| newTableIdentifier != null, | ||
| "The table identifier cannot be null, please provide a valid table identifier for the new iceberg table"); | ||
| Preconditions.checkArgument( | ||
| deltaTableLocation != null, | ||
| "The delta lake table location cannot be null, please provide a valid location of the delta lake table to be snapshot"); | ||
|
|
||
| String ctx = "delta lake snapshot target"; | ||
| CatalogPlugin defaultCatalog = spark.sessionState().catalogManager().currentCatalog(); | ||
| Spark3Util.CatalogAndIdentifier catalogAndIdent = | ||
| Spark3Util.catalogAndIdentifier(ctx, spark, newTableIdentifier, defaultCatalog); | ||
| return DeltaLakeToIcebergMigrationActionsProvider.defaultActions() | ||
| .snapshotDeltaLakeTable(deltaTableLocation) | ||
| .as(TableIdentifier.parse(catalogAndIdent.identifier().toString())) | ||
| .deltaLakeConfiguration(spark.sessionState().newHadoopConf()) | ||
| .icebergCatalog(Spark3Util.loadIcebergCatalog(spark, catalogAndIdent.catalog().name())); | ||
| } | ||
| } |
73 changes: 73 additions & 0 deletions
73
delta-lake/src/integration/java/org/apache/iceberg/delta/SparkDeltaLakeSnapshotTestBase.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you 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 org.apache.iceberg.delta; | ||
|
|
||
| import java.util.Map; | ||
| import org.apache.hadoop.hive.conf.HiveConf; | ||
| import org.apache.iceberg.hive.TestHiveMetastore; | ||
| import org.apache.spark.sql.SparkSession; | ||
| import org.apache.spark.sql.internal.SQLConf; | ||
| import org.junit.AfterClass; | ||
| import org.junit.BeforeClass; | ||
|
|
||
| @SuppressWarnings("VisibilityModifier") | ||
| public abstract class SparkDeltaLakeSnapshotTestBase { | ||
| protected static TestHiveMetastore metastore = null; | ||
| protected static HiveConf hiveConf = null; | ||
| protected static SparkSession spark = null; | ||
|
|
||
| @BeforeClass | ||
| public static void startMetastoreAndSpark() { | ||
| SparkDeltaLakeSnapshotTestBase.metastore = new TestHiveMetastore(); | ||
| metastore.start(); | ||
| SparkDeltaLakeSnapshotTestBase.hiveConf = metastore.hiveConf(); | ||
|
|
||
| SparkDeltaLakeSnapshotTestBase.spark = | ||
| SparkSession.builder() | ||
| .master("local[2]") | ||
| .config(SQLConf.PARTITION_OVERWRITE_MODE().key(), "dynamic") | ||
| .config( | ||
| "spark.hadoop." + HiveConf.ConfVars.METASTOREURIS.varname, | ||
| hiveConf.get(HiveConf.ConfVars.METASTOREURIS.varname)) | ||
| .config("spark.sql.legacy.respectNullabilityInTextDatasetConversion", "true") | ||
| .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") | ||
| .enableHiveSupport() | ||
| .getOrCreate(); | ||
| } | ||
|
|
||
| @AfterClass | ||
| public static void stopMetastoreAndSpark() throws Exception { | ||
| if (metastore != null) { | ||
| metastore.stop(); | ||
| SparkDeltaLakeSnapshotTestBase.metastore = null; | ||
| } | ||
| if (spark != null) { | ||
| spark.stop(); | ||
| SparkDeltaLakeSnapshotTestBase.spark = null; | ||
| } | ||
| } | ||
|
|
||
| public SparkDeltaLakeSnapshotTestBase( | ||
| String catalogName, String implementation, Map<String, String> config) { | ||
|
|
||
| spark.conf().set("spark.sql.catalog." + catalogName, implementation); | ||
| config.forEach( | ||
| (key, value) -> spark.conf().set("spark.sql.catalog." + catalogName + "." + key, value)); | ||
| } | ||
| } |
Oops, something went wrong.