PySpark 读取 Amazon S3 文件报错 "File not present on S3" 的解决方法

2024-10-16 11:40:19

在使用 PySpark 处理存储在 Amazon S3 上的数据时，你可能会遇到 "File not present on S3" 的错误。这个错误提示通常出现在你尝试使用 foreach() 方法对 S3 文件进行操作，例如复制或删除时。这个问题的出现，主要与 Spark 读取 S3 文件的机制以及 S3 本身的最终一致性特性有关。

Spark 如何读取 S3 文件

当 Spark 需要读取 S3 上的文件时，它首先会获取文件的元数据信息，包括文件大小、修改时间等等。Spark 利用这些元数据信息来规划任务的执行，例如决定如何将文件分割成不同的数据块，以及如何将这些数据块分配给不同的执行器。

但是，如果在 Spark 获取元数据信息之后，S3 上的文件发生了变化，比如被删除或者被修改了，那么 Spark 在执行任务时就会发现文件找不到了，于是就抛出了 "File not present on S3" 的错误。

S3 的最终一致性

S3 是一种具有最终一致性特性的存储服务。这意味着，当你对 S3 上的文件执行操作，例如复制或者删除文件时，S3 并不能保证这些操作会立即生效。换句话说，你的代码可能已经执行完了 copy 操作，但是 S3 仍然需要一些时间才能真正完成文件的复制过程。如果 Spark 在这段时间内尝试读取文件，那么它就可能找不到文件，从而导致错误的发生。

解决 "File not present on S3" 错误的几种方法

针对这个问题，我们可以采取以下几种解决方案：

1. 强制 Spark 使用一致性读取

我们可以通过修改 Hadoop 的配置参数，强制 Spark 在读取 S3 文件时始终获取最新的元数据信息，而不是使用缓存中的信息。具体来说，我们需要在 SparkSession 中设置 fs.s3a.consistent.metadata 参数为 true。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("S3FileCopy") \
    .config("fs.s3a.consistent.metadata", "true") \
    .getOrCreate()

这个配置会告诉 Spark，每次读取 S3 文件时都要去 S3 获取最新的元数据信息，确保读取到的是最新的文件状态。

2. 引入延迟机制

在执行文件复制或者删除操作之后，我们可以引入一小段时间的延迟，给 S3 足够的时间来完成操作。例如，我们可以使用 time.sleep() 函数：

import time

# ... 其他代码 ...

def move_files_in_s3(self, row):
    # ... 复制文件代码 ...

    self.client.delete_object(
        Bucket=self.bucket_name,
        Key=row.source_file_path
    )

    time.sleep(5)  # 延迟 5 秒

延迟时间的长短需要根据实际情况进行调整。

3. 使用 Spark 的 DataFrame 操作

foreach() 方法是在每个分区上执行一个函数，它并不保证操作的顺序和原子性。如果你的操作需要保证顺序和原子性，建议使用 Spark 的 DataFrame 操作，例如 write.parquet() 或者 write.csv()。

例如，你可以将文件路径存储在一个 DataFrame 中，然后使用 DataFrameWriter 将文件复制到目标路径：

from pyspark.sql.functions import input_file_name, regexp_replace

# 读取源文件
df = spark.read.json("s3a://{}/{}".format(self.args['SOURCE_S3_BUCKET'], self.args['SOURCE_S3_PATH']))

# 获取源文件路径
df = df.withColumn("source_file_path", input_file_name())

# 构造目标文件路径
df = df.withColumn("destination_file_path", 
                    regexp_replace("source_file_path", 
                                    self.args['SOURCE_S3_PATH'], 
                                    self.args['DESTINATION_S3_PATH']))

# 将文件复制到目标路径
df.write.parquet("s3a://{}/{}".format(self.args['SOURCE_S3_BUCKET'], self.args['DESTINATION_S3_PATH']))

# 删除源文件 (可选)
# ...