spark 读取 hdfs 数据分区规则

下文以读取 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

前者对分区规则做了一些优化,如果 文件是:

  1. 没有分桶的情况

    分区大小计算公式:

    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

    eg. 读入一份 2048M 大小的数据

    Tip: partitionSize的计算过程简化,实际上会先对读入的每个分区按maxSplitBytes做切割,切割完后如果的小文件如果大小不足maxSplitBytes的,会合并到一个partition,直到大小 > maxSplitBytes。

    //如果 spark.default.parallelism 设置为 1000,最终的分区数量是 512,每个分区大小为4M
    maxSplitBytes = Math.min(128M, Math.max(4M, 2M))
    partitionSize = 2048 / 4 = 512 
    
    //如果 spark.default.parallelism 设置为 100, 最终的分区数量是 100,每个分区大小为20.48M
    maxSplitBytes = Math.min(128M, Math.max(4M, 20.48M))
    partitionSize = 2048 / 20.48 = 100     
    
    //如果 spark.default.parallelism 设置为 10, 最终的分区数量是 16,每个分区大小为128M
    maxSplitBytes = Math.min(128M, Math.max(4M, 204.8M))
    partitionSize = 2048 / 128 = 16     
    
  2. 分桶的情况下:

    分区数取决于桶的数量。

HiveTableScanExec

通过文件数量,大小进行分区。

eg. 读入一份 2048M 大小的数据,hdfs 块大小设置为 128M

  1. 该目录有1000个小文件,则会生成1000个partition。
  2. 如果只有1个文件,则会生成 16 个分区。
  3. 如果有一个大文件1024M,其余 999 个文件共 1024M,则会生成 1009 个分区。
    原文作者:breeze_lsw
    原文地址: https://www.jianshu.com/p/182901f03296
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞