feat: 添加mma.hive.only.schema配置，允许hive只传schema

aliyun · Apr 22, 2022 · e9bc293 · e9bc293
1 parent 17ff810
commit e9bc293
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 7 deletions.
diff --git a/mma-commons/src/main/java/com/aliyun/odps/mma/config/AbstractConfiguration.java b/mma-commons/src/main/java/com/aliyun/odps/mma/config/AbstractConfiguration.java
@@ -68,6 +68,7 @@ public abstract class AbstractConfiguration implements Map<String, String>, Conf
   // Hive
   public static final String METADATA_SOURCE_HIVE_IMPL = "mma.metadata.source.hive.impl";
   public static final String METADATA_SOURCE_HIVE_IMPL_DEFAULT_VALUE = "HMS";
+  public static final String MMA_HIVE_ONLY_SCHEMA = "mma.hive.only.schema";
   // Hive Meta store
   public static final String METADATA_SOURCE_HIVE_METASTORE_URIS = "mma.metadata.source.hive.metastore.uris";
   public static final String METADATA_SOURCE_HIVE_METASTORE_SASL_ENABLED = "mma.metadata.source.hive.metastore.sasl.enabled";

diff --git a/mma-server/src/main/java/com/aliyun/odps/mma/server/job/HiveToMcTableJob.java b/mma-server/src/main/java/com/aliyun/odps/mma/server/job/HiveToMcTableJob.java
@@ -62,6 +62,7 @@ DirectedAcyclicGraph<Task, DefaultEdge> generateDag() throws Exception {
       MetaSource metaSource = metaSourceFactory.getMetaSource(config);
       String catalogName = config.get(JobConfiguration.SOURCE_CATALOG_NAME);
       String tableName = config.get(JobConfiguration.SOURCE_OBJECT_NAME);
+      boolean onlySchema = config.getOrDefault(JobConfiguration.MMA_HIVE_ONLY_SCHEMA, "false").equals("true");
 
       TableMetaModel hiveTableMetaModel = metaSource.getTableMeta(catalogName, tableName);
       SchemaTransformResult schemaTransformResult = SchemaTransformerFactory
@@ -80,15 +81,20 @@ DirectedAcyclicGraph<Task, DefaultEdge> generateDag() throws Exception {
           hiveTableMetaModel,
           mcTableMetaModel,
           pendingSubJobs);
-      List<Task> dataTransmissionTasks = getDataTransmissionTasks(
-          metaSource,
-          hiveTableMetaModel,
-          mcTableMetaModel,
-          pendingSubJobs);
 
       dag.addVertex(setUpTask);
-      dataTransmissionTasks.forEach(dag::addVertex);
-      dataTransmissionTasks.forEach(t -> dag.addEdge(setUpTask, t));
+
+      if (!onlySchema) {
+        List<Task> dataTransmissionTasks = getDataTransmissionTasks(
+                metaSource,
+                hiveTableMetaModel,
+                mcTableMetaModel,
+                pendingSubJobs);
+
+        dataTransmissionTasks.forEach(dag::addVertex);
+        dataTransmissionTasks.forEach(t -> dag.addEdge(setUpTask, t));
+      }
+
       return dag;
     } catch (Exception e) {
       String stackTrace = ExceptionUtils.getStackTrace(e);