Skip to content

Commit a1e3649

Browse files
jeanlynliancheng
authored andcommitted
[SPARK-8379] [SQL] avoid speculative tasks write to the same file
The issue link [SPARK-8379](https://issues.apache.org/jira/browse/SPARK-8379) Currently,when we insert data to the dynamic partition with speculative tasks we will get the Exception ``` org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException): Lease mismatch on /tmp/hive-jeanlyn/hive_2015-06-15_15-20-44_734_8801220787219172413-1/-ext-10000/ds=2015-06-15/type=2/part-00301.lzo owned by DFSClient_attempt_201506031520_0011_m_000189_0_-1513487243_53 but is accessed by DFSClient_attempt_201506031520_0011_m_000042_0_-1275047721_57 ``` This pr try to write the data to temporary dir when using dynamic parition avoid the speculative tasks writing the same file Author: jeanlyn <jeanlyn92@gmail.com> Closes #6833 from jeanlyn/speculation and squashes the following commits: 64bbfab [jeanlyn] use FileOutputFormat.getTaskOutputPath to get the path 8860af0 [jeanlyn] remove the never using code e19a3bd [jeanlyn] avoid speculative tasks write same file
1 parent 41ab285 commit a1e3649

File tree

2 files changed

+5
-7
lines changed

2 files changed

+5
-7
lines changed

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -198,7 +198,6 @@ case class InsertIntoHiveTable(
198198
table.hiveQlTable.getPartCols().foreach { entry =>
199199
orderedPartitionSpec.put(entry.getName, partitionSpec.get(entry.getName).getOrElse(""))
200200
}
201-
val partVals = MetaStoreUtils.getPvals(table.hiveQlTable.getPartCols, partitionSpec)
202201

203202
// inheritTableSpecs is set to true. It should be set to false for a IMPORT query
204203
// which is currently considered as a Hive native command.

sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveWriterContainers.scala

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -228,12 +228,11 @@ private[spark] class SparkHiveDynamicPartitionWriterContainer(
228228
newFileSinkDesc.setCompressCodec(fileSinkConf.getCompressCodec)
229229
newFileSinkDesc.setCompressType(fileSinkConf.getCompressType)
230230

231-
val path = {
232-
val outputPath = FileOutputFormat.getOutputPath(conf.value)
233-
assert(outputPath != null, "Undefined job output-path")
234-
val workPath = new Path(outputPath, dynamicPartPath.stripPrefix("/"))
235-
new Path(workPath, getOutputName)
236-
}
231+
// use the path like ${hive_tmp}/_temporary/${attemptId}/
232+
// to avoid write to the same file when `spark.speculation=true`
233+
val path = FileOutputFormat.getTaskOutputPath(
234+
conf.value,
235+
dynamicPartPath.stripPrefix("/") + "/" + getOutputName)
237236

238237
HiveFileFormatUtils.getHiveRecordWriter(
239238
conf.value,

0 commit comments

Comments
 (0)