每个分区中对象数量的火花合并

Question

我们开始在团队中尝试spark。在我们减少spark中的工作后，我们希望将结果写入S3，但是我们希望避免收集Spark结果。目前，我们正在为RDD的每个分区写文件，但是这会产生很多小文件。我们希望能够将html" target="_blank">数据聚合到几个文件中，这些文件按照写入文件的对象数量进行分区。例如，我们的总数据是100万个对象(这是不变的)，我们希望生成40万个对象文件，而我们当前的分区生成大约2万个对象文件(这因每个作业而异)。理想情况下，我们希望生成3个文件，每个文件包含400k、400k和200k，而不是50个20K对象的文件

有人有好的建议吗？

我的思考过程是让每个分区处理它应该写入的索引，假设每个分区将大致产生相同数量的对象。例如，分区0将写入第一个文件，而分区21将写入第二个文件，因为它将假设对象的起始索引为20000*21=42000，这比文件大小大。分区41将写入第三个文件，因为它大于2*文件大小限制。然而，这并不总是导致完美的400k文件大小限制，更多的是近似值。

我知道有合并，但正如我所理解的，合并是根据想要的分区数量减少分区数量。我想要的是根据每个分区中对象的数量合并数据，有没有好的方法可以做到这一点？

Answer 1

我们已经决定继续使用生成的文件数量，并确保每个文件包含少于100万个行项目

Answer 2

您要做的是将文件重新分区为三个分区;数据将拆分为每个分区大约 333K 条记录。分区将是近似的，每个分区不会正好是 333，333。我不知道有什么方法可以获得您想要的 400k/400k/200k 分区。

如果您有一个DataFrame'df'，您可以重新分区为n个分区

df.repartition(n)

由于您需要每个分区的最大记录数，我建议您这样做（您没有指定Scala或pyspark，所以我选择Scala；您可以在pyspard中这样做）：

val maxRecordsPerPartition = ???
val numPartitions = (df.count() / maxRecordsPerPartition).toInt + 1
df
    .repartition(numPartitions)
    .write
    .format('json')
    .save('/path/file_name.json')

这将确保您的分区小于maxRecordsPerPartition。