Spark 找不到 S3 文件

Question

我正在尝试使用以下命令使用 Apache Spark (Pyspark) 读取位于 S3 上的文件，并收到以下错误。该文件是年/月/日/小时/分钟分区中的压缩 JSON。我已经成功地从这家商店提取了几个月的数据，但是由于这个特定的文件，我收到了错误。我在连接到 AWS EMR 的 AWS Sagemaker Notebook 上运行它。

命令

# both fail
spark.read.json('s3://my_bucket/my_prefix/2021/08/31/08/53/').show()
spark.read.json('s3://my_bucket/my_prefix/2021/08/31/08/53/my_file.gz').show()

错误

An error was encountered:
An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 10.0 failed 4 times, most recent failure: Lost task 0.3 in stage 10.0 (TID 22) (ip-10-251-32-235.eu-west-1.compute.internal executor 19): java.io.FileNotFoundException: No such file or directory 's3://my_bucket/my_prefix/2021/08/31/08/53/my_file.gz'
It is possible the underlying files have been updated. You can explicitly invalidate the cache in Spark by running 'REFRESH TABLE tableName' command in SQL or by recreating the Dataset/DataFrame involved.
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.org$apache$spark$sql$execution$datasources$FileScanRDD$$anon$$readCurrentFile(FileScanRDD.scala:194)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:240)
    at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:159)
    at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
    at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
    at org.apache.spark.api.python.SerDeUtil$AutoBatchedPickler.hasNext(SerDeUtil.scala:85)
    at scala.collection.Iterator.foreach(Iterator.scala:941)
    at scala.collection.Iterator.foreach$(Iterator.scala:941)
    at org.apache.spark.api.python.SerDeUtil$AutoBatchedPickler.foreach(SerDeUtil.scala:80)
    at org.apache.spark.api.python.PythonRDD$.writeIteratorToStream(PythonRDD.scala:307)
    at org.apache.spark.api.python.PythonRunner$$anon$2.writeIteratorToStream(PythonRunner.scala:621)
    at org.apache.spark.api.python.BasePythonRunner$WriterThread.$anonfun$run$1(PythonRunner.scala:397)
    at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1996)
    at org.apache.spark.api.python.BasePythonRunner$WriterThread.run(PythonRunner.scala:232)

我到目前为止所做的尝试：

• 重新启动 spark 集群
• 验证文件；它之前已在此错误发生前 10 天更新
• 已验证这是唯一的 Spark 问题；我能够使用下面的非火花命令看到该文件

# this command successfully prints out the erroneous file
import boto3
s3 = boto3.resource('s3')
my_bucket = s3.Bucket('my_bucket')

for my_bucket_object in my_bucket.objects.filter(Prefix='my_prefix/2021/08/31/08/53'):
    print(my_bucket_object)

• 创建表并刷新它

# Tried this
spark.sql("CREATE OR REPLACE TEMPORARY VIEW bad_file USING json OPTIONS" + 
      " (path 's3://my_bucket/my_prefix/2021/08/31/08/53/')")
spark.sql('REFRESH TABLE bad_file')

# And this
spark.read.json('s3://my_bucket/my_prefix/2021/08/31/08/53/my_file.gz').cache().unpersist()

我看到这是一个常见的问题，但大多数人似乎通过刷新表格来解决它。任何帮助将不胜感激。

Answer 1

检查以确保您的 IAM S3 存储桶策略按预期运行。 EMR 可能无权访问该特定存储桶。