Spark AWS emr 检查点位置

Posted 2023-04-18

【中文标题】Spark AWS emr 检查点位置

【英文标题】：Spark AWS emr checkpoint location

【发布时间】：2017-02-24 15:04:07

【问题描述】：

我正在 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 上检查点？

【问题讨论】：

请提供 spark 配置文件以正确查看情况。

@FaigB 刚刚做了，谢谢

【参考方案1】：

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

如果它让您感觉更好，检查点的工作方式：write then rename() 在对象存储上足够慢，您可能会发现自己在本地更好地检查点，然后自己上传到 s3。

【参考方案2】：

在 master 分支中对此进行了修复，以允许检查点到 s3。我能够针对它进行构建并且它有效，所以这应该是下一个版本的一部分。

【讨论】：

很高兴听到。一定要在检查点之间添加相当长的间隔，因为保存需要很长时间......从不频繁的检查点重建它可能会更快，并且检查点本身对性能的影响会更少。 YMMV

【参考方案3】：

尝试使用 AWS 身份验证，例如：

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)

【讨论】：

sparkContext中没有.getOrCreate(...)