如何从按月分区的镶木地板文件中删除特定月份

Question

我有过去 5 年的 monthly 收入数据，并且我在 append 模式下以 parquet 格式存储各个月份的数据帧，但 partitioned by month 列。这是下面的伪代码 -

def Revenue(filename):
    df = spark.read.load(filename)
    .
    .
    df.write.format('parquet').mode('append').partitionBy('month').save('/path/Revenue')

Revenue('Revenue_201501.csv')
Revenue('Revenue_201502.csv')
Revenue('Revenue_201503.csv')
Revenue('Revenue_201504.csv')
Revenue('Revenue_201505.csv')

df 每月以parquet 格式存储，如下所示 -

问题：如何删除特定月份对应的parquet文件夹？

一种方法是将所有这些parquet 文件加载到一个大的df 中，然后使用.where() 子句过滤掉该特定月份，然后将其保存回parquet 格式partitionBy 月份@ 987654341@模式，像这样-

# If we want to remove data from Feb, 2015
df = spark.read.format('parquet').load('Revenue.parquet')
df = df.where(col('month') != lit('2015-02-01'))
df.write.format('parquet').mode('overwrite').partitionBy('month').save('/path/Revenue')

但是，这种方法相当麻烦。

另一种方法是直接删除该特定月份的文件夹，但我不确定这是否是正确的处理方式，以免我们以不可预见的方式更改metadata。

删除特定月份的parquet 数据的正确方法是什么？

Answer 1

Spark 支持删除分区，包括数据和元数据。
引用scala代码注释

/**
 * Drop Partition in ALTER TABLE: to drop a particular partition for a table.
 *
 * This removes the data and metadata for this partition.
 * The data is actually moved to the .Trash/Current directory if Trash is configured,
 * unless 'purge' is true, but the metadata is completely lost.
 * An error message will be issued if the partition does not exist, unless 'ifExists' is true.
 * Note: purge is always false when the target is a view.
 *
 * The syntax of this command is:
 * {{{
 *   ALTER TABLE table DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...] [PURGE];
 * }}}
 */

在您的情况下，没有后备表。 我们可以将数据帧注册为临时表并使用上述语法(temp table documentation)

通过 pyspark，我们可以使用link 中的语法运行 SQL 示例：

df = spark.read.format('parquet').load('Revenue.parquet'). registerTempTable("tmp")
spark.sql("ALTER TABLE tmp DROP IF EXISTS PARTITION (month='2015-02-01') PURGE")

Answer 2

以下语句只会删除与分区信息相关的元数据。

ALTER TABLE db.yourtable DROP IF EXISTS PARTITION(loaded_date="2019-08-22");

如果您还想删除数据，则需要将 hive 外部表的 tblproperties 设置为 False。它会将您的配置单元表设置为托管表。

alter table db.yourtable set TBLPROPERTIES('EXTERNAL'='FALSE');

您可以将其设置回外部表。

alter table db.yourtable set TBLPROPERTIES('EXTERNAL'='TRUE');

我尝试使用 spark session 设置给定属性，但遇到了一些问题。

 spark.sql("""alter table db.test_external set tblproperties ("EXTERNAL"="TRUE")""")
pyspark.sql.utils.AnalysisException: u"Cannot set or change the preserved property key: 'EXTERNAL';"

我确信一定有办法做到这一点。我最终使用了python。我在 pyspark 中定义了下面的函数，它完成了这项工作。

query=""" hive -e 'alter table db.yourtable set tblproperties ("EXTERNAL"="FALSE");ALTER TABLE db.yourtable DROP IF EXISTS PARTITION(loaded_date="2019-08-22");' """

def delete_partition():
        print("I am here")
        import subprocess
        import sys
        p=subprocess.Popen(query,shell=True,stderr=subprocess.PIPE)
        stdout,stderr = p.communicate()
        if p.returncode != 0:
            print stderr
            sys.exit(1) 

>>> delete_partition()

这将同时删除元数据和数据。 笔记。我已经使用 Hive ORC 外​​部分区表对此进行了测试，该表在加载日期进行分区

# Partition Information
# col_name              data_type               comment

loaded_date             string

更新： 基本上，您的数据位于名为

的子目录中的 hdfs 位置

/Revenue/month=2015-02-01
/Revenue/month=2015-03-01
/Revenue/month=2015-03-01

等等

def delete_partition(month_delete):
      print("I am here")
      hdfs_path="/some_hdfs_location/Revenue/month="
      final_path=hdfs_path+month_delete
      import subprocess
      subprocess.call(["hadoop", "fs", "-rm", "-r", final_path])
      print("got deleted")

delete_partition("2015-02-01")