partitionBy在使用S3保存数据集时花费的时间太长

Question

我正在尝试使用partitionBy在S3上使用pyspark保存数据集。我在日期列上进行分区。星火作业需要超过一个小时才能执行。如果我在没有partitionBy的情况下运行代码，只需3-4个薄荷糖即可。有人能帮我调一下调子吗？

Answer 1

好吧，所以火花在做IO方面是很糟糕的。特别是在s3方面。目前，当您正在写入的火花，它将使用一个完整的执行器，以顺序写入数据。在s3和spark之间来回移动会导致它变得非常慢。因此，您可以做一些事情来帮助缓解/解决这些问题。

• 如果可能的话，使用不同的分区策略，目标是最小化编写的文件。
• 如果在写入之前涉及到洗牌，您可以在编写之前更改默认洗牌大小的设置：spark.sql.shuffle.partitions 200 // 200 is the default you'll probably want to reduce this和/或重新分区数据。
• 您可以绕过sparks，编写自己的hdfs编写器，或者直接使用s3 api。使用类似于前端分区的东西，然后使用一个函数来写入s3。那样的话，事情就会并行地写，而不是顺序地写。
• 最后，您可能希望在编写( partitionBy )时一起使用重新分区和DataFrame partitionBy to a single Parquet file (per partition)。当与上面的maxRecordsPerFile混合时，这将导致每个分区一个文件，这将有助于降低文件大小。

顺便提一句:您可以使用选项spark.sql.files.maxRecordsPerFile 1000000来帮助控制文件大小，以确保它们不会失控。

简而言之，您应该避免创建太多的文件，特别是小文件。还要注意:当你回去读那些2000*n的文件时，你会看到一个很大的性能冲击。

我们在不同的情况下使用上述所有的策略。但是一般来说，我们只是尝试在写之前使用合理的分区策略+重新分区。另一个注意事项:如果执行了洗牌，您的分区就会被破坏，并且火花自动分区将接管。因此，需要不断地重新分区。

希望这些建议能有所帮助。SparkIO是相当令人沮丧的，但是请记住将文件的读/写保持在最低限度，并且您应该看到良好的性能。

Answer 2

使用FileOutputCommiter的第2版

.set("mapreduce.fileoutputcommitter.algorithm.version", "2")