如何为流数据自动更新 Hive 外部表元数据分区答案

【问题标题】：How to automatically update the Hive external table metadata partitions for streaming data如何为流数据自动更新 Hive 外部表元数据分区
【发布时间】：2022-02-14 04:06:45
【问题描述】：

我正在使用 pyspark 将 spark 流数据写入 hdfs 分区。请找到代码

  data = (spark.readStream.format("json").schema(fileSchema).load(inputDirectoryOfJsonFiles))

  output = (data.writeStream
   .format("parquet")
   .partitionBy("date")
   .option("compression", "none")
   .option("path" , "/user/hdfs/stream-test")
   .option("checkpointLocation", "/user/hdfs/stream-ckp")
   .outputMode("append")
   .start().awaitTermination())

将数据写入hdfs后，我正在创建hive外部分区表。

CREATE EXTERNAL TABLE test (id string,record string) 
PARTITIONED BY (`date` date) 
STORED AS PARQUET 
LOCATION '/user/hdfs/stream-test/'
TBLPROPERTIES ('discover.partitions' = 'true');

但是新创建的分区没有被 Hive Metastore 识别。我正在使用 msck 命令更新元存储。

msck repair table test sync partitions

现在对于流数据如何使用实时分区自动更新配置单元元存储的任务。

请提出解决此问题的方法。

【问题讨论】：

标签： apache-spark pyspark hive spark-streaming hive-partitions

【解决方案1】：

Spark 结构化流本身不支持此功能，但您可以使用 foreachBatch 作为解决方法

val yourStream = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .load()

val query = yourStream.writeStream.foreachBatch((batchDF: DataFrame, batchId: Long) => {
  batchDF
      .write
      .mode(SaveMode.Append)
      .insertInto("your_db.your_hive_table");
}).start()

query.awaitTermination()

【讨论】：