1 问题描述

最近工作中有使用到spark sql的DataFrameWriter.insertInto函数往Hive表插入数据。在一次测试中，执行到该函数时，HDFS上产生了大量的小文件和目录，最终导致测试环境的namenode发生failover。

经过一些investigation后，发现是因为dataframe中的column list和hive表的column list排列顺序不一致，导致一个基数(cardinality)非常大的column被误认为partition column，进而产生了大量的临时文件和目录。

这个问题的解决方案本身很简单，只要确保dataframe的columns和hive表的columns保持名称和顺序都一致就可以了。但是，这个问题引发了我对spark sql insertInto函数内部实现原理的好奇心。在我们的case中，data frame的column names和hive表的column names已经是一样的，只不过顺序不完全一致，为什么spark没有按列名匹配呢？另外，还想搞清楚每个dataframe partition的数据是怎样写入到各个hive partition中的。

ok，所以我们有了两个问题：

    1. DataFrameWriter.insertInto函数写入hive表时，是怎样确定dataframe columns和hive表columns的对应关系的？

    2. 在将DataFrame的每个partition写入hive表时，是怎样把单个RDD partition的数据写入到单个或多个hive partition中的？

2 源码分析

有了问题，我们就要带着问题去查阅源码，找寻答案（注意，本文的源码版本为2.2.3) 。DataFrameWriter.insertInto函数的处理和执行过程涉及了spark sql的analyzer，optimizer，spark planner, catalog等模块，本文不打算go through每个环节，只会对与上述两个问题密切相关的模块进行源码分析，包括：

    1. 对insertInto语句进行预处理的analyzer中的规则：PreprocessTableInsertion

    2. 将数据写入hive表的逻辑计划(logical plan)：InsertIntoHiveTable

2.1 PreprocessTableInsertion

DataFrameWriter.insertInto方法会生成逻辑计划InsertIntoTable, 该逻辑计划会被analyzer中的规则PreprocessTableInsertion预处理，PreprocessTableInsertion会调用其preprocess方法进行处理：

PreprocessTableInsertion的apply方法

因为我们插入的是hive表，所以我们的relation会匹配HiveTableRelation。下文源码分析中，我们都会基于hive表作为目标表的前提来讨论，但读者需要清楚hive表不是InsertIntoTable的唯一目标数据源。

来看看PreprocessTableInsertion的preprocess方法里做了什么：

2.1.1 partition column的规范化检查

partition column的规范化检查

preprocess方法会对传入的partition columns进行normalize处理，这里的insert.partition是在insert into语句中指定的partition columns信息，partColNames是hive表的partition columns信息。 PartitioningUtils.normalizePartitionSpec方法做了以下事情：

1. 做大小写转换处理，将所有列名都转换成小写；

2. 检查指定的partition columns是否都是hive表的partition column；

3. 检查指定的partition columns是否有重复，如果有则直接抛出异常。

在我们的case中，通过DataFrameWriter.insertInto方法插入数据，并没有指定partition columns，所以在这里我们的insert.partition是一个空map。

然后，preprocess方法会抽取出所有的static partition columns (就是在insert into 语句中指定的常量分区列，例如，insert into tableA partition (dt='2019-06-18') ...)，除了static partition columns以外的partition columns就是dynamic partition columns。hive表中除了static partition columns以外的所有columns(包括dynamic partition columns和非分区columns)都需要由insert.query提供，所以这里会验证expectedColumns和insert.query.schema的长度是否匹配，如果不匹配则直接抛出异常。

2.1.2 output columns的重命名和转换

rename and cast of output columns

做完partition columns的规范化后，preprocess方法会判断normalizedPartSpec是否为空，

如果不为空，则说明用户指定了分区信息，则直接将normalizedPartSpec作为insertIntoTable逻辑计划的分区信息。

如果为空，则说明用户没有指定分区信息（比如直接调用DataFrameWriter.insertInto方法就不会指定分区信息），那么spark会将目标hive表的分区列partColNames作为insertIntoTable逻辑计划的分区信息。注意，这里partColNames.map(_ -> None).toMap生成的是一个partition column name到partition column value的map，这里所有partition column name都映射为None，表示所有分区列都是动态分区列。

最后，不管normalizedPartSpec是否为空，spark都会调用castAndRenameChildOutput方法将insertIntoTable逻辑计划的query的output columns强制重命名和转换成和目标hive表完全一致：

output columns的强制转化

可以看到，spark并没有根据列名来映射query和hive表的column list，而是直接根据column排列的顺序一一比对，只要不一致就直接将query的column重名为hive表的对应column，如果类型不匹配则会进行强制类型转换。是不是有点暴力？

2.2 InsertIntoHiveTable

经过PreprocessTableInsertion规则处理后的InsertIntoTable逻辑计划会进一步被规则HiveAnalysis处理。HiveAnalysis规则会将InsertIntoTable逻辑计划转换成InsertIntoHiveTable逻辑计划。

InsertIntoHiveTable继承自RunnableCommand, 而RunnableCommand最终都会被转换成物理计划ExecutedCommandExec, 本文不讨论spark的物理执行计划，关于spark逻辑计划到物理计划的转换读者可阅读SparkStrategies类的源码，上面提到的RunnableCommand逻辑计划就是在SparkStrategies的BasicOperators策略中被转换成ExecutedCommandExec物理计划的。

ExecutedCommandExec执行时最终会调用对应RunnableCommand对象的run方法，在我们这里就是InsertIntoHiveTable的run方法。下面我们就来看看InsertIntoHiveTable的run方法主要做了什么。

2.2.1 InsertIntoHiveTable.run方法

在正式写入数据之前，InsertIntoHiveTable.run方法会先获取和设置一系列的元数据信息，比如hive表的location，文件格式，压缩算法等。这里不讨论这些细节，有兴趣的读者可查阅InsertIntoHiveTable类的源码。这里主要讲一下写数据的过程，InsertIntoHiveTable.run方法调用了FileFormatWriter.write方法进行实际的数据写入工作：

FileFormatWriter.write called in InsertIntoHiveTable.run

2.2.2 FileFormatWriter.write方法

FileFormatWriter.write方法最核心的代码如下：

Sort the query by partition columns and run spark job to write data

1. 按partition columns排序

在运行spark job进行数据写入之前，FileFormatWriter.write方法会先判断InsertIntoHiveTable中的query的ordering是否满足hive partition的要求，即是否已经按照hive的partition columns排过序了（这里同样会检查bucket和非partition column的ordering要求）。

如果满足要求，则直接使用InsertIntoHiveTable中的query，否则就要加一个SortExec的物理计划对query的数据按照partition columns进行一次排序(如果有bucket或非partition column的ordering要求，也会将其加入进行排序)，注意这里的global=false, 所以是每个partition内部的局部排序，不是全局排序。

2. run spark job写入数据

最后FileFormatWriter.write方法会调用SparkContext.runJob方法起一个spark job来执行数据写入的任务。这个runJob方法的签名是：

FileFormatWriter.write调用的runJob方法的签名

我们看到，传入的rdd就是query对应的rdd，而传入的function是调用FileFormatWriter.executeTask方法。 FileFormatWriter.executeTask方法会根据写入的数据中是否存在动态分区的列来决定生成什么样的ExecuteWriteTask来执行数据写入任务：

生成ExecuteWriteTask对象

在我们的case中存在动态分区，所以我们讨论DynamicPartitionWriteTask，SingleDirectoryWriteTask比较简单，有兴趣的读者可自行阅读源码。

2.2.3 DynamicPartitionWriteTask

DynamicPartitionWriteTask的核心在其execute方法，DynamicPartitionWriteTask.execute方法的核心代码：

DynamicPartitionWriteTask.execute方法

DynamicPartitionWriteTask.execute方法会遍历单个rdd partition的每行数据，获取每行数据的partition columns。这里的getPartitionColsAndBucketId是一个UnsafeProjection对象，用于从row中抽取出partition和bucket columns。注意，这里的抽取方法是根据column name找到每个hive表partition column在row中的column index，也就是说这里我们是按列名而不是顺序匹配Hive表和query的columns的。

看到这里，有没有觉得spark做得有点不合理？既然前面在PreprocessTableInsertion已经按列的顺序做了columns的强制重命名和类型转换，那这里的按列名查找岂不是很多余？个人觉得PreprocessTableInsertion对Hive表和query的columns的映射机制可以做的更细化一些。比如，在我们的case中，query(data frame)和Hive表的column名字是一样的，只是顺序不一致而已，在这种情况下就不应该按列顺序做强制重命名和类型转换。我们后来修改了spark的代码，在PreprocessTableInsertion中去掉了按列顺序重命名的步骤，然后我们用重新编译的spark测试了我们的case，结果一切正常，没有出现大量文件的问题。当然，这只是针对我们的case，我们的修改也只是for test purpose. 至于该如何改进spark的这个行为，留给读者思考。

我们接着说，找到每行数据的partition columns后，DynamicPartitionWriteTask.execute方法会判断当前行和上一行是否同属一个partition，如果不是，则认为在当前partition数据中发现了一个新的hive partition，相应地就会在HDFS上新建一个目录来存放该partition的数据文件。因为前面我们已经按hive partition columns排过序了，所以这里的逻辑是合理的。新建目录和文件在方法newOutputWriter中完成。

最终，每条数据都会被写入到HDFS文件中：currentWriter.write(getOutputRow(row)). 注意，这里的getOutputRow也是根据列名而不是列顺序从row中获取需要写入到HDFS文件的数据的。

3 回答问题

ok，分析完了，现在来回答文章开头提出的两个问题：

    1. DataFrameWriter.insertInto函数写入hive表时，是怎样确定dataframe columns和hive表columns的对应关系的？

答：在进行逻辑计划的analysis时，PreprocessTableInsertion规则是按照列顺序将dataframe columns映射到hive表columns的（强制重命名和类型转换）；在执行数据写入hive表任务的DynamicPartitionWriteTask中，又是根据列名进行映射的。

    2. 在将DataFrame的每个partition写入hive表时，是怎样把单个RDD partition的数据写入到单个或多个hive partition中的？

答：DynamicPartitionWriteTask处理的单个RDD partition数据是已经按partition columns拍过序的，所以DynamicPartitionWriteTask可以在遍历每行数据时判断当前行数据的partition是否和上一行数据不一致，如果不一致则生成一个新的partition的output writer将数据写到新的hive partition对应的文件中去。

4 总结

本文从工作中遇到的大量文件夹和文件问题出发，剖析了DataFrameWriter.insertInto函数涉及的两个重要模块：PreprocessTableInsertion规则和InsertIntoHiveTable逻辑计划的实现细节，解释了为什么会出现大量文件夹和文件的问题，并对spark中query和hive表的列映射机制谈了下自己的看法，如有不对之处，望读者指出，谢谢。