使用 Spark 删除 Azure blob 中文件的更快方法是什么？答案

【问题标题】：What is a faster way to delete files in an Azure blob with Spark?使用 Spark 删除 Azure blob 中文件的更快方法是什么？
【发布时间】：2019-10-17 21:51:19
【问题描述】：

我在 Azure 上使用 Databricks/Spark。我想删除存储在 blob 中的超过 100,000 个文本文件。 blob 已安装。我使用 Python (PySpark) 以并行方式删除这些文件的代码如下。

import os

# use dbutils to list all files
# very slow operation
text_files = ['/dbfs/mnt/myblob/myfolder/{}'.format(fi.name) for fi in dbutils.fs.ls('/dbfs/mnt/myblob/myfolder')]

# use spark context to parallelize list of text files
rdd = sc.parallelize(text_files)

# now delete files
# seems to be very slow as well when spark job kicks off
rdd.map(lambda p: os.remove(p)).count()

请注意，我使用dbutils 列出挂载目录中的文件。这个操作很慢。另请注意，我使用 Python 的os.remove(...) 删除文件，这个操作似乎也很慢。

关于在 Spark/Databricks 上下文中更快地删除 Azure blob 中的文件的任何提示？

还有相关的帖子here和here。但是，前者没有指定上下文/环境（引用的链接是针对 Scala 的，这些方法在 PySpark 中的 Spark 上下文中不存在），后者想要实现并行删除但没有说明解决方案是否需要与火花。

【问题讨论】：

使用 spark 这样做可能不是最好的解决方案。只需在您的文件夹上添加dbutils.rm。

标签： azure apache-spark pyspark azure-blob-storage azure-databricks

【解决方案1】：

目前，您正在从分布式文件系统中读取数据，然后将它们导入 python 驱动程序，将它们导出到 Spark，然后使用 Python 再次读取它们以执行删除...

我认为您可以删除 spark 步骤，在 python 中操作您的列表并直接使用 dbutils 删除文件。

【讨论】：

但这不是并行操作。
只需与一些为此而设计的库并行执行...但不是 spark。 Spark 应该处理数据，所以你只是在转移它的功能。您甚至不确定您的数据是否在集群中分布良好，也许您没有任何分区。

【解决方案2】：

我意识到没有直接回答您的问题，但看到您正在使用 Azure 并且可能会使用数据工厂（或者可能会在未来编排/生产您的数据块工作时）......

您可能希望考虑数据工厂删除活动，它是数据工厂 v2 的一个相当新的补充：https://docs.microsoft.com/en-us/azure/data-factory/delete-activity

它具有并行删除的能力。

【讨论】：

我会提倡这种方法。 Spark 使用驱动节点上的单核来列出文件夹中的文件——这通常也是导入数据的问题。我不确定 dbutils 是否可以使用并行线程来删除 - 但我怀疑它不会。
不可接受，除非我们可以以编程方式指定要删除的数据工厂。这可能吗？
嗨，简，是的，这是可能的。还有一个用于构建数据工厂的编程 c# API：docs.microsoft.com/en-us/azure/data-factory/…

【解决方案3】：

我在 Azure Blob Gen2 存储上使用了以下方法 (scala)：

subMask 包含由partitionBy 在 parquet 文件持久化期间创建的结构。

mssparkutils.fs.rm(x, true) 递归删除目录。

    val rootPath = "abfss://**@**.dfs.core.windows.net/parquet/poc-out/"
    val subMask = s"/year=$year/month=$month/day=$dayOfMonth/"    

    mssparkutils.fs.ls(rootPath).filter(_.isDir).foreach( r => 
        {
            val sd = s"${r.path}$subMask"
            val pathExists = mssparkutils.fs.exists(sd)
            if(pathExists){
                mssparkutils.fs.rm(sd, true) //remove path + files
            }
        }
    )

【讨论】：