Spark：如何将单个行以 JPG 格式写入 S3/HDFS

Question

我必须将数据作为单个 JPG 文件（约数百万）从 PySpark 写入 S3 存储桶。

我尝试了多种选择：

设置：AWS EMR 集群和 Jupyter 笔记本。

1. 在 'foreach' 方法中创建一个 boto3 客户端并写入 S3 ==> 太慢且效率低下，因为我们为每个任务打开客户端。

def get_image(y):
    res = requests.get(img_url, stream=True)
    file_name = "./" +str(cid) + ".jpg"
    client = boto3.client('s3')
    file_name = str(cid) + ".jpg"
    client.put_object(Body=res.content, Bucket='test',  Key='out_images/'+file_name)

myRdd.foreach(get_image)

1. 写入本地文件系统并运行“aws S3 复制”到 S3 => 如果将这些数据写入每个单独的工作节点的卷，则不清楚如何访问这些数据。在作业运行时登录到工作节点，但无法准确找到 JPG 的写入位置。

def get_image(y):
    res = requests.get(img_url, stream=True)
    file_name = "./" +str(cid) + ".jpg"
    with open(file_name, 'wb') as f:
        f.write(res.content)

myRdd.foreach(get_image)

1. 写入 HDFS 并稍后运行 s3-dist-cp。可能是最有效的，但尚未在代码方面取得成功。 I get path cannot be found exceptions

def get_image(y):
    res = requests.get(img_url, stream=True)
    file_name = "hdfs://" +str(cid) + ".jpg"
    with open(file_name, 'wb') as f:
        f.write(res.content)

myRdd.foreach(get_image)

有人可以提出一个实现这一目标的好方法吗？

Answer 1

如果将 foreach 替换为 foreachPartition，则解决方案 1 效果很好。此更改后，每个分区仅创建一个客户端：

def get_image(y_it):
    client = boto3.client('s3')
    for y in y_it:
        img_url = ...
        cid = ...
        res = requests.get(img_url, stream=True)
        file_name = str(cid) + ".jpg"
        client.put_object(Body=res.content, Bucket='test',  Key='out_images/'+file_name)

myRdd.foreachPartition(get_image)

在y_it 的循环内，重复使用了同一个客户端。

如果将requests.Sessions 用于this answer 中所述的http 调用，事情甚至会变得更快。在这种情况下，会在循环外通过y_it（如客户端）创建一个 http 会话，然后在循环内重复使用。