apache-spark-sql - Spark SQL 如何决定从 Hive 表加载数据时将使用的分区数？

Question

这个问题同Number of partitions of a spark dataframe created by reading the data from Hive table

但我认为这个问题没有得到正确答案。请注意，问题是询问当由于使用 SparkSession.sql 对 HIVE 表执行 sql 查询而创建数据帧时将创建多少个分区。方法。

IIUC，上面的问题不同于询问当由于执行诸如 spark.read.json("examples/src/main/resources/people.json") 之类的代码而创建数据帧时将创建多少个分区。它直接从文件系统加载数据——可能是 HDFS。我认为后一个问题的答案是由 spark.sql.files.maxPartitionBytes 给出的。

spark.sql.files.maxPartitionBytes 134217728 (128 MB) The maximum number of bytes to pack into a single partition when reading files.

实验上，我尝试从 HIVE 表创建一个数据框，但我得到的分区数量并未由 total data in hive table / spark.sql.files.maxPartitionBytes 解释。

还添加到 OP ，最好知道如何控制分区的数量，即，当您想强制 spark 使用与默认情况不同的数量时。

引用:

https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

Answer 1

TL;博士:从 Hive 读取数据时的默认分区数将由 HDFS 块大小控制。可以通过设置 mapreduce.job.maps 来增加分区数为适当的值，可以通过设置 mapreduce.input.fileinputformat.split.minsize 减小适当的值(value)

Spark SQL 创建 HadoopRDD 的实例从配置单元表加载数据时。

An RDD that provides core functionality for reading data stored in Hadoop (e.g., files in HDFS, sources in HBase, or S3), using the older MapReduce API (org.apache.hadoop.mapred).

HadoopRDD 依次根据 computeSplitSize 拆分输入文件 org.apache.hadoop.mapreduce.lib.input.FileInputFormat 中定义的方法(新 API)和 org.apache.hadoop.mapred.FileInputFormat (旧的 API)。

新 API:

protected long computeSplitSize(long blockSize, long minSize,
                                  long maxSize) {
    return Math.max(minSize, Math.min(maxSize, blockSize));
  }

旧 API:

protected long computeSplitSize(long goalSize, long minSize,
                                       long blockSize) {
    return Math.max(minSize, Math.min(goalSize, blockSize));
  }

computeSplitSize 根据 HDFS 块大小拆分文件，但如果块大小小于 minSize或大于 maxSize然后它被夹在那些极端。 HDFS 块大小可以从

hdfs getconf -confKey dfs.blocksize

根据 Hadoop 的权威指南表 8.5， minSize来自 mapreduce.input.fileinputformat.split.minsize和 maxSize来自 mapreduce.input.fileinputformat.split.maxsize .



然而，书中也提到了关于 mapreduce.input.fileinputformat.split.maxsize那:

This property is not present in the old MapReduce API (with the exception of CombineFileInputFormat). Instead, it is calculated indirectly as the size of the total input for the job, divided by the guide number of map tasks specified by mapreduce.job.maps (or the setNumMapTasks() method on JobConf).

this post 还使用总输入大小除以 map 任务数来计算 maxSize。