浅析Hive/Spark SQL读文件时的输入任务划分-阿里云开发者社区

开发者社区> Apache Spark中国社区> 正文

浅析Hive/Spark SQL读文件时的输入任务划分

简介: 本文最后留个思考题给读者们:如何设置参数彻底关闭Spark SQL data source表的文件合并? 积极回答问题即可获得社区礼物。

作者:
王道远,花名健身,阿里云EMR技术专家,Apache Spark活跃贡献者,主要关注大数据计算优化相关工作。


Hive以及Spark SQL等大数据计算引擎为我们操作存储在HDFS上结构化数据提供了易于上手的SQL接口,大大降低了ETL等操作的门槛,也因此在实际生产中有着广泛的应用。SQL是非过程化语言,我们写SQL的时候并不能控制具体的执行过程,它们依赖执行引擎决定。而Hive和Spark SQL作为Map-Reduce模型的分布式执行引擎,其执行过程首先就涉及到如何将输入数据切分成一个个任务,分配给不同的Map任务。在本文中,我们就来讲解Hive和Spark SQL是如何切分输入路径的。

Hive

Hive是起步较早的SQL on Hadoop项目,最早也是诞生于Hadoop中,所以输入划分这部分的代码与Hadoop相关度非常高。现在Hive普遍使用的输入格式是CombineHiveInputFormat,它继承于HiveInputFormat,而HiveInputFormat实现了Hadoop的InputFormat接口,其中的getSplits方法用来获取具体的划分结果,划分出的一份输入数据被称为一个“Split”。在执行时,每个Split对应到一个map任务。在划分Split时,首先挑出不能合并到一起的目录——比如开启了事务功能的路径。这些不能合并的目录必须单独处理,剩下的路径交给私有方法getCombineSplits,这样Hive的一个map task最多可以处理多个目录下的文件。在实际操作中,我们一般只要通过set mapred.max.split.size=xx;即可控制文件合并的大小。当一个文件过大时,父类的getSplits也会帮我们完成相应的切分工作。

Spark SQL

Spark的表有两种:DataSource表和Hive表。另外Spark后续版本中DataSource V2也将逐渐流行,目前还在不断发展中,暂时就不在这里讨论。我们知道Spark SQL其实底层是Spark RDD,而RDD执行时,每个map task会处理RDD的一个Partition中的数据(注意这里的Partition是RDD的概念,要和表的Partition进行区分)。因此,Spark SQL作业的任务切分关键在于底层RDD的partition如何切分。

Data Source表

Spark SQL的DataSource表在最终执行的RDD类为FileScanRDD,由FileSourceScanExec创建出来。在创建这种RDD的时候,具体的Partition直接作为参数传给了构造函数,因此划分输入的方法也在DataSourceScanExec.scala文件中。具体分两步:首先把文件划分为PartitionFile,再将较小的PartitionFile进行合并。

第一步部分代码如下:

  if (fsRelation.fileFormat.isSplitable(
    fsRelation.sparkSession, fsRelation.options, file.getPath)) {
    (0L until file.getLen by maxSplitBytes).map { offset =>
    val remaining = file.getLen - offset
    val size = if (remaining > maxSplitBytes) maxSplitBytes else remaining
    val hosts = getBlockHosts(blockLocations, offset, size)
    PartitionedFile(
      partition.values, file.getPath.toUri.toString,
      offset, size, partitionDeleteDeltas, hosts)
    }
  } else {
    val hosts = getBlockHosts(blockLocations, 0, file.getLen)
    Seq(PartitionedFile(partition.values, file.getPath.toUri.toString,
    0, file.getLen, partitionDeleteDeltas, hosts))
  }

我们可以看出,Spark SQL首先根据文件类型判断单个文件是否能够切割,如果可以则按maxSplitBytes进行切割。如果一个文件剩余部分无法填满maxSplitBytes,也单独作为一个Partition。

第二部分代码如下所示:

  splitFiles.foreach { file =>
    if (currentSize + file.length > maxSplitBytes) {
      closePartition()
    }
    // Add the given file to the current partition.
    currentSize += file.length + openCostInBytes
    currentFiles += file
  }

这样我们就可以依次遍历第一步切好的块,再按照maxSplitBytes进行合并。注意合并文件时还需加上打开文件的预估代价openCostInBytes。那么maxSplitBytesopenCostInBytes这两个关键参数怎么来的呢?

  val defaultMaxSplitBytes =
    fsRelation.sparkSession.sessionState.conf.filesMaxPartitionBytes
  val openCostInBytes = fsRelation.sparkSession.sessionState.conf.filesOpenCostInBytes
  val defaultParallelism = fsRelation.sparkSession.sparkContext.defaultParallelism
  val totalBytes = selectedPartitions.flatMap(_.files.map(_.getLen + openCostInBytes)).sum
  val bytesPerCore = totalBytes / defaultParallelism

  val maxSplitBytes = Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))

不难看出,主要是spark.sql.files.maxPartitionBytesspark.sql.files.openCostInBytes、调度器默认并发度以及所有输入文件实际大小所控制。

Hive表

Spark SQL中的Hive表底层的RDD类为HadoopRDD,由HadoopTableReader类实现。不过这次,具体的Partition划分还是依赖HadoopRDDgetPartitions方法,具体实现如下:

  override def getPartitions: Array[Partition] = {
    ...
    try {
      val allInputSplits = getInputFormat(jobConf).getSplits(jobConf, minPartitions)
      val inputSplits = if (ignoreEmptySplits) {
        allInputSplits.filter(_.getLength > 0)
      } else {
        allInputSplits
      }
      val array = new Array[Partition](inputSplits.size)
      for (i <- 0 until inputSplits.size) {
        array(i) = new HadoopPartition(id, i, inputSplits(i))
      }
      array
    } catch {
      ...
    }
  }

不难看出,在处理Hive表的时候,Spark SQL把任务划分又交给了Hadoop的InputFormat那一套。不过需要注意的是,并不是所有Hive表都归为这一类,Spark SQL会默认对ORC和Parquet的表进行转化,用自己的Data Source实现OrcFileFormatParquetFileFormat来把这两种表作为Data Source表来处理。

总结

切分输入路径只是大数据处理的第一步,虽然不起眼,但是也绝对不可或缺。低效的文件划分可能会给端到端的执行速度带来巨大的负面影响,更有可能影响到输出作业的文件布局,从而影响到整个数据流水线上所有作业的执行效率。万事开头难,为程序输入选择合适的配置参数,可以有效改善程序执行效率。

留个思考题给读者们:如何设置参数彻底关闭Spark SQL data source表的文件合并?

加入钉群或微信群,微信联系小编回复你的方案,即可收到社区礼物。


阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区,定期推送精彩案例,技术专家直播,问答区近万人Spark技术同学在线提问答疑,只为营造纯粹的Spark氛围,欢迎钉钉扫码加入!
image.png
对开源大数据和感兴趣的同学可以加小编微信(下图二维码,备注“进群”)进入技术交流微信群。image.png

版权声明:本文中所有内容均属于阿里云开发者社区所有,任何媒体、网站或个人未经阿里云开发者社区协议授权不得转载、链接、转贴或以其他方式复制发布/发表。申请授权请邮件developerteam@list.alibaba-inc.com,已获得阿里云开发者社区协议授权的媒体、网站,在转载使用时必须注明"稿件来源:阿里云开发者社区,原文作者姓名",违者本社区将依法追究责任。 如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件至:developer2020@service.aliyun.com 进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

阿里巴巴开源大数据技术团队成立Apache Spark中国技术社区,定期推送精彩案例,问答区数个Spark技术同学每日在线答疑,只为营造纯粹的Spark氛围,欢迎加入!钉钉群号:21784001,或点击链接申请加入 https://qr.dingtalk.com/action/joingroup?code=v1,k1,jr4ucSRJs6kkK7QZ5sw2oSHRGNoFjONsQgKxZGiTFXA=&_dt_no_comment=1&origin=11

官方博客
官网链接