在AWS EMR上使用自定义jar的Spark流作业在写入时失败

Question

我正在尝试使用流数据帧将一个文件（csv.gz格式）转换为拼花地板。我必须使用流式数据帧，因为压缩的文件大小约为700 MB。作业是使用AWS EMR上的自定义jar运行的。源、目标和检查点位置都在AWS S3上。但一旦我尝试写入检查点，作业就会失败，并出现以下错误：

java.lang.IllegalArgumentException: 
Wrong FS: s3://my-bucket-name/transformData/checkpoints/sourceName/fileType/metadata,
expected: hdfs://ip-<ip_address>.us-west-2.compute.internal:8020

在EMR集群上运行的其他spark作业从S3读写并成功运行到S3（但它们不使用spark流）。所以我不认为这是S3文件系统访问的问题，正如本文所建议的那样。我也看了这个问题，但答案对我的情况没有帮助。我使用的是Scala:2.11.8和Spark:2.1.0。以下是我到目前为止的代码

...

    val spark = conf match {
      case null =>
        SparkSession
          .builder()
          .appName(this.getClass.toString)
          .getOrCreate()
      case _ =>
        SparkSession
          .builder()
          .config(conf)
          .getOrCreate()
    }

    // Read CSV file into structured streaming dataframe
    val streamingDF = spark.readStream
      .format("com.databricks.spark.csv")
      .option("header", "true")
      .option("delimiter","|")
      .option("timestampFormat", "dd-MMM-yyyy HH:mm:ss")
      .option("treatEmptyValuesAsNulls", "true")
      .option("nullValue","")
      .schema(schema)
      .load(s"s3://my-bucket-name/rawData/sourceName/fileType/*/*/fileNamePrefix*")
      .withColumn("event_date", "event_datetime".cast("date"))
      .withColumn("event_year", year($"event_date"))
      .withColumn("event_month", month($"event_date"))

    // Write the results to Parquet
    streamingDF.writeStream
      .format("parquet")
      .option("path", "s3://my-bucket-name/transformedData/sourceName/fileType/")
      .option("compression", "gzip")
      .option("checkpointLocation", "s3://my-bucket-name/transformedData/checkpoints/sourceName/fileType/")
      .partitionBy("event_year", "event_month")
      .trigger(ProcessingTime("900 seconds"))
      .start()

我还尝试在URI中使用s3n://而不是s3://但这似乎没有任何效果。

Answer 1

Tl；dr升级spark或避免使用s3作为检查点位置

ApacheSpark（结构化流）：S3检查点支持

此外，您可能应该使用s3a：//指定写入路径

S3a:系统是S3本机s3n://文件系统的继承者，它使用Amazon的库与S3进行交互。这允许S3a支持更大的文件（不超过5GB限制）、更高性能的操作等。该文件系统旨在取代S3 Native的/succession：通过替换URL模式，可以从s3n://URL访问的所有对象也可以从s3a访问。

https://wiki.apache.org/hadoop/AmazonS3