apache-spark - 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 , minPartitions binaryFiles() 的论据被忽略。见 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 放入每个分区，减慢了我的应用程序的速度。)我设置了 spark.files.maxPartitionBytes调用前 40 M binaryFiles() :

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() 的行为: 每个输入文件都有一个分区。