This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8652531f6 [spark] Support migrate database procedure for spark (#4226)
8652531f6 is described below
commit 8652531f64df919799ed8bd7730aefd2e0b4b86f
Author: xuzifu666 <[email protected]>
AuthorDate: Mon Sep 23 10:15:36 2024 +0800
[spark] Support migrate database procedure for spark (#4226)
---
docs/content/spark/procedures.md | 12 ++
.../flink/procedure/MigrateDatabaseProcedure.java | 37 +++--
.../flink/procedure/MigrateDatabaseProcedure.java | 23 +++-
.../org/apache/paimon/spark/SparkProcedures.java | 2 +
.../spark/procedure/MigrateDatabaseProcedure.java | 150 +++++++++++++++++++++
.../paimon/spark/utils/TableMigrationUtils.java | 22 +++
.../procedure/MigrateDatabaseProcedureTest.scala | 104 ++++++++++++++
7 files changed, 337 insertions(+), 13 deletions(-)
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index 2fd1bb620..9f6e061c1 100644
--- a/docs/content/spark/procedures.md
+++ b/docs/content/spark/procedures.md
@@ -130,6 +130,18 @@ This section introduce all available spark procedures
about paimon.
CALL sys.rollback(table => 'default.T', version => 10)
</td>
</tr>
+ <tr>
+ <td>migrate_database</td>
+ <td>
+ Migrate hive table to a paimon table. Arguments:
+ <li>source_type: the origin table's type to be migrated, such as
hive. Cannot be empty.</li>
+ <li>database: name of the origin database to be migrated. Cannot
be empty.</li>
+ <li>options: the table options of the paimon table to migrate.</li>
+ <li>options_map: Options map for adding key-value options which is
a map.</li>
+ <li>parallelism: the parallelism for migrate process, default is
core numbers of machine.</li>
+ </td>
+ <td>CALL sys.migrate_database(source_type => 'hive', database => 'db01',
options => 'file.format=parquet', options_map => map('k1','v1'), parallelism =>
6)</td>
+ </tr>
<tr>
<td>migrate_table</td>
<td>
diff --git
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
index b398b3258..a1ae35e32 100644
---
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
+++
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
@@ -23,12 +23,16 @@ import org.apache.paimon.migrate.Migrator;
import org.apache.paimon.utils.ParameterUtils;
import org.apache.flink.table.procedure.ProcedureContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.List;
/** Migrate procedure to migrate all hive tables in database to paimon table.
*/
public class MigrateDatabaseProcedure extends ProcedureBase {
+ private static final Logger LOG =
LoggerFactory.getLogger(MigrateDatabaseProcedure.class);
+
@Override
public String identifier() {
return "migrate_database";
@@ -54,12 +58,8 @@ public class MigrateDatabaseProcedure extends ProcedureBase {
Runtime.getRuntime().availableProcessors(),
ParameterUtils.parseCommaSeparatedKeyValues(properties));
- for (Migrator migrator : migrators) {
- migrator.executeMigrate();
- migrator.renameTable(false);
- }
-
- return new String[] {"Success"};
+ String retStr = handleMigrators(migrators);
+ return new String[] {retStr};
}
public String[] call(
@@ -78,11 +78,28 @@ public class MigrateDatabaseProcedure extends ProcedureBase
{
p,
ParameterUtils.parseCommaSeparatedKeyValues(properties));
+ String retStr = handleMigrators(migrators);
+ return new String[] {retStr};
+ }
+
+ public String handleMigrators(List<Migrator> migrators) {
+ int errorCount = 0;
+ int successCount = 0;
+
for (Migrator migrator : migrators) {
- migrator.executeMigrate();
- migrator.renameTable(false);
+ try {
+ migrator.executeMigrate();
+ migrator.renameTable(false);
+ successCount++;
+ } catch (Exception e) {
+ errorCount++;
+ LOG.error("Call migrate_database error:" + e.getMessage());
+ }
}
-
- return new String[] {"Success"};
+ String retStr =
+ String.format(
+ "migrate database is finished, success cnt: %s ,
failed cnt: %s",
+ String.valueOf(successCount),
String.valueOf(errorCount));
+ return retStr;
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
index 936d9069e..8c6e8730b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/MigrateDatabaseProcedure.java
@@ -26,12 +26,16 @@ import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.ProcedureHint;
import org.apache.flink.table.procedure.ProcedureContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.List;
/** Migrate procedure to migrate all hive tables in database to paimon table.
*/
public class MigrateDatabaseProcedure extends ProcedureBase {
+ private static final Logger LOG =
LoggerFactory.getLogger(MigrateDatabaseProcedure.class);
+
@Override
public String identifier() {
return "migrate_database";
@@ -64,11 +68,24 @@ public class MigrateDatabaseProcedure extends ProcedureBase
{
p,
ParameterUtils.parseCommaSeparatedKeyValues(properties));
+ int errorCount = 0;
+ int successCount = 0;
+
for (Migrator migrator : migrators) {
- migrator.executeMigrate();
- migrator.renameTable(false);
+ try {
+ migrator.executeMigrate();
+ migrator.renameTable(false);
+ successCount++;
+ } catch (Exception e) {
+ errorCount++;
+ LOG.error("Call migrate_database error:" + e.getMessage());
+ }
}
+ String retStr =
+ String.format(
+ "migrate database is finished, success cnt: %s ,
failed cnt: %s",
+ String.valueOf(successCount),
String.valueOf(errorCount));
- return new String[] {"Success"};
+ return new String[] {retStr};
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index 0a9076247..b8e43c19c 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -28,6 +28,7 @@ import
org.apache.paimon.spark.procedure.ExpirePartitionsProcedure;
import org.apache.paimon.spark.procedure.ExpireSnapshotsProcedure;
import org.apache.paimon.spark.procedure.FastForwardProcedure;
import org.apache.paimon.spark.procedure.MarkPartitionDoneProcedure;
+import org.apache.paimon.spark.procedure.MigrateDatabaseProcedure;
import org.apache.paimon.spark.procedure.MigrateFileProcedure;
import org.apache.paimon.spark.procedure.MigrateTableProcedure;
import org.apache.paimon.spark.procedure.Procedure;
@@ -66,6 +67,7 @@ public class SparkProcedures {
procedureBuilders.put("create_branch", CreateBranchProcedure::builder);
procedureBuilders.put("delete_branch", DeleteBranchProcedure::builder);
procedureBuilders.put("compact", CompactProcedure::builder);
+ procedureBuilders.put("migrate_database",
MigrateDatabaseProcedure::builder);
procedureBuilders.put("migrate_table", MigrateTableProcedure::builder);
procedureBuilders.put("migrate_file", MigrateFileProcedure::builder);
procedureBuilders.put("remove_orphan_files",
RemoveOrphanFilesProcedure::builder);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateDatabaseProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateDatabaseProcedure.java
new file mode 100644
index 000000000..db0c7cebb
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/MigrateDatabaseProcedure.java
@@ -0,0 +1,150 @@
+/*
+ * 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.paimon.spark.procedure;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.migrate.Migrator;
+import org.apache.paimon.spark.catalog.WithPaimonCatalog;
+import org.apache.paimon.spark.utils.TableMigrationUtils;
+import org.apache.paimon.utils.ParameterUtils;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Migrate database procedure. Usage:
+ *
+ * <pre><code>
+ * CALL sys.migrate_database(source_type => 'hive', database => 'db01',
options => 'x1=y1,x2=y2')
+ * </code></pre>
+ */
+public class MigrateDatabaseProcedure extends BaseProcedure {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MigrateDatabaseProcedure.class);
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("source_type", StringType),
+ ProcedureParameter.required("database", StringType),
+ ProcedureParameter.optional("options", StringType),
+ ProcedureParameter.optional(
+ "options_map", DataTypes.createMapType(StringType,
StringType)),
+ ProcedureParameter.optional("parallelism", IntegerType),
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", StringType, true,
Metadata.empty())
+ });
+
+ protected MigrateDatabaseProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ String format = args.getString(0);
+ String database = args.getString(1);
+ String properties = args.isNullAt(2) ? null : args.getString(2);
+ MapData mapData = args.isNullAt(3) ? null : args.getMap(3);
+ Map<String, String> optionMap = mapDataToHashMap(mapData);
+ int parallelism =
+ args.isNullAt(4) ? Runtime.getRuntime().availableProcessors()
: args.getInt(4);
+
+ Catalog paimonCatalog = ((WithPaimonCatalog)
tableCatalog()).paimonCatalog();
+
+ Map<String, String> options =
ParameterUtils.parseCommaSeparatedKeyValues(properties);
+ options.putAll(optionMap);
+
+ List<Migrator> migrators =
+ TableMigrationUtils.getImporters(
+ format, paimonCatalog, database, parallelism, options);
+
+ int errorCount = 0;
+ int successCount = 0;
+
+ for (Migrator migrator : migrators) {
+ try {
+ migrator.executeMigrate();
+ migrator.renameTable(false);
+ successCount++;
+ } catch (Exception e) {
+ errorCount++;
+ LOG.error("Call migrate_database error:" + e.getMessage());
+ }
+ }
+ String retStr =
+ String.format(
+ "migrate database is finished, success cnt: %s ,
failed cnt: %s",
+ String.valueOf(successCount),
String.valueOf(errorCount));
+
+ return new InternalRow[] {newInternalRow(retStr)};
+ }
+
+ public static Map<String, String> mapDataToHashMap(MapData mapData) {
+ HashMap<String, String> map = new HashMap<>();
+ if (mapData != null) {
+ for (int index = 0; index < mapData.numElements(); index++) {
+ map.put(
+ mapData.keyArray().getUTF8String(index).toString(),
+ mapData.valueArray().getUTF8String(index).toString());
+ }
+ }
+ return map;
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<MigrateDatabaseProcedure>() {
+ @Override
+ public MigrateDatabaseProcedure doBuild() {
+ return new MigrateDatabaseProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "MigrateDatabaseProcedure";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
index 73702bf35..3499a5e40 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/utils/TableMigrationUtils.java
@@ -24,6 +24,7 @@ import org.apache.paimon.hive.HiveCatalog;
import org.apache.paimon.hive.migrate.HiveMigrator;
import org.apache.paimon.migrate.Migrator;
+import java.util.List;
import java.util.Map;
/** Migration util to choose importer according to connector. */
@@ -58,4 +59,25 @@ public class TableMigrationUtils {
throw new UnsupportedOperationException("Unsupported connector
" + connector);
}
}
+
+ public static List<Migrator> getImporters(
+ String connector,
+ Catalog catalog,
+ String sourceDatabase,
+ Integer parallelism,
+ Map<String, String> options) {
+ switch (connector) {
+ case "hive":
+ if (catalog instanceof CachingCatalog) {
+ catalog = ((CachingCatalog) catalog).wrapped();
+ }
+ if (!(catalog instanceof HiveCatalog)) {
+ throw new IllegalArgumentException("Only support Hive
Catalog");
+ }
+ return HiveMigrator.databaseMigrators(
+ (HiveCatalog) catalog, sourceDatabase, options,
parallelism);
+ default:
+ throw new UnsupportedOperationException("Don't support
connector " + connector);
+ }
+ }
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateDatabaseProcedureTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateDatabaseProcedureTest.scala
new file mode 100644
index 000000000..332a52d36
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/MigrateDatabaseProcedureTest.scala
@@ -0,0 +1,104 @@
+/*
+ * 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.paimon.spark.procedure
+
+import org.apache.paimon.spark.PaimonHiveTestBase
+
+import org.apache.spark.sql.Row
+class MigrateDatabaseProcedureTest extends PaimonHiveTestBase {
+ Seq("parquet", "orc", "avro").foreach(
+ format => {
+ test(s"Paimon migrate database procedure: migrate $format
non-partitioned database") {
+ withTable("hive_tbl", "hive_tbl1") {
+ // create hive table
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl (id STRING, name STRING, pt
STRING)
+ |USING $format
+ |""".stripMargin)
+
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl1 (id STRING, name STRING, pt
STRING)
+ |USING $format
+ |""".stripMargin)
+
+ var rows0 = spark.sql("SHOW CREATE TABLE hive_tbl").collect()
+ assert(!rows0.apply(0).toString().contains("USING paimon"))
+
+ rows0 = spark.sql("SHOW CREATE TABLE hive_tbl1").collect()
+ assert(!rows0.apply(0).toString().contains("USING paimon"))
+
+ spark.sql(s"INSERT INTO hive_tbl VALUES ('1', 'a', 'p1'), ('2', 'b',
'p2')")
+
+ spark.sql(
+ s"CALL sys.migrate_database(source_type => 'hive', database =>
'$hiveDbName', options => 'file.format=$format')")
+
+ checkAnswer(
+ spark.sql(s"SELECT * FROM hive_tbl ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Nil)
+
+ var rows1 = spark.sql("SHOW CREATE TABLE hive_tbl").collect()
+ assert(rows1.apply(0).toString().contains("USING paimon"))
+
+ rows1 = spark.sql("SHOW CREATE TABLE hive_tbl1").collect()
+ assert(rows1.apply(0).toString().contains("USING paimon"))
+
+ }
+ }
+ })
+
+ Seq("parquet", "orc", "avro").foreach(
+ format => {
+ test(
+ s"Paimon migrate database procedure: migrate $format database with
setting parallelism") {
+ withTable("hive_tbl_01", "hive_tbl_02") {
+ // create hive table
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl_01 (id STRING, name STRING, pt
STRING)
+ |USING $format
+ |""".stripMargin)
+
+ spark.sql(s"""
+ |CREATE TABLE hive_tbl_02 (id STRING, name STRING, pt
STRING)
+ |USING $format
+ |""".stripMargin)
+
+ var rows0 = spark.sql("SHOW CREATE TABLE hive_tbl_01").collect()
+ assert(!rows0.apply(0).toString().contains("USING paimon"))
+
+ rows0 = spark.sql("SHOW CREATE TABLE hive_tbl_02").collect()
+ assert(!rows0.apply(0).toString().contains("USING paimon"))
+
+ spark.sql(s"INSERT INTO hive_tbl_01 VALUES ('1', 'a', 'p1'), ('2',
'b', 'p2')")
+
+ spark.sql(
+ s"CALL sys.migrate_database(source_type => 'hive', database =>
'$hiveDbName', options => 'file.format=$format', parallelism => 6)")
+
+ checkAnswer(
+ spark.sql(s"SELECT * FROM hive_tbl_01 ORDER BY id"),
+ Row("1", "a", "p1") :: Row("2", "b", "p2") :: Nil)
+
+ var rows1 = spark.sql("SHOW CREATE TABLE hive_tbl_01").collect()
+ assert(rows1.apply(0).toString().contains("USING paimon"))
+
+ rows1 = spark.sql("SHOW CREATE TABLE hive_tbl_02").collect()
+ assert(rows1.apply(0).toString().contains("USING paimon"))
+ }
+ }
+ })
+}
