Optimizing SparkSQL Writes to Hive Dynamic Partitions

Liao Jiayi Liao Jiayi #Spark#Apache Spark

Hive tables are often partitioned by time and populated through SparkSQL dynamic-partition inserts. This reveals an opportunity for optimization.

Translated from Chinese with AI · Read the original

In real applications, we often create time-partitioned Hive tables and insert through SparkSQL dynamic partitions. I found an opportunity for optimization.

Application Background

For behavioral-data ingestion, differences between SDK timestamps and reception times produce historical records we do not want to discard. We therefore use SQL like this to load the warehouse.

// 假设time的时间单位是天
insert into table user_behaviours partition(time)
select
    *,
    toDate(time)
from
    user_behaviours_table

Most records actually belong to the same day, with a small amount of bad or delayed data belonging to earlier dates. How does Spark execute a dynamic-partition insert in this situation?

SparkPlan

I recommend reading the source alongside this explanation. In InsertIntoHiveTable.scala, doExecute reveals that sideEffectResult is the entry point for execution. Skipping the preliminary steps, Spark provides two Hive write paths according to numDynamicPartitions:

  • SparkHiveWriterContainer (numDynamicPartitions = 0)
  • SparkHiveDynamicPartitionWriterContainer (numDynamicPartitions > 0)

SparkHiveWriterContainer

Since the destination partition name is known, Spark simply initializes the appropriate Hive writer and writes the data.

SparkHiveDynamicPartitionWriterContainer

I added comments to the source below for reference.

// 新建external sorter以key进行排序
val sorter: UnsafeKVExternalSorter = new UnsafeKVExternalSorter(
  StructType.fromAttributes(partitionOutput),
  StructType.fromAttributes(dataOutput),
  SparkEnv.get.blockManager,
  SparkEnv.get.serializerManager,
  TaskContext.get().taskMemoryManager().pageSizeBytes,
  SparkEnv.get.conf.getLong("spark.shuffle.spill.numElementsForceSpillThreshold",
    UnsafeExternalSorter.DEFAULT_NUM_ELEMENTS_FOR_SPILL_THRESHOLD))


while (iterator.hasNext) {
  val inputRow = iterator.next()
  val currentKey = getPartitionKey(inputRow)
  sorter.insertKV(currentKey, getOutputRow(inputRow))
}


logInfo(s"Sorting complete. Writing out partition files one at a time.")


// 排序结束，开始写入文件
val sortedIterator = sorter.sortedIterator()
try {
  while (sortedIterator.next()) {
    // 由于sortedIterator根据key有序(key就是动态分区的值)，所以当key不等于上一个key时，表示应该写入一个新文件，且旧文件的输出流可以关闭。
    if (currentKey != sortedIterator.getKey) {
      if (currentWriter != null) {
        currentWriter.close(false)
      }
      currentKey = sortedIterator.getKey.copy()
      logDebug(s"Writing partition: $currentKey")
      // 根据新的currentKey新建HiveWriter
      currentWriter = newOutputWriter(currentKey)
    }


    // 写入数据
    var i = 0
    while (i < fieldOIs.length) {
      outputData(i) = if (sortedIterator.getValue.isNullAt(i)) {
        null
      } else {
        wrappers(i)(sortedIterator.getValue.get(i, dataTypes(i)))
      }
      i += 1
    }
    currentWriter.write(serializer.serialize(outputData, standardOI))
  }
} finally {
  if (currentWriter != null) {
    currentWriter.close(false)
  }
}
commit()

Spark sorts by key before writing. Why?

Presumably Spark assumes dynamic inserts create many partitions. Without sorting, concurrent writes to many files cause random disk I/O and reduce write performance. Sorted data ensures sequential writes to each file.

Improvement

If most data belongs to one partition, writes will already be mostly sequential without sorting. Sorting can then cost more than occasional random I/O. We can introduce a new parameter:

spark.sql.hive.useDynamicPartitionWriter

Users can choose whether to enable DynamicPartitionWriter. If disabled, create a map caching writers for all distinct keys. The changes are:

logInfo("UseDynamicPartitionContains is false, skip sorting.")
val keyWriterMap: mutable.Map[InternalRow, FileSinkOperator.RecordWriter] = mutable.Map()
try {
  while (iterator.hasNext) {
    val internalRow = iterator.next()
    val key = getPartitionKey(internalRow)
    if (currentKey != key) {
      currentKey = key.copy()
      logDebug(s"Writing partition: $currentKey")
      currentWriter = keyWriterMap.getOrElseUpdate(currentKey, newOutputWriter(currentKey))
    }


    var i = 0
    while (i < fieldOIs.length) {
      outputData(i) = internalRow.get(i, dataTypes(i))
      i += 1
    }
    currentWriter.write(serializer.serialize(outputData, standardOI))
  }
} finally {
  keyWriterMap.values.foreach(writer => {
    if (writer != null) {
      writer.close(false)
    }
  })
}
commit()

Jiayi Blog