refer: https://www.jianshu.com/p/182901f03296
本文以读取 parquet 文件 / parquet hive table 为例:
hive metastore 和 parquet 转化的方式通过 spark.sql.hive.convertMetastoreParquet 控制,默认为 true。
如果设置为 true ,会使用 org.apache.spark.sql.execution.FileSourceScanExec ,否则会使用 org.apache.spark.sql.hive.execution.HiveTableScanExec。
FileSourceScanExec
前者对分区规则做了一些优化,如果 文件是:
-
没有分桶的情况
分区大小计算公式:
bytesPerCore = totalBytes / defaultParallelism
maxSplitBytes = Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))
- defaultMaxSplitBytes:spark.sql.files.maxPartitionBytes,默认为128M,每个分区读取的最大数据量
- openCostInBytes: spark.sql.files.openCostInBytes,默认为4M,小于这个大小的文件将会合并到一个分区,可以理解为每个分区的最小量,避免碎文件造成的大量碎片任务。
- defaultParallelism: spark.default.parallelism,yarn默认为应用cores数量或2。
- bytesPerCore:数据总大小 / defaultParallelism
-
分桶的情况下
分区数取决于桶的数量
HiveTableScanExec
通过文件数量,大小进行分区。
例如:读入一份 2048M 大小的数据,hdfs 块大小设置为 128M
该目录有1000个小文件,则会生成1000个partition。
如果只有1个文件,则会生成 16 个分区。
如果有一个大文件1024M,其余 999 个文件共 1024M,则会生成 1007个分区。
网友评论