获取Spark写入Hive metastore的所有新分区

Question

我有一个数据帧，我用它来插入到现有的分区配置单元表中，使用spark sql（使用动态分区）。一旦数据帧被写入，我想知道我的数据帧刚刚在蜂巢中创建的分区是什么。

我可以在数据帧中查询不同的分区，但这需要很长时间，因为它必须启动数据帧的整个进程。

我可以在写入配置单元之前持久化数据帧，这样，写操作和distinct partition_column操作就发生在缓存的数据帧之上。但我的数据帧非常大，不想花更多的时间来坚持。

我知道所有的分区信息都存储在Hive元存储中。Spark 中是否有任何元存储 API 可以帮助仅检索已创建的新分区？

Answer 1

val epochTime = <epoch time before inserting the dataframe>
val partitionName = <Partition Column Name>
df.write.insertInto("<db_name>.<tbl_name>")
val catalogPartitions = spark.sharedState.externalCatalog.listPartitions("<db_name>", "<tbl_name>")
val partitionValues = catalogPartitions.filter(cp => ((cp.parameters.get("transient_lastDdlTime").isDefined && cp.parameters.getOrElse("transient_lastDdlTime", "0").toLong >= epochTime / 1000) || cp.lastAccessTime >= epochTime || cp.createTime >= epochTime) && cp.spec.contains(datePartition)).map(cp => cp.spec.getOrElse(datePartition, "")
      }).toList
processedPartitions = partitionValues.toList   

在大多数情况下，lastAccessTime作为0出现。createTime是创建分区的时间。但是在参数中，我发现了一个新参数transient_lastDdlTime，其中包含分区的更新时间戳。从更安全的角度来看，检查这三个分区，给我在给定纪元时间后创建或修改的分区。

Answer 2

您可以使用HiveMetastoreClient检索表的分区数据：

import org.apache.hadoop.hive.conf.HiveConf
import scala.collection.JavaConverters._
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient

val hiveConf = new HiveConf(spark.sparkContext.hadoopConfiguration, classOf[HiveConf])
val cli = new HiveMetaStoreClient(hiveConf)

/* Get list of partition values prior to DF insert */
val existingPartitions = cli.listPartitions("<db_name>", "<tbl_name>", Short.MaxValue).asScala.map(_.getValues.asScala.mkString(","))
/* Insert DF contents to table */
df.write.insertInto("<db_name>.<tbl_name>")
/* Fetch list of partition values again, and diff with previous list */
val newPartitions = cli.listPartitions("<db_name>", "<tbl_name>", Short.MaxValue).asScala.map(_.getValues.asScala.mkString(","))
val deltaPartitions = newPartitions.diff(existingPartitions)