PySpark如何将单条数据作为JPG文件写入S3或HDFS

优先推荐分区复用客户端直写S3方案，实现简单、运维成本低，性能完全满足百万级数据的导出需求；超大规模场景可选择HDFS+distcp方案。

方案1：foreachPartition复用客户端直写S3（优先推荐） #

原方案问题 #

原foreach逻辑每条数据都创建boto3客户端、重建HTTP连接，无用开销占比超过90%，是运行过慢的核心原因。

优化逻辑 #

使用foreachPartition按分区初始化资源，每个分区仅创建1次boto3客户端、1次requests会话，大幅降低连接开销。

实现代码 #

import boto3
import requests
from botocore.config import Config

def process_img_partition(partition_data):
    # 每个分区仅初始化一次S3客户端和请求会话
    s3_config = Config(
        retries = {"max_attempts": 3, "mode": "standard"},
        max_pool_connections = 50
    )
    s3_client = boto3.client("s3", config=s3_config)
    req_session = requests.Session()
    s3_bucket = "test"
    s3_key_prefix = "out_images/"

    for item in partition_data:
        # 从当前行数据中取图片ID和下载地址，按实际数据结构调整字段名
        cid = item["cid"]
        img_url = item["img_url"]
        try:
            resp = req_session.get(img_url, stream=True, timeout=10)
            if resp.status_code == 200:
                s3_client.put_object(
                    Body = resp.content,
                    Bucket = s3_bucket,
                    Key = f"{s3_key_prefix}{cid}.jpg"
                )
        except Exception as e:
            # 单条图片处理失败仅打印日志，不会中断整个分区任务
            print(f"图片{cid}处理失败: {str(e)}")

# 调整RDD分区数，建议每个分区处理1000-5000张图片，避免分区过多或过大
myRdd = myRdd.repartition(200)
myRdd.foreachPartition(process_img_partition)

性能优化点 #

• 分区数调整：可根据总数据量动态调整，100万条数据建议设为200-500个分区
• 大文件优化：单张图片超过10MB可开启boto3多分片上传配置
• 并发控制：可结合线程池在单个分区内并行下载+上传，进一步提升效率

方案2：HDFS + s3-dist-cp同步（超大规模场景推荐） #

原方案报错原因 #

Python内置open方法不支持直接写入hdfs://协议路径，需要调用HDFS专用的客户端接口完成写入。

实现步骤 #

1. 所有EMR节点安装HDFS Python依赖：sudo pip3 install pyarrow
2. 分区写入HDFS临时目录

import requests
from pyarrow import fs

def write_hdfs_partition(partition_data):
    # 初始化HDFS客户端，按实际集群配置调整host和port
    hdfs_client = fs.HadoopFileSystem(host="localhost", port=9000)
    req_session = requests.Session()
    hdfs_temp_path = "/data/img_temp/"

    for item in partition_data:
        cid = item["cid"]
        img_url = item["img_url"]
        try:
            resp = req_session.get(img_url, stream=True, timeout=10)
            if resp.status_code == 200:
                file_path = f"{hdfs_temp_path}{cid}.jpg"
                with hdfs_client.open_output_stream(file_path) as f:
                    f.write(resp.content)
        except Exception as e:
            print(f"图片{cid}写入HDFS失败: {str(e)}")

myRdd.foreachPartition(write_hdfs_partition)

3. 所有图片写入完成后，在EMR主节点执行distcp同步到S3：

s3-dist-cp --src hdfs:///data/img_temp --dest s3://test/out_images/

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