Azure Data Factory+Batch Service+Python：无法追加Blob存储CSV分块数据

核心原因 #

1. 本地文件追加逻辑不兼容ADLS对象存储：本地用mode='a'是基于流式文件系统的追加，但ADLS属于对象存储体系，普通Pythonopen()操作ABFS路径时，mode='a'无法实现真正的末尾追加，每次写入都会重新创建对象或覆盖已有内容，最终只保留最后一次写入的分块。
2. Batch任务执行环境的隔离性：如果分块处理在Batch的不同节点/进程并行执行，多个进程同时写入同一ABFS路径时，会因为对象存储的强一致性导致互相覆盖，甚至生成多个独立的分块文件。

解决办法 #

方法一：本地生成完整文件后再上传 #

先在Batch节点的本地临时磁盘完成所有分块的追加拼接，再一次性上传到ADLS，完全复用本地的追加逻辑：

from azure.storage.filedatalake import DataLakeServiceClient
import pandas as pd
import os

# 定义Batch节点本地临时文件路径
local_temp_path = "/tmp/full_processed_data.csv"

# 分块处理并追加到本地文件
for chunk in your_data_chunks:
    # 替换为你的数据转换逻辑
    processed_chunk = chunk.apply(your_transform_function)
    # 首次写入带表头，后续追加跳过表头
    processed_chunk.to_csv(local_temp_path, mode='a', header=not os.path.exists(local_temp_path), index=False)

# 上传完整文件到ADLS
storage_conn_str = "你的存储账户连接字符串"
service_client = DataLakeServiceClient.from_connection_string(storage_conn_str)
fs_client = service_client.get_file_system_client(file_system="你的容器名")
file_client = fs_client.get_file_client("目标存储路径/full_data.csv")

with open(local_temp_path, "rb") as f:
    file_client.upload_data(f, overwrite=True)

方法二：使用ADLS追加Blob特性 #

针对ADLS Gen2的Blob存储层，用Azure SDK的追加Blob API实现分块追加：

from azure.storage.blob import BlobServiceClient, AppendBlobClient
import pandas as pd
import io

storage_conn_str = "你的存储账户连接字符串"
append_blob_client = BlobServiceClient.from_connection_string(storage_conn_str).get_append_blob_client(
    container="你的容器名", blob="目标存储路径/full_data.csv"
)

# 不存在则创建空的追加Blob
if not append_blob_client.exists():
    append_blob_client.create_append_blob()

for chunk in your_data_chunks:
    processed_chunk = chunk.apply(your_transform_function)
    # 将处理后的分块转为字节流
    chunk_buffer = io.StringIO()
    processed_chunk.to_csv(chunk_buffer, header=False, index=False)
    chunk_bytes = chunk_buffer.getvalue().encode('utf-8')
    # 追加到Blob
    append_blob_client.append_block(chunk_bytes)

方法三：统一Batch任务执行环境 #

如果之前是并行执行分块任务，调整Batch作业配置，让所有分块处理在同一个节点的同一个进程中运行，避免多进程写入冲突，再结合上述两种方法完成写入。

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