You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/03/13 03:40:10 UTC
(incubator-paimon) branch master updated: [flink] Remove MigrateDatabaseProcedure's dependency on Hive (#2997)
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/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f7a4c819a [flink] Remove MigrateDatabaseProcedure's dependency on Hive (#2997)
f7a4c819a is described below
commit f7a4c819ac4d23a3ce7df81c361b31344db9327c
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Wed Mar 13 11:40:06 2024 +0800
[flink] Remove MigrateDatabaseProcedure's dependency on Hive (#2997)
---
.../java/org/apache/paimon/migrate/Migrator.java | 2 +
paimon-flink/paimon-flink-common/pom.xml | 54 ----------------------
.../flink/procedure/MigrateDatabaseProcedure.java | 38 +++++----------
.../paimon/flink/utils/TableMigrationUtils.java | 14 ++++++
.../apache/paimon/hive/migrate/HiveMigrator.java | 36 +++++++++++++++
5 files changed, 63 insertions(+), 81 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java b/paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java
index 7420528e4..b09a9ba38 100644
--- a/paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/migrate/Migrator.java
@@ -22,4 +22,6 @@ package org.apache.paimon.migrate;
public interface Migrator {
void executeMigrate() throws Exception;
+
+ void renameTable(boolean ignoreIfNotExists) throws Exception;
}
diff --git a/paimon-flink/paimon-flink-common/pom.xml b/paimon-flink/paimon-flink-common/pom.xml
index 74c53a85b..c4bc2a1d1 100644
--- a/paimon-flink/paimon-flink-common/pom.xml
+++ b/paimon-flink/paimon-flink-common/pom.xml
@@ -35,7 +35,6 @@ under the License.
<properties>
<flink.version>1.18.1</flink.version>
- <hive.version>2.3.9</hive.version>
</properties>
<dependencies>
@@ -67,59 +66,6 @@ under the License.
<scope>provided</scope>
</dependency>
- <dependency>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-metastore</artifactId>
- <version>${hive.version}</version>
- <scope>provided</scope>
- <exclusions>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-annotations</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-databind</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.orc</groupId>
- <artifactId>orc-core</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.pentaho</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hive</groupId>
- <artifactId>hive-storage-api</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.avro</groupId>
- <artifactId>avro</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-hdfs</artifactId>
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 4f2a849c9..166544eb7 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
@@ -18,24 +18,18 @@
package org.apache.paimon.flink.procedure;
-import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.utils.TableMigrationUtils;
import org.apache.paimon.hive.HiveCatalog;
+import org.apache.paimon.migrate.Migrator;
import org.apache.paimon.utils.ParameterUtils;
import org.apache.flink.table.procedure.ProcedureContext;
-import org.apache.hadoop.hive.metastore.IMetaStoreClient;
-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);
- private static final String PAIMON_SUFFIX = "_paimon_";
-
@Override
public String identifier() {
return "migrate_database";
@@ -57,28 +51,18 @@ public class MigrateDatabaseProcedure extends ProcedureBase {
throw new IllegalArgumentException("Only support Hive Catalog");
}
HiveCatalog hiveCatalog = (HiveCatalog) this.catalog;
- IMetaStoreClient client = hiveCatalog.getHmsClient();
- List<String> sourceTables = client.getAllTables(sourceDatabasePath);
- for (String sourceTable : sourceTables) {
- String sourceTablePath = sourceDatabasePath + "." + sourceTable;
- String targetPaimonTablePath = sourceTablePath + PAIMON_SUFFIX;
-
- Identifier sourceTableId = Identifier.fromString(sourceTablePath);
- Identifier targetTableId = Identifier.fromString(targetPaimonTablePath);
+ List<Migrator> migrators =
+ TableMigrationUtils.getImporters(
+ connector,
+ hiveCatalog,
+ sourceDatabasePath,
+ ParameterUtils.parseCommaSeparatedKeyValues(properties));
- TableMigrationUtils.getImporter(
- connector,
- (HiveCatalog) this.catalog,
- sourceTableId.getDatabaseName(),
- sourceTableId.getObjectName(),
- targetTableId.getDatabaseName(),
- targetTableId.getObjectName(),
- ParameterUtils.parseCommaSeparatedKeyValues(properties))
- .executeMigrate();
-
- LOG.info("rename " + targetTableId + " to " + sourceTableId);
- this.catalog.renameTable(targetTableId, sourceTableId, false);
+ for (Migrator migrator : migrators) {
+ migrator.executeMigrate();
+ migrator.renameTable(false);
}
+
return new String[] {"Success"};
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableMigrationUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableMigrationUtils.java
index 32f263ab3..32d017be6 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableMigrationUtils.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableMigrationUtils.java
@@ -22,6 +22,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. */
@@ -48,4 +49,17 @@ public class TableMigrationUtils {
throw new UnsupportedOperationException("Don't support connector " + connector);
}
}
+
+ public static List<Migrator> getImporters(
+ String connector,
+ HiveCatalog paimonCatalog,
+ String sourceDatabase,
+ Map<String, String> options) {
+ switch (connector) {
+ case "hive":
+ return HiveMigrator.databaseMigrators(paimonCatalog, sourceDatabase, options);
+ default:
+ throw new UnsupportedOperationException("Don't support connector " + connector);
+ }
+ }
}
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
index 642236cfc..fd31ec3ec 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
@@ -42,6 +42,9 @@ import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.PrimaryKeysRequest;
import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Comparator;
@@ -60,9 +63,13 @@ import static org.apache.paimon.utils.FileUtils.COMMON_IO_FORK_JOIN_POOL;
/** Migrate hive table to paimon table. */
public class HiveMigrator implements Migrator {
+ private static final Logger LOG = LoggerFactory.getLogger(HiveMigrator.class);
+
private static final Predicate<FileStatus> HIDDEN_PATH_FILTER =
p -> !p.getPath().getName().startsWith("_") && !p.getPath().getName().startsWith(".");
+ private static final String PAIMON_SUFFIX = "_paimon_";
+
private final FileIO fileIO;
private final HiveCatalog hiveCatalog;
private final IMetaStoreClient client;
@@ -89,6 +96,27 @@ public class HiveMigrator implements Migrator {
this.options = options;
}
+ public static List<Migrator> databaseMigrators(
+ HiveCatalog hiveCatalog, String sourceDatabase, Map<String, String> options) {
+ IMetaStoreClient client = hiveCatalog.getHmsClient();
+ try {
+ return client.getAllTables(sourceDatabase).stream()
+ .map(
+ sourceTable ->
+ new HiveMigrator(
+ hiveCatalog,
+ sourceDatabase,
+ sourceTable,
+ sourceDatabase,
+ sourceTable + PAIMON_SUFFIX,
+ options))
+ .collect(Collectors.toList());
+ } catch (TException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
public void executeMigrate() throws Exception {
if (!client.tableExists(sourceDatabase, sourceTable)) {
throw new RuntimeException("Source hive table does not exist");
@@ -178,6 +206,14 @@ public class HiveMigrator implements Migrator {
client.dropTable(sourceDatabase, sourceTable, true, true);
}
+ @Override
+ public void renameTable(boolean ignoreIfNotExists) throws Exception {
+ Identifier targetTableId = Identifier.create(targetDatabase, targetTable);
+ Identifier sourceTableId = Identifier.create(sourceDatabase, sourceTable);
+ LOG.info("Last step: rename {} to {}.", targetTableId, sourceTableId);
+ hiveCatalog.renameTable(targetTableId, sourceTableId, ignoreIfNotExists);
+ }
+
private void checkPrimaryKey() throws Exception {
PrimaryKeysRequest primaryKeysRequest = new PrimaryKeysRequest(sourceDatabase, sourceTable);
if (!client.getPrimaryKeys(primaryKeysRequest).isEmpty()) {