Python 如何有效地读取存储在 pyspark 数据框列中的路径

目前，我只使用RDD API中的mapPartitions，并使用GCS python客户端手动执行批处理读取.这是我想出来的:

from pyspark.sql import Row
from google.cloud import storage
from urllib.parse import urlparse
import itertools

            
def _get_bucket(row):
    return urlparse(row.image_path).netloc

def _get_path_beneath_bucket(row):
    return urlparse(row.image_path).path.strip("/")
            
def process_partition(rows):
    client = storage.Client()
    
    # group by bucket, so we can efficiently read one bucket at a time
    bucket_blobs = itertools.groupby(rows, key=_get_bucket)
    
    for bucket_name, rows in bucket_blobs:
        bucket = client.get_bucket(bucket_name)
        # eval iterator, since we'll need it twice
        rows = list(rows)
        
        with client.batch():
            blobs = [bucket.blob(_get_path_beneath_bucket(row)) for row in rows]
            for blob in blobs:
                blob.reload()
        
        for row, blob in zip(rows, blobs):
            yield Row(image_path=row.image_path, image_blob=blob.download_as_bytes())

然后，我使用如下函数:

rdd = df.rdd.mapPartitions(process_partition)
result_df = rdd.toDF()

这不是很理想，部分原因是我不得不离开钨运行时来访问RDD API(我猜这将以一些性能代价结束)，部分原因是它有点混乱.

如果你知道更好的方法，我很乐意接受不同的解决方案.:)