如何将Amazon EMR PySpark生成的动态命名输出文件从Amazon S3读取到Flask应用

我来帮你搞定这个问题！EMR用PySpark输出的CSV都是动态命名的part文件，确实给Flask读取带来了小麻烦，不过有几个实用的方案，结合你的场景我整理了最靠谱的思路：

方案1：直接读取S3前缀下所有符合条件的CSV文件（推荐大数据量场景） #

这是最稳妥且高效的方式——不用纠结文件名，直接读取目标路径下所有part-*.csv文件，忽略自动生成的_SUCCESS标记文件。你可以用s3fs配合pandas或者Dask来实现，代码非常简洁：

用pandas + s3fs读取并合并 #

import pandas as pd
import s3fs

# 初始化S3文件系统（用默认AWS凭证即可，无需额外配置）
fs = s3fs.S3FileSystem()

# 匹配所有part开头的CSV文件，排除_SUCCESS文件
csv_files = fs.glob('s3://my-bucket/output/part-*.csv')

# 读取所有文件并合并成一个DataFrame
dfs = [pd.read_csv(f's3://{file}') for file in csv_files]
combined_df = pd.concat(dfs, ignore_index=True)

# 之后就可以在Flask里用combined_df做业务处理了

如果你的数据量很大，推荐用Dask代替pandas，它能并行处理文件，内存压力小很多：

import dask.dataframe as dd

# 直接读取整个前缀下的CSV，自动匹配所有part文件
dask_df = dd.read_csv('s3://my-bucket/output/part-*.csv')

# 转换为pandas DataFrame（数据量能放进内存时），或直接用Dask做计算
combined_df = dask_df.compute()

方案2：PySpark写入时合并为单个文件（适合小数据集） #

如果你的数据集不大，可以让PySpark把所有数据合并到一个分区，生成单个part文件，之后再重命名成固定名称。注意：coalesce(1)会把所有数据拉到一个节点，大数据量会导致内存溢出，谨慎使用！

步骤1：PySpark合并写入临时路径 #

# 合并到1个分区，写入临时目录
df.coalesce(1).write.mode('overwrite').csv('s3://my-bucket/output_temp')

步骤2：重命名为固定文件名（用boto3实现） #

在Flask里或者EMR的后续步骤中，把临时目录里的part文件重命名：

import boto3

s3 = boto3.client('s3')
bucket_name = 'my-bucket'
temp_prefix = 'output_temp/'

# 列出临时目录下的part文件
response = s3.list_objects_v2(Bucket=bucket_name, Prefix=temp_prefix)
part_file = [obj['Key'] for obj in response['Contents'] if obj['Key'].endswith('.csv')][0]

# 重命名到目标路径
s3.copy_object(
    Bucket=bucket_name,
    CopySource={'Bucket': bucket_name, 'Key': part_file},
    Key='output/final_result.csv'
)

# 删除临时目录
s3.delete_object(Bucket=bucket_name, Key=temp_prefix)

方案3：EMR步骤后自动合并重命名（适合批量处理场景） #

如果你的EMR任务是定时或批量运行的，可以在EMR的步骤列表里加一个Shell步骤，直接用AWS CLI把所有part文件合并成固定名称的CSV：

# 合并所有part文件并上传到固定路径
aws s3 cp s3://my-bucket/output/part-*.csv - | aws s3 cp - s3://my-bucket/output/final_result.csv

这样任务跑完后，S3里就有一个固定名称的final_result.csv，Flask直接读取这个文件就行。

高效实现建议 #

• 大数据量优先选方案1，利用s3fs的glob特性直接读取所有part文件，避免合并带来的性能损耗；
• 小数据集可以用方案2，但要注意分区合并的性能风险；
• 批量任务场景用方案3，把重命名逻辑放到EMR步骤里，Flask端完全不用处理动态文件名问题。

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