Spark parquet 数据帧分区数

Question

我有一个 HDFS 文件夹，其中包含两个 250MB 的 parquet 文件。 hadoop df 块大小设置为 128MB。 有以下代码：

    JavaSparkContext sparkContext = new JavaSparkContext();

    SQLContext sqlContext = new SQLContext(sparkContext);
    DataFrame dataFrame = sqlContext.read().parquet("hdfs:////user/test/parquet-folder");
    LOGGER.info("Nr. of rdd partitions: {}", dataFrame.rdd().getNumPartitions());

    sparkContext.close();

我在集群上使用spark.executor.instances=3和spark.executor.cores=4运行它。我可以看到 parquet 文件的读取分为 3 个执行器 X 4 个核心 = 12 个任务：

   spark.SparkContext: Starting job: parquet at VerySimpleJob.java:25
   scheduler.DAGScheduler: Got job 0 (parquet at VerySimpleJob.java:25) with 12 output partitions

但是，当我获取数据帧 RDD（或使用 toJavaRDD() 创建 RDD）调用时，我只得到 4 个分区。这是由 hdfs 块大小控制的吗 - 每个文件 2 个块，因此有 4 个分区？

为什么这与 parquet（父级？）操作中的分区数量不匹配？

Answer 1

当您使用 Spark 读取文件时，执行程序的数量和核心的数量都不会以任何方式影响任务的数量。分区的数量（以及由此产生的任务）仅由输入中的块数量决定。如果您有 4 个小于 HDFS 块大小的文件 - 无论如何，这将是 4 个块，从而产生 4 个分区。公式为 number_of_files * number_of_blocks_in_file。因此，请查看您的文件夹并计算它包含多少个文件以及每个文件的大小。这应该可以回答你的问题。

UPD：如果您没有手动重新分区 DataFrame 并且您的 DataFrame 不是由于连接或任何其他随机操作而创建的，则上述所有内容都是正确的。

UPD：修复了答案详细信息。

Answer 2

这是一个老问题，但我也遇到了同样的问题。这是我修复它的方法：

spark.conf.set("spark.sql.files.maxPartitionBytes", "16777216")

我有 235 个文件，每个文件都不到 10MB。我正在使用动态分配，并且想要旋转更多任务。另外，我不想使用

repartition

来解决明显的性能问题。因此，通过将

spark.sql.files.maxPartitionBytes

的值设置为 16MB，我得到了 235 个分区，而不是 21 个（默认）。