如何用PySpark并行将Spark DataFrame中图片内容写入ADLS容器？

这个错误的核心原因是Azure Storage客户端对象（如BlobServiceClient）包含无法序列化的_thread._local成员，而Spark的map操作需要将传递给Worker节点的对象序列化，直接在Driver端初始化客户端再传到map里就会触发序列化失败。

下面是几种可行的解决方案：

方案1：在map函数内部初始化Azure Storage客户端 #

每个Worker节点的任务在处理行数据时自行创建客户端，避免序列化客户端对象：

def consume_row(row):
    from azure.storage.blob import BlobServiceClient
    # 从Databricks Secrets安全获取ADLS连接字符串
    conn_str = dbutils.secrets.get(scope="your-secret-scope", key="adls-conn-str")
    # 初始化客户端
    blob_service_client = BlobServiceClient.from_connection_string(conn_str)
    container_client = blob_service_client.get_container_client("your-target-container")
    # 写入字节内容到ADLS
    container_client.upload_blob(
        name=row.file_name, 
        data=row.file_bytes, 
        overwrite=True
    )
    return row.file_name

# 转换RDD并执行操作
results_rdd = df.rdd.map(consume_row)
# 触发执行（collect仅用于验证，大数据量建议用foreach）
results_rdd.collect()

注意：这种方式每行都会创建一个客户端连接，小数据量可用，大数据量会有性能损耗。

方案2：使用Spark原生binaryFile格式写入（推荐） #

Spark内置了binaryFile格式，专门用于处理二进制文件写入，无需手动管理Azure客户端，完全规避序列化问题：

from pyspark.sql.functions import concat, lit, col

# 构造符合binaryFile要求的DataFrame：需包含path（目标文件路径）和content（字节内容）列
target_df = df.withColumn(
    "path", 
    concat(
        lit("abfss://your-container@your-account.dfs.core.windows.net/images/"), 
        col("file_name")
    )
).select("path", col("file_bytes").alias("content"))

# 写入ADLS
target_df.write.format("binaryFile").mode("overwrite").save()

优势：Spark原生优化，性能最优，代码简洁，无需处理客户端细节。

方案3：用foreachPartition减少客户端创建开销 #

在每个分区内初始化一次客户端，比每行创建更高效，适合大数据量场景：

def process_partition(partition):
    from azure.storage.blob import BlobServiceClient
    # 初始化客户端（每个分区仅执行一次）
    conn_str = dbutils.secrets.get(scope="your-secret-scope", key="adls-conn-str")
    blob_service_client = BlobServiceClient.from_connection_string(conn_str)
    container_client = blob_service_client.get_container_client("your-target-container")
    
    # 处理分区内的所有行
    for row in partition:
        container_client.upload_blob(
            name=row.file_name, 
            data=row.file_bytes, 
            overwrite=True
        )

# 对RDD分区执行处理
df.rdd.foreachPartition(process_partition)

优势：平衡性能与控制，减少连接创建次数，适合大规模数据处理。

额外注意事项 #

• 确保Databricks集群拥有ADLS容器的访问权限（可通过MSI、服务主体或SAS配置）；
• 始终使用dbutils.secrets存储和获取连接字符串，避免硬编码敏感信息；
• 大数据量场景优先选择方案2或3，减少资源开销。

内容的提问来源于stack exchange，提问作者Adrian Arroyo Perez