使用“partitionBy”对同一路径进行多个写入任务，当在FileOutputCommitter的cleanupJob中删除_temporary时，将失败，就像没有这样的文件或目录。
测试代码:
def batchTask[A](TASK_tag: String, taskData: TraversableOnce[A], batchSize: Int, fTask: A => Unit, fTaskId: A => String): Unit = {
var list = new scala.collection.mutable.ArrayBuffer[(String, java.util.concurrent.Future[Int])]()
val executors = java.util.concurrent.Executors.newFixedThreadPool(batchSize)
try {
taskData.foreach(d => {
val task = executors.submit(new java.util.concurrent.Callable[Int] {
override def call(): Int = {
fTask(d)
1
}
})
list += ((fTaskId(d), task))
})
var count = 0
list.foreach(r => if (!r._2.isCancelled) count += r._2.get())
} finally {
executors.shutdown()
}
}
def testWriteFail(outPath: String)(implicit spark: SparkSession, sc: SparkContext): Unit = {
println(s"try save: ${outPath}")
import org.apache.spark.sql.functions._
import spark.sqlContext.implicits._
batchTask[Int]("test", 1 to 20, 6, t => {
val df1 =
Seq((1, "First Value", java.sql.Date.valueOf("2010-01-01")), (2, "Second Value", java.sql.Date.valueOf("2010-02-01")))
.toDF("int_column", "string_column", "date_column")
.withColumn("t0", lit(t))
df1.repartition(1).write
.mode("overwrite")
.option("mapreduce.fileoutputcommitter.marksuccessfuljobs", false)
.partitionBy("t0").csv(outPath)
}, t => f"task.${t}%4d")
println(s"fail: count=${spark.read.csv(outPath).count()}")
}
try {
testWriteFail(outPath + "/fail")
} catch {
case e: Throwable =>
}
失败
使用 OutputCommitter：
package org.jar.spark.util
import java.io.IOException
object JMultiWrite {
val JAR_Write_Cache_Flag = "jar.write.cache.flag"
val JAR_Write_Cache_TaskId = "jar.write.cache.taskId"
val JAR_Write_Cache_Overwrite = "jar.write.cache.overwrite"
implicit class ImplicitWrite[T](dw: org.apache.spark.sql.DataFrameWriter[T]) {
def multiWrite(outDir: String, taskId: String, cacheDir: String = "_jarTaskCache", overwrite: Boolean = false): Boolean = {
val p = path(outDir, cacheDir, taskId)
dw.options(options(cacheDir, taskId))
.option(JAR_Write_Cache_Overwrite, overwrite)
.mode(org.apache.spark.sql.SaveMode.Overwrite)
.save(p)
true
}
}
def options(cacheDir: String, taskId: String): Map[String, String] = {
Map(JAR_Write_Cache_Flag -> cacheDir,
JAR_Write_Cache_TaskId -> taskId,
"mapreduce.fileoutputcommitter.marksuccessfuljobs" -> "false",
"mapreduce.job.outputformat.class" -> classOf[JarOutputFormat].getName
)
}
def path(outDir: String, cacheDir: String, taskId: String): String = {
assert(outDir != "", "need OutDir")
assert(cacheDir != "", "need CacheDir")
assert(taskId != "", "needTaskId")
outDir + "/" + cacheDir + "/" + taskId
}
class JarOutputFormat extends org.apache.hadoop.mapreduce.lib.output.TextOutputFormat {
var committer: org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter = _
override def getOutputCommitter(context: org.apache.hadoop.mapreduce.TaskAttemptContext): org.apache.hadoop.mapreduce.OutputCommitter = {
if (this.committer == null) {
val output = org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.getOutputPath(context)
this.committer = new JarOutputCommitter(output, context)
}
this.committer
}
}
class JarOutputCommitter(output: org.apache.hadoop.fs.Path, context: org.apache.hadoop.mapreduce.TaskAttemptContext)
extends org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter(output, context) {
override def commitJob(context: org.apache.hadoop.mapreduce.JobContext): Unit = {
val finalOutput = this.output
val cacheFlag = context.getConfiguration.get(JAR_Write_Cache_Flag, "")
val myTaskId = context.getConfiguration.get(JAR_Write_Cache_TaskId, "")
val overwrite = context.getConfiguration.getBoolean(JAR_Write_Cache_Overwrite, false)
val hasCacheFlag = finalOutput.getName == myTaskId && finalOutput.getParent.getName == cacheFlag
val finalReal = if (hasCacheFlag) finalOutput.getParent.getParent else finalOutput
val fs = finalOutput.getFileSystem(context.getConfiguration)
val jobAttemptPath = getJobAttemptPath(context)
val arr$ = fs.listStatus(jobAttemptPath, new org.apache.hadoop.fs.PathFilter {
override def accept(path: org.apache.hadoop.fs.Path): Boolean = !"_temporary".equals(path.getName())
})
if (hasCacheFlag && overwrite)
{
if (fs.isDirectory(finalReal)) arr$.foreach(stat =>
if (fs.isDirectory(stat.getPath)) fs.listStatus(stat.getPath).foreach(stat2 => {
val p1 = stat2.getPath
val p2 = new org.apache.hadoop.fs.Path(finalReal, p1.getName)
if (fs.isDirectory(p1) && fs.isDirectory(p2) && !fs.delete(p2, true)) throw new IOException("Failed to delete " + p2)
})
)
}
arr$.foreach(stat => {
mergePaths(fs, stat, finalReal)
})
cleanupJob(context)
if (hasCacheFlag) {
try {
fs.delete(finalOutput, false)
val pp = finalOutput.getParent
if (fs.listStatus(pp).isEmpty)
fs.delete(pp, false)
} catch {
case e: Exception =>
}
}
}
}
@throws[IOException]
def mergePaths(fs: org.apache.hadoop.fs.FileSystem, from: org.apache.hadoop.fs.FileStatus, to: org.apache.hadoop.fs.Path): Unit = {
if (from.isFile) {
if (fs.exists(to) && !fs.delete(to, true)) throw new IOException("Failed to delete " + to)
if (!fs.rename(from.getPath, to)) throw new IOException("Failed to rename " + from + " to " + to)
}
else if (from.isDirectory) if (fs.exists(to)) {
val toStat = fs.getFileStatus(to)
if (!toStat.isDirectory) {
if (!fs.delete(to, true)) throw new IOException("Failed to delete " + to)
if (!fs.rename(from.getPath, to)) throw new IOException("Failed to rename " + from + " to " + to)
}
else {
val arr$ = fs.listStatus(from.getPath)
for (subFrom <- arr$) {
mergePaths(fs, subFrom, new org.apache.hadoop.fs.Path(to, subFrom.getPath.getName))
}
}
}
else if (!fs.rename(from.getPath, to)) throw new IOException("Failed to rename " + from + " to " + to)
}
}
然后：
def testWriteOk(outPath: String)(implicit spark: SparkSession, sc: SparkContext): Unit = {
println(s"try save: ${outPath}")
import org.apache.spark.sql.functions._
import org.jar.spark.util.JMultiWrite.ImplicitWrite
import spark.sqlContext.implicits._
batchTask[Int]("test.ok", 1 to 20, 6, t => {
val taskId = t.toString
val df1 =
Seq((1, "First Value", java.sql.Date.valueOf("2010-01-01")), (2, "Second Value", java.sql.Date.valueOf("2010-02-01")))
.toDF("int_column", "string_column", "date_column")
.withColumn("t0", lit(taskId))
df1.repartition(1).write
.partitionBy("t0")
.format("csv")
.multiWrite(outPath, taskId, overwrite = true)
}, t => f"task.${t}%4d")
println(s"ok: count=${spark.read.csv(outPath).count()}")
}
try {
testWriteOk(outPath + "/ok")
} catch {
case e: Throwable =>
}
成功：
$ ls ok/
t0=1 t0=10 t0=11 t0=12 t0=13 t0=14 t0=15 t0=16 t0=17 t0=18 t0=19 t0=2 t0=20 t0=3 t0=4 t0=5 t0=6 t0=7 t0=8 t0=9
对于其他输出格式同样适用，请注意使用 overwrite。
在 Spark 2.11.8 上进行测试。
感谢 @Tal Joffe。
_temporary文件夹（顺便说一句，这是个坏主意..） - WestCoastProjects