如何通过Python Azure函数高效批量插入上万条数据至Azure存储表

针对你遇到的批量插入速度慢、持久性差的问题，结合Azure存储表的特性，我来给你梳理下最优的实现方案，顺便帮你排查之前方法的问题：

先分析你之前方法的问题 #

1. 逐个插入：每个实体单独发起网络请求，开销极大，1万条数据至少要1万次请求，慢是必然的。
2. 进程池：Azure函数的沙箱环境（尤其是消费计划）对多进程有严格限制，确实没法用，换成线程池就没问题了。
3. 事务提交：你可能没注意到Azure表事务的硬性要求——所有操作必须属于同一个PartitionKey，而且每批最多100个操作。如果你的操作跨了分区，事务根本没法生效，速度自然上不去。
4. 旧SDK的TableBatch：你用的是已经被弃用的azure-storage-table SDK里的TableBatch，性能和稳定性都不如最新的azure-data-tables SDK，而且代码里没处理最后一批不足100条的情况，还缺少异常重试逻辑，这可能是持久性问题的根源。

最优实现方案（基于最新SDK） #

核心思路 #

• 用最新的azure-data-tables SDK，API更高效稳定；
• 按PartitionKey分组，同分区的实体批量提交事务（每批最多100个）；
• 用线程池并行处理不同分区的插入（Azure函数允许线程池，不受沙箱限制）；
• 添加异常捕获和重试，确保数据持久性。

具体代码实现 #

首先安装最新SDK：

pip install azure-data-tables

然后编写批量插入代码：

from azure.data.tables import TableClient, TableTransactionAction
from concurrent.futures import ThreadPoolExecutor
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
import os
import logging

# 开启日志，方便排查问题
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 给事务添加重试逻辑，处理网络波动等临时问题
@retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter(jitter=1))
def submit_batch(table_client, operations, partition_key):
    try:
        table_client.submit_transaction(operations)
        logger.info(f"Successfully inserted {len(operations)} entities for partition {partition_key}")
    except Exception as e:
        logger.error(f"Failed to insert batch for partition {partition_key}: {str(e)}")
        raise  # 抛出异常触发重试

def bulk_insert_partition(table_client, partition_key, entities):
    batch_size = 100
    total = len(entities)
    # 分批次处理，包括最后一批不足100条的情况
    for start_idx in range(0, total, batch_size):
        end_idx = min(start_idx + batch_size, total)
        batch_entities = entities[start_idx:end_idx]
        
        # 构建事务操作列表
        operations = [
            TableTransactionAction(
                action_type="create",
                entity={
                    "PartitionKey": partition_key,
                    "RowKey": str(e["rowkey"]),
                    "someKey": e["someValue"],
                    "someOtherKey": e["someOtherValue"]
                }
            )
            for e in batch_entities
        ]
        
        # 提交批次（带重试）
        submit_batch(table_client, operations, partition_key)

def main():
    # 从环境变量获取存储连接字符串（Azure函数里可以配置应用设置）
    connection_string = os.getenv("AZURE_STORAGE_CONNECTION_STRING")
    table_name = "your-target-table"
    
    # 初始化TableClient，用with确保资源正确释放
    with TableClient.from_connection_string(connection_string, table_name) as table_client:
        # 确保表存在，不存在则创建
        table_client.create_table_if_not_exists()
        
        # 假设这是你的1万条待插入数据
        dataToStore = [
            {"rowkey": i, "someValue": f"data-{i}", "someOtherKey": f"meta-{i}"} 
            for i in range(10000)
        ]
        
        # 按PartitionKey分组（如果你的数据有多个分区，这里会自动分组）
        partition_groups = {}
        for item in dataToStore:
            # 这里如果你的数据有自定义PartitionKey，就替换成item["your-partition-key-field"]
            pk = "PARTITION1"
            if pk not in partition_groups:
                partition_groups[pk] = []
            partition_groups[pk].append(item)
        
        # 用线程池并行处理不同分区的插入（线程数根据分区数调整，避免过多）
        max_workers = min(len(partition_groups), 8)
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = []
            for pk, entities in partition_groups.items():
                futures.append(executor.submit(bulk_insert_partition, table_client, pk, entities))
            
            # 等待所有任务完成，确保所有数据都插入
            for future in futures:
                future.result()

if __name__ == "__main__":
    main()

关键优化点说明 #

1. 事务批量提交：每100个同分区实体提交一次事务，把1万次请求压缩到100次（如果是单分区），极大减少网络开销。
2. 线程池并行：不同分区的插入可以并行执行，充分利用Azure存储表的分区并行写入能力，进一步提升速度。
3. 重试机制：用tenacity库实现指数退避重试，处理网络波动导致的临时失败，确保数据不会丢失。
4. 完整的批次处理：循环逻辑会自动处理最后一批不足100条的情况，不会遗漏数据。
5. 最新SDK：azure-data-tables是官方推荐的SDK，性能和维护性都远优于旧的azure-storage-table。

额外建议 #

• 如果你的数据有多个PartitionKey，尽量均匀分配，这样并行插入的效率会更高；
• 在Azure函数里，建议把存储连接字符串配置到应用设置里，不要硬编码；
• 如果数据量超过10万条，可以考虑用Azure Data Factory做批量导入，但1万条用上面的代码完全足够；
• 可以通过日志查看每个批次的耗时，排查是否有其他性能瓶颈（比如数据生成速度慢）。

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