根据Learning Spark
请记住,重新划分数据是一项相当昂贵的操作。 Spark还有一个repartition()的优化版本,称为coalesce(),它允许避免数据移动,但仅当您正在减少RDD分区的数量时。
我得到的一个区别是,使用repartition()可以增加/减少分区的数量,但使用coalesce()只能减少分区的数量。
如果分区分布在多台机器上,并且运行了coalesce(),它如何避免数据移动?
根据Learning Spark
请记住,重新划分数据是一项相当昂贵的操作。 Spark还有一个repartition()的优化版本,称为coalesce(),它允许避免数据移动,但仅当您正在减少RDD分区的数量时。
我得到的一个区别是,使用repartition()可以增加/减少分区的数量,但使用coalesce()只能减少分区的数量。
如果分区分布在多台机器上,并且运行了coalesce(),它如何避免数据移动?
当前回答
所有的答案都为这个经常被问到的问题增添了一些伟大的知识。
所以根据这个问题的传统时间轴,这里是我的2美分。
我发现在非常具体的情况下,重新分区比合并更快。
在我的应用程序中,当我们估计的文件数量低于某个阈值时,重新分区工作得更快。
这就是我的意思
if(numFiles > 20)
df.coalesce(numFiles).write.mode(SaveMode.Overwrite).parquet(dest)
else
df.repartition(numFiles).write.mode(SaveMode.Overwrite).parquet(dest)
在上面的代码片段中,如果我的文件小于20,合并将永远无法完成,而重新分区要快得多,因此上面的代码。
当然,这个数字(20)将取决于工作人员的数量和数据量。
希望这能有所帮助。
其他回答
对于所有这些伟大的答案,我想补充的是,重新分区是利用数据并行化的最佳选择之一。而coalesce提供了一个廉价的选择来减少分区,并且在将数据写入HDFS或其他接收器以利用大写入时非常有用。
我发现这在以拼花格式写数据时很有用,可以充分利用它。
重新分区-建议在增加分区数量的同时使用它,因为它涉及到所有数据的洗牌。
Coalesce—建议在使用它的同时减少分区的数量。例如,如果你有3个分区,你想把它减少到2个,coalesce将把第3个分区的数据移动到分区1和分区2。分区1和分区2将保留在同一个容器中。 另一方面,重新分区将打乱所有分区中的数据,因此执行程序之间的网络使用将很高,这将影响性能。
在减少分区数量的同时,Coalesce比重分区的性能更好。
对于那些从PySpark (AWS EMR)生成单个csv文件并将其保存在s3上的问题,使用重新分区会有所帮助。原因是,合并不能进行完全洗牌,但重新分区可以。从本质上讲,您可以使用重分区增加或减少分区的数量,但使用合并只能减少分区的数量(而不是1)。以下是为试图从AWS EMR写入csv到s3的任何人编写的代码:
df.repartition(1).write.format('csv')\
.option("path", "s3a://my.bucket.name/location")\
.save(header = 'true')
基本上,重分区允许您增加或减少分区的数量。重分区重新分配来自所有分区的数据,这导致完全shuffle,这是非常昂贵的操作。
Coalesce是重新分区的优化版本,您只能减少分区的数量。由于我们只能减少分区的数量,它所做的是将一些分区合并为一个分区。通过合并分区,与重新分区相比,跨分区的数据移动更低。所以在Coalesce中是最小的数据移动,但说Coalesce不做数据移动是完全错误的说法。
另一件事是通过提供分区的数量来重新分区,它试图在所有分区上均匀地重新分配数据而在Coalesce的情况下,在某些情况下我们仍然可能有倾斜的数据。
有一个重分区>>合并的用例,即使在@Rob的回答中提到的分区号减少,也就是将数据写入单个文件。
@Rob的回答暗示了一个好的方向,但我认为需要一些进一步的解释来理解引擎盖下面发生了什么。
如果您需要在写入数据之前过滤数据,那么重新分区比coalesce更适合,因为coalesce将在加载操作之前下推。
例如: load () . map(…).filter(…).coalesce (1) .save ()
翻译: load () .coalesce (1) . map(…).filter(…).save ()
这意味着您的所有数据将被压缩到一个单独的分区中,在那里它将被过滤,失去所有的并行性。 这种情况甚至会发生在非常简单的过滤器,如column='value'。
load().map(…).filter(…).repartition(1).save()
在这种情况下,在原始分区上并行地进行过滤。
举个数量级的例子,在我的例子中,当从Hive表加载后过滤109M行(~105G)和~1000个分区时,运行时从合并(1)的~6h下降到重新分区(1)的~2m。
具体示例取自AirBnB的这篇文章,这篇文章非常好,甚至涵盖了Spark中重新分区技术的更多方面。