为什么在使用saveAsTextFile时，在Google Dataproc中运行的Spark将临时文件存储在外部存储（GCS）而不是本地磁盘或HDFS上？

Question

我已经运行下面的代码PySpark：

from pyspark import SparkContext 

sc = SparkContext() 

data = sc.textFile('gs://bucket-name/input_blob_path') 
sorted_data = data.sortBy(lambda x: sort_criteria(x)) 
sorted_data.saveAsTextFile(
    'gs://bucket-name/output_blob_path', 
    compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec" 
) 

工作顺利完成。但是，在作业执行期间，Spark在以下路径gs://bucket-name/output_blob_path/_temporary/0/中创建了许多临时Blob。我意识到，在这段时间内，除去所有这些临时blob占用了一半的作业执行时间，并且在此期间CPU利用率为1％（巨大的资源浪费）。

有没有办法将临时文件存储在本地驱动器（或HDFS）而不是GCP？我仍然希望将最终结果（排序后的数据集）保存到GCP。

我们使用Dataproc Spark集群（VM类型16核，60GM）和10个工作节点。输入数据量为10TB。

Answer 1

您看到的_temporary文件很可能是在引擎盖下使用的FileOutputCommitter的工件。重要的是，这些临时数据块并非严格意义上的“临时”数据，而是实际完成的输出数据，只有在工作完成时才会被重命名为最终目的地。通过重命名这些文件的“提交”实际上很快，因为源和目的地都位于GCS上;因此，无法将临时文件放置在HDFS上，然后“提交”到GCS中，因此无法替换工作流程的这一部分，因为此时提交需要将整个输出数据集从HDFS重新导回到GCS中。具体而言，底层的Hadoop FileOutputFormat类不支持这种习惯用法。

GCS本身并不是一个真正的文件系统，而是一个“对象存储”，而Dataproc内部的GCS连接器只能尽其所能地模仿HDFS。其中一个结果是，删除文件的目录填充实际上需要GCS删除引擎盖下的单个对象，而不是真正的文件系统只是解除连接inode。

实际上，如果您碰到这个问题，可能意味着您的输出无论如何都会被分割成太多的文件，因为一次批量清理会出现大约1000个文件。因此，成千上万的输出文件通常不会明显变慢。拥有太多文件也会使这些文件的未来工作变得更慢。最简单的修复方法通常只是尽可能减少输出文件的数量，例如使用repartition()：

from pyspark import SparkContext 

sc = SparkContext() 

data = sc.textFile('gs://bucket-name/input_blob_path') 
sorted_data = data.sortBy(lambda x: sort_criteria(x)) 
sorted_data.repartition(1000).saveAsTextFile(
    'gs://bucket-name/output_blob_path', 
    compressionCodecClass="org.apache.hadoop.io.compress.GzipCodec" 
)