从 Spark 分段上传到 S3 时出错

Question

尝试关闭序列文件编写器时，我收到错误“部分编号的上传尝试：2 已达到最大限制：5，将引发异常并失败”。异常的完整日志如下：

16/12/30 19时47分01秒INFO s3n.MultipartUploadOutputStream：uploadPart的/ mnt / S3 / 57b63810-C20A-438C-a73f-48d50e0be7d2-0001 94317523字节MD5：05ww / fe3pNni9Zvfm + l4Gg == md5hex：d39c30fdf7b7a4d9e2f59bdf9be9781a
16/12/30 19时47分12秒INFO s3n.MultipartUploadOutputStream：uploadPart / mnt1上/ S3 / 57b63810-C20A-438C-a73f-48d50e0be7d2-0002 94317523字节MD5：05ww / fe3pNni9Zvfm + l4Gg == md5hex：d39c30fdf7b7a4d9e2f59bdf9be9781a
16/12/30 19点47分23秒INFO s3n.MultipartUploadOutputStream：uploadPart的/ mnt / S3 / 57b63810-C20A-438C-a73f-48d50e0be7d2-0003 94317523字节MD5：05ww / fe3pNni9Zvfm + l4Gg == md5hex：d39c30fdf7b7a4d9e2f59bdf9be9781a
16/12/30 19时47分35秒INFO s3n.MultipartUploadOutputStream：uploadPart / mnt1上/ S3 / 57b63810-C20A-438C-a73f-48d50e0be7d2-0004 94317523字节MD5：05ww / fe3pNni9Zvfm + l4Gg == md5hex：d39c30fdf7b7a4d9e2f59bdf9be9781a
16/12/30 19时47分46秒INFO s3n.MultipartUploadOutputStream：uploadPart的/ mnt / S3 / 57b63810-C20A-438C-a73f-48d50e0be7d2-0005 94317523字节MD5：05ww / fe3pNni9Zvfm + l4Gg == md5hex：d39c30fdf7b7a4d9e2f59bdf9be9781a
16/12/30 19:47:57 错误 s3n.MultipartUploadOutputStream：第 2 部分的上传尝试已达到最大限制：5，将引发异常并失败
16/12/30 19:47:57 信息 s3n.MultipartUploadOutputStream: completeMultipartUpload 键错误：输出/part-20176
java.lang.IllegalStateException：已达到部分上传尝试的最大限制
    在 com.amazon.ws.emr.hadoop.fs.s3n.MultipartUploadOutputStream.spawnNewFutureIfNeeded(MultipartUploadOutputStream.java:362)
    在 com.amazon.ws.emr.hadoop.fs.s3n.MultipartUploadOutputStream.uploadMultiParts(MultipartUploadOutputStream.java:422)
    在 com.amazon.ws.emr.hadoop.fs.s3n.MultipartUploadOutputStream.close(MultipartUploadOutputStream.java:471)
    在 org.apache.hadoop.fs.FSDataOutputStream$PositionCache.close(FSDataOutputStream.java:74)
    在 org.apache.hadoop.fs.FSDataOutputStream.close(FSDataOutputStream.java:108)
    在 org.apache.hadoop.io.SequenceFile$Writer.close(SequenceFile.java:1290)
   ...
    在 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$18.apply(RDD.scala:727)
    在 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsWithIndex$1$$anonfun$apply$18.apply(RDD.scala:727)
    在 org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    在 org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:300)
    在 org.apache.spark.rdd.RDD.iterator(RDD.scala:264)
    在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    在 org.apache.spark.scheduler.Task.run(Task.scala:88)
    在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
    在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    在 java.lang.Thread.run(Thread.java:745)
16/12/30 19:47:59 信息 s3n.MultipartUploadOutputStream：uploadPart 错误 com.amazonaws.AbortedException： 
16/12/30 19:48:18 信息 s3n.MultipartUploadOutputStream：uploadPart 错误 com.amazonaws.AbortedException： 

我只是收到 5 次重试失败的错误。我不明白这样做的原因。有没有人见过这个错误？这可能是什么原因？

我正在使用我自己的多输出格式实现编写序列文件：

class MultiOutputSequenceFileWriter(prefix: String, suffix: String) extends Serializable {
   private val writers = collection.mutable.Map[String, SequenceFile.Writer]()

   /**
     * @param pathKey    folder within prefix where the content will be written
     * @param valueKey   key of the data to be written
     * @param valueValue value of the data to be written
     */
   def write(pathKey: String, valueKey: Any, valueValue: Any) = {
     if (!writers.contains(pathKey)) {
       val path = new Path(prefix + "/" + pathKey + "/" + "part-" + suffix)
       val hadoopConf = new conf.Configuration()
       hadoopConf.setEnum("io.seqfile.compression.type", SequenceFile.CompressionType.NONE)
       val fs = FileSystem.get(hadoopConf)
       writers(pathKey) = SequenceFile.createWriter(hadoopConf, Writer.file(path),
         Writer.keyClass(valueKey.getClass()),
         Writer.valueClass(valueValue.getClass()),
         Writer.bufferSize(fs.getConf().getInt("io.file.buffer.size", 4096)), //4KB
         Writer.replication(fs.getDefaultReplication()),
         Writer.blockSize(1073741824), // 1GB
         Writer.progressable(null),
         Writer.metadata(new Metadata()))
     }
     writers(pathKey).append(valueKey, valueValue)
   }
   def close = writers.values.foreach(_.close())
}

我正在尝试按如下方式编写序列文件：

...
rdd.mapPartitionsWithIndex { (p, it) => {
  val writer = new MultiOutputSequenceFileWriter("s3://bucket/output/", p.toString)
  for ( (key1, key2, data) <- it) {
    ...
    writer.write(key1, key2, data)
    ...
  }
  writer.close
  Nil.iterator
}.foreach( (x:Nothing) => ()) // To trigger iterator
}
...

笔记：

• 当我尝试关闭编写器时出现异常（我认为编写器尝试在关闭之前编写内容，并且我认为由于此原因会出现异常）。
• 我用相同的输入再试了两次相同的工作。我在第一次重新运行时没有出现错误，但在第二次重新运行时出现了三个错误。这可能只是 S3 中的暂时性问题吗？
• 失败的零件文件不在 S3 中。

Answer 1

AWS 支持工程师提到，在发生错误时，存储桶上有很多点击。该作业正在重试默认次数 (5)，并且很可能所有重试都受到限制。现在，我在提交作业时添加了以下配置参数，从而增加了重试次数。

spark.hadoop.fs.s3.maxRetries=20

此外，我压缩了输出，以便减少对 S3 的请求数量。在这些更改之后，我还没有看到几次运行失败。