Spark AWS EMR检查点位置

Question

我正在在EMR上运行Spark作业，但需要创建一个检查点。我尝试使用S3，但获得了此错误消息

17/02/24 14:34:35 ERROR ApplicationMaster: User class threw exception: 
java.lang.IllegalArgumentException: Wrong FS: s3://spark-
jobs/checkpoint/31d57e4f-dbd8-4a50-ba60-0ab1d5b7b14d/connected-
components-e3210fd6/2, expected: hdfs://ip-172-18-13-18.ec2.internal:8020
java.lang.IllegalArgumentException: Wrong FS: s3://spark-
jobs/checkpoint/31d57e4f-dbd8-4a50-ba60-0ab1d5b7b14d/connected-
components-e3210fd6/2, expected: hdfs://ip-172-18-13-18.ec2.internal:8020

这是我的示例代码

...
val sparkConf = new SparkConf().setAppName("spark-job")
  .set("spark.default.parallelism", (CPU * 3).toString)
  .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .registerKryoClasses(Array(classOf[Member], classOf[GraphVertex], classOf[GraphEdge]))
  .set("spark.dynamicAllocation.enabled", "true")

implicit val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
sparkSession.sparkContext.setCheckpointDir("s3://spark-jobs/checkpoint")
....

我如何在AWS EMR上检查点？

Answer 1

现在有一个已修复的Spark错误，这意味着您只能检查到默认的FS，而不是其他任何一个（例如S3）。它已修复在大师中，不知道backports。

如果它使您感觉更好，则检查点的工作方式是：Write Rename（）在对象存储上足够慢，您可能会发现自己在本地更好地检查了自己，然后自己将上传到S3。

Answer 2

主分支中有一个修复程序，以允许检查点到S3。我能够对其进行构建，并且可以使用它，所以这应该是下一个版本的一部分。

Answer 3

尝试使用AWS Authenticaton尝试一些东西：

val hadoopConf: Configuration = new Configuration()
  hadoopConf.set("fs.s3.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
  hadoopConf.set("fs.s3n.awsAccessKeyId", "id-1")
  hadoopConf.set("fs.s3n.awsSecretAccessKey", "secret-key")
  sparkSession.sparkContext.getOrCreate(checkPointDir, () => 
      { createStreamingContext(checkPointDir, config) }, hadoopConf)