17370845950

本文介绍一种内存友好的方式：通过按字节范围分块下载 azure blob 中的大 csv 文件，并在写入时直接追加到单一文件中，避免因截断行导致的数据损坏或解析错误。

在处理超大 CSV 文件（如数 GB 级）从 Azure Blob Storage 下载时，一次性加载到内存不仅低效，还极易触发 MemoryError。虽然分块下载（download_blob(offset=..., length=...)）能有效缓解内存压力，但若将每块单独保存为独立 CSV 文件（如 c

hunk_1.csv, chunk_2.csv），后续合并将面临跨块行断裂（partial line）这一关键风险——例如某一行数据恰好被切分在两个相邻 chunk 的边界处，此时简单拼接二进制内容或调用 pd.concat() 都会破坏 CSV 结构，导致列错位、引号失配甚至解析失败。

✅ 正确解法是：只打开一个目标文件，以二进制写入模式（"wb"）初始化，然后逐块读取、连续写入。由于 Azure Blob 存储的 download_blob() 返回的是按指定字节范围精确提取的原始字节流，只要原始 CSV 文件本身是合法且连续存储的（无编码混杂、无 BOM 干扰），这种“流式拼接”即可 100% 还原原始文件字节顺序，包括换行符和所有转义字符。

以下是优化后的完整实现：

import os
from azure.storage.blob import BlobServiceClient

# 假设 blob_client 已正确初始化
dest_path = os.path.join(dest_folder, "merged_output.csv")

blob_properties = blob_client.get_blob_properties()
blob_size = blob_properties['size']
chunk_size = 100 * 1024 * 1024  # 100 MB

with open(dest_path, "wb") as f:  # ✅ 单次打开，全程写入
    for i in range((blob_size + chunk_size - 1) // chunk_size):
        start_offset = i * chunk_size
        end_offset = min(start_offset + chunk_size, blob_size)
        length = end_offset - start_offset

        # 下载当前 chunk 的原始字节
        blob_data = blob_client.download_blob(offset=start_offset, length=length)
        f.write(blob_data.readall())  # ✅ 直接追加，无中间文件

print(f"✅ 下载完成：{dest_path}（{blob_size:,} 字节）")

⚠️ 注意事项：

• 不要使用文本模式（"w"）或添加换行符：CSV 是二进制格式，必须用 "wb" 模式写入原始字节，否则可能因平台换行符转换（如 \r\n → \n）或编码隐式解码引发损坏。
• 确保 blob 内容为纯 CSV 且无前置/后置元数据：该方法假设整个 blob 就是目标 CSV 文件本体。若 blob 包含 HTTP 头、ZIP 封装或 Base64 编码等，需先解包或解码。
• 推荐配合进度提示与异常重试：生产环境建议加入 tqdm 进度条、max_retries 参数及 try/except 块，提升鲁棒性。
• 无需额外“合并”步骤：此方案本质是流式重建，最终输出即为结构完整、可直接用 pandas.read_csv() 或任何 CSV 工具加载的合法文件。

总结：分块下载的核心价值在于控制内存占用，而保持数据完整性不依赖于“合并逻辑”，而在于严格维持原始字节流顺序。通过单文件流式写入，我们既规避了内存瓶颈，又彻底消除了行断裂风险——这才是处理大型文本 Blob 的最佳实践。