优化Azure函数中Blob存储多CSV文件合并的Python代码

原代码通过将所有CSV加载为DataFrame再合并的方式，存在内存占用高、不必要的序列化开销等问题，在处理大量或大体积CSV时会拖慢执行速度，还可能导致Azure函数因内存不足触发超时或扩容，增加成本。以下是针对性的优化方案，兼顾性能与成本控制：

一、流式合并（无Pandas，内存友好首选） #

直接流式读取每个CSV文件的内容，仅保留当前行在内存，仅在第一个文件写入表头，后续文件跳过表头后直接追加内容。这种方式内存占用极低，适合任意大小的CSV文件，Azure函数可选用更低配置的实例，显著降低成本。

核心优化点： #

• 复用ContainerClient，减少连接建立开销
• 直接操作字节流，避免StringIO和文本转换的额外开销
• 仅写入一次表头，避免重复表头
• 支持超大文件的逐行处理，不会触发内存瓶颈

from azure.identity import DefaultAzureCredential
from azure.storage.blob import ContainerClient

# 使用托管身份直接认证，无需从Key Vault获取连接字符串（更安全高效）
credential = DefaultAzureCredential()

# 初始化容器客户端
storage_account_url = "https://yourstorageaccount.blob.core.windows.net"
source_container = ContainerClient(
    account_url=storage_account_url,
    container_name="sourceContainerName",
    credential=credential
)
target_container = ContainerClient(
    account_url=storage_account_url,
    container_name="targetContainerName",
    credential=credential
)

# 筛选所有CSV格式的Blob
csv_blobs = [blob for blob in source_container.list_blobs() if blob.name.endswith(".csv")]

# 初始化目标Blob
target_blob_client = target_container.get_blob_client("combinedCSV.csv")
try:
    target_blob_client.delete_blob()
except Exception:
    pass

# 流式合并CSV
wrote_header = False
for blob in csv_blobs:
    with source_container.get_blob_client(blob.name) as source_blob:
        with source_blob.download_blob() as download_stream:
            lines = download_stream.readlines()
            if not wrote_header:
                # 第一个文件写入完整内容（包含表头）
                target_blob_client.append_block(b''.join(lines))
                wrote_header = True
            else:
                # 后续文件跳过第一行表头，写入剩余内容
                if len(lines) > 1:
                    target_blob_client.append_block(b''.join(lines[1:]))

二、Pandas优化方案（需DataFrame处理场景） #

如果业务逻辑依赖DataFrame进行数据清洗、转换等操作，可通过以下方式优化内存占用与执行速度：

核心优化点： #

• 分块读取大CSV，避免一次性加载全量数据
• 边读边合并写入，无需存储所有DataFrame到列表
• 指定列类型减少内存消耗
• 直接写入Blob块，避免生成完整CSV字符串

from azure.identity import DefaultAzureCredential
from azure.storage.blob import ContainerClient
import pandas as pd
from io import BytesIO

credential = DefaultAzureCredential()
storage_account_url = "https://yourstorageaccount.blob.core.windows.net"
source_container = ContainerClient(
    account_url=storage_account_url,
    container_name="sourceContainerName",
    credential=credential
)
target_container = ContainerClient(
    account_url=storage_account_url,
    container_name="targetContainerName",
    credential=credential
)

csv_blobs = [blob for blob in source_container.list_blobs() if blob.name.endswith(".csv")]

# 初始化目标Blob
target_blob_client = target_container.get_blob_client("combinedCSV.csv")
try:
    target_blob_client.delete_blob()
except Exception:
    pass

# 分块合并CSV
wrote_header = False
for blob in csv_blobs:
    with source_container.get_blob_client(blob.name) as source_blob:
        download_stream = BytesIO(source_blob.download_blob().readall())
        # 按10000行分块读取，可根据内存配置调整chunksize
        for chunk in pd.read_csv(download_stream, chunksize=10000, low_memory=False):
            output = chunk.to_csv(index=False, sep=',', header=not wrote_header).encode("utf-8")
            target_blob_client.append_block(output)
            if not wrote_header:
                wrote_header = True

三、额外性能&成本优化建议 #

• 异步Blob操作：使用Azure Storage异步API结合Azure函数异步触发器，提升并发处理速度，减少等待时间
• 精准过滤Blob：通过list_blobs(name_starts_with="指定目录/")仅遍历目标目录下的CSV，减少遍历开销
• 批量操作：对大量CSV采用批量获取Blob列表的方式，减少API调用次数
• 实例选型：流式方案内存占用极低，可选用Consumption计划的小实例；超大文件场景可临时调高内存配置，处理完成后恢复

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