会员标签4.0

src/main/java/com/gic/spark/datasource/entity/DataSourceMongodb.java

@@ -53,7 +53,7 @@ public class DataSourceMongodb extends DataSourceEntity {
         enterpriseList.toArray(enterpriseIds);
         Dataset<Row> dataset = MongodbRddManager.getInstance().getDocumentsByColumn(database, collection, partitionColumn, enterpriseIds).toDF();
         if (!isTableCreated) {
-            SparkHiveUtil.createHivePartitionTable(dataset, partitionColumn, getSchema(), database + "_" + collection);
+            SparkHiveUtil.createHivePartitionTable(dataset, partitionColumn, getSchema(), database + "_" + collection,sparkSession);
             isTableCreated = true;
         }
         HivePartitionUtil.saveDatasetToPartitionTable(sparkSession, dataset, getHiveTableName());

src/main/java/com/gic/spark/datasource/entity/DataSourceMysql.java

@@ -51,7 +51,7 @@ public class DataSourceMysql extends DataSourceEntity {
         SparkSession sparkSession = SparkEnvManager.getInstance().getSparkSession();
         Dataset<Row> sourceDataset = datasource.buildRddManager().getDatasetByEnterpriseIds(targetTable, enterpriseList, null, null).repartition(new Column("enterprise_id"));
 
-        SparkHiveUtil.createHivePartitionTable(sourceDataset, "enterprise_id", getSchema(), datasource.getSchema().replaceAll("\\.", "_") + "_" + targetTable);
+        SparkHiveUtil.createHivePartitionTable(sourceDataset, "enterprise_id", getSchema(), datasource.getSchema().replaceAll("\\.", "_") + "_" + targetTable,sparkSession);
 
         HivePartitionUtil.saveDatasetToPartitionTable(sparkSession, sourceDataset, getHiveTableName());