PySpark：使用 binaryFiles() 函数读取二进制文件时进行分区

Question

sc = SparkContext("Local")
rdd = sc.binaryFiles(Path to the binary file , minPartitions = 5).partitionBy(8)

或

sc = SparkContext("Local")
rdd = sc.binaryFiles(Path to the binary file , minPartitions = 5).repartition(8)

使用上述任一代码，我试图在我的 RDD 中创建 8 个分区 {其中，我希望数据均匀分布在所有分区上}。当我打印 {rdd.getNumPartitions()} 时，显示的分区数仅为 8，但在 Spark UI 上，我观察到虽然创建了 8 个分区，但所有整个二进制文件数据只放在一个分区上。

注意：minPartition 属性不起作用。即使设置 minPartitions = 5，RDD 中的分区数也只有 1。因此，使用了 partitionBy/repartition 函数。

这是期望的行为还是我错过了什么？

Answer 1

Spark 2.4+，问题应该已解决，请参阅此答案下方的@Rahul 评论。

Spark 2.1-2.3，binaryFiles() 的 minPartitions 参数被忽略。请参阅Spark-16575 和commit changes to function setMinPartitions()。请注意，在提交更改中minPartitions 不再在函数中使用！

如果您使用binaryFiles() 读取多个二进制文件，输入文件将根据以下内容合并为分区：

• spark.files.maxPartitionBytes，默认 128 MB
• spark.files.openCostInBytes，默认 4 MB
• spark.default.parallelism
• 输入的总大小

前三个配置项描述为here。请参阅上面的提交更改以查看实际计算。

我有一个场景，我希望每个输入分区最多 40 MB，因此每个任务 40 MB...以增加解析时的并行度。 （Spark 将 128 MB 放入每个分区，降低了我的应用程序的速度。）我在调用 binaryFiles() 之前将 spark.files.maxPartitionBytes 设置为 40 M：

spark = SparkSession \
   .builder \
   .config("spark.files.maxPartitionBytes", 40*1024*1024)

对于只有一个输入文件，@user9864979 的答案是正确的：不能仅使用 binaryFiles() 将单个文件拆分为多个分区。

---

使用 Spark 1.6 读取多个文件时，minPartitions 参数确实有效，您必须使用它。否则，您将遇到Spark-16575 问题：您的所有输入文件都将被读入两个分区！

您会发现 Spark 通常会为您提供比您要求的更少的输入分区。我有一个场景，我希望每两个输入二进制文件都有一个输入分区。我发现将 minPartitions 设置为“输入文件的数量 * 7 / 10”大致可以满足我的需求。
我有另一种情况，我希望每个输入文件都有一个输入分区。我发现将minPartitions 设置为“输入文件数 * 2”给了我想要的结果。

Spark 1.5 行为binaryFiles()：每个输入文件都有一个分区。

Answer 2

TL;DR这是预期的行为。

由于您使用binaryFiles 读取文件，因此文件的全部内容将作为单个记录加载，并且单个记录不能跨多个分区拆分。这里根本没有什么可以分发的。