。

可以设置spark.sql.maxConcurrentOutputFileWriters

这部分主要分为三个部分:

Spark中的实现中对于多分区的写入默认会先排序这是没必要的。

可以设置spark.sql.maxConcurrentOutputFileWriters

这部分主要分为三个部分:

另一个是FileFormatWriter中的dataWriter的选择;

直接转到代码部分:

!write.child.isInstanceOf[WriteFiles]

val

AttributeMap(write.query.output.zip(newQuery.output))val

writeFiles

write.partitionColumns,write.bucketSpec,

write.options,

writeFiles.transformExpressions

{case

a.withExprId(attrMap(a).exprId)}val

newWrite

a.withExprId(attrMap(a).exprId)}newWrite}}

else

prepareQuery是对满足条件的计划前加上Sort逻辑排序其中prepareQuery关键的代码如下:

val

write.requiredOrdering.map(_.transform

{case

a)}.asInstanceOf[SortOrder])val

outputOrdering

empty2NullPlan.outputOrderingval

orderingMatched

isOrderingMatched(requiredOrdering.map(_.child),

outputOrdering)if

empty2NullPlan)}write.requiredOrdering中涉及到的类为InsertIntoHadoopFsRelationCommand和InsertIntoHiveTable,且这两个物理计划中的requiredOrdering实现都是

V1WritesUtils.getSortOrder(outputColumns,

partitionColumns,

V1WritesUtils.getBucketSortColumns(bucketSpec,

dataColumns)if

(SQLConf.get.maxConcurrentOutputFileWriters

sortColumns.isEmpty)

columns.(dynamicPartitionColumns

writerBucketSpec.map(_.bucketIdExpression)

spark.sql.maxConcurrentOutputFileWriters为0默认值为0则会加上Sort逻辑计划具体的实现可以参考SPARK-37287

如果spark.sql.maxConcurrentOutputFileWriters为0默认值为0且

sortColumns为空(大部分情况下为空除非建表是partition加上bucket),则不会加上Sort逻辑计划

FileFormatWriter

InsertIntoHadoopFsRelationCommand和InsertIntoHiveTable

这两个物理计划中最终写入文件/数据的时候会调用到FileFormatWriter.write方法这里有个concurrentOutputWriterSpecFunc函数变量的设置

val

outputSpec)createConcurrentOutputWriterSpec(sparkSession,

sortPlan,

committer,concurrentOutputWriterSpecFunc

concurrentOutputWriterSpecFunc)executeWrite(sparkSession,

plan,

job)设置concurrentOutputWriterSpecFunc的代码如下

private

createConcurrentOutputWriterSpec(sparkSession:

Seq[Attribute]):

Option[ConcurrentOutputWriterSpec]

{val

sparkSession.sessionState.conf.maxConcurrentOutputFileWritersval

maxWriters

{Some(ConcurrentOutputWriterSpec(maxWriters,

else

spark.sql.maxConcurrentOutputFileWriters为0默认值为0,则ConcurrentOutputWriterSpec为None

spark.sql.maxConcurrentOutputFileWriters大于0且

sortColumns为空(大部分情况下为空除非建表是partition加上bucket)则为Some(ConcurrentOutputWriterSpec(maxWriters,

其中executeWrite会调用WriteFilesExec.doExecuteWrite方法,从而调用FileFormatWriter.executeTask,这里就涉及到dataWriter选择

val

EmptyDirectoryDataWriter(description,

taskAttemptContext,

(description.partitionColumns.isEmpty

description.bucketSpec.isEmpty)

{new

SingleDirectoryDataWriter(description,

taskAttemptContext,

DynamicPartitionDataConcurrentWriter(description,

taskAttemptContext,

DynamicPartitionDataSingleWriter(description,

taskAttemptContext,

concurrentOutputWriterSpec来选择不同的dataWriter,默认情况下为DynamicPartitionDataSingleWriter

否则就会为DynamicPartitionDataConcurrentWriter

Spark中为什么会加上Sort

至于Spark在写入文件的时候会加上Sort这个是跟写入的实现有关的也就是DynamicPartitionDataSingleWriter和DynamicPartitionDataConcurrentWriter的区别:

DynamicPartitionDataSingleWriter

在任何时刻只有一个writer在写文件这能保证写入的稳定性不会在写入文件的时候消耗大量的内存但是速度会慢DynamicPartitionDataConcurrentWriter

会有多个

同时写文件能加快写入文件的速度但是因为多个文件的同时写入可能会导致OOM

对于DynamicPartitionDataSingleWriter

会根据partition或者bucket作为最细粒度来作为writer的标准如果相邻的两条记录所属不同的partition或者bucket则会切换writer所以说如果不根据partition或者bucket排序的话会导致writer频繁的切换这会大大降低文件的写入速度。

所以说需要根据partition或者bucket进行排序。

[SPARK-37287][SQL]