EMR中PySpark任务因Parquet文件名变更引发Py4JJavaError问题求助

在EMR上运行PySpark脚本时遭遇如下异常：

py4j.protocol.Py4JJavaError: An error occurred while calling o103.parquet.

排查后确认，该错误是由于构建DataFrame依赖的多个.parquet文件在任务运行期间被更名，导致Spark无法找到文件，错误日志包含类似信息：

Caused by: java.io.FileNotFoundException: No such file or directory 's3://filename.parquet'

已尝试的方案：

• 捕获Py4JJavaError异常并重新创建DataFrame对象
• 创建DataFrame后通过SQL语句强制缓存：

df.createOrReplaceTempView("df")
spark.sql("CACHE TABLE df")

可行解决方案 #

1. 快照式读取固定文件列表 #

由于S3没有文件系统级别的锁机制，读取前先主动获取当前路径下的所有Parquet文件列表，后续仅读取这些固定的文件路径，避免后续文件更名影响读取操作。

示例代码：

import boto3

# 初始化S3客户端
s3_client = boto3.client('s3')
bucket = 'your-bucket-name'
prefix = 'target/path/'

# 遍历获取当前所有Parquet文件（处理分页场景）
file_paths = []
continuation_token = None
while True:
    if continuation_token:
        response = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix, ContinuationToken=continuation_token)
    else:
        response = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix)
    
    for content in response.get('Contents', []):
        key = content['Key']
        if key.endswith('.parquet'):
            file_paths.append(f"s3://{bucket}/{key}")
    
    continuation_token = response.get('NextContinuationToken')
    if not continuation_token:
        break

# 读取固定的文件列表构建DataFrame
df = spark.read.parquet(*file_paths)

2. 确保缓存完全生效 #

之前的缓存操作因Spark的懒执行特性，可能并未真正将数据加载到缓存中，后续操作仍会读取源文件。需触发Action操作强制完成缓存：

示例代码：

from pyspark.storagelevel import StorageLevel

# 读取源数据
df = spark.read.parquet("s3://your-target-path/")
# 指定缓存级别（内存+磁盘，避免内存不足）
df.persist(StorageLevel.MEMORY_AND_DISK)
# 触发Action，强制将数据写入缓存
df.count()

# 后续所有操作直接使用已缓存的DataFrame

也可使用cache()（默认内存级别）替代persist()，但需同样执行count()这类Action触发缓存。

3. 从源端优化文件更新策略 #

如果文件更名是由其他业务流程触发，建议改为原子性更新：

• 先将新文件上传至S3临时路径（如s3://bucket/tmp/）
• 完成上传后，使用S3的原子操作（copy_object+删除旧文件）替换目标路径下的文件
• 或按时间/版本划分前缀（如s3://bucket/data/20240520/），读取时指定固定版本的前缀，避免跨版本文件干扰

4. 启用S3版本控制（可选） #

若需保留文件历史版本，可开启S3桶的版本控制，读取时指定文件的特定版本ID，确保读取的是任务启动时存在的文件版本。示例代码需结合版本ID获取逻辑，读取时指定版本：

# 示例：读取指定版本的文件
df = spark.read.parquet("s3://bucket/path/file.parquet?versionId=your-version-id")

内容的提问来源于stack exchange，提问作者Buklau