如何使用Azure Function v2（Python）将Azure Blob中CSV批量插入Azure SQL？

核心修改要点 #

• 替换触发器：将Http触发器改为Blob触发器，监听指定Blob容器下的CSV文件上传事件
• 添加依赖：需要azure-storage-blob读取Blob内容，csv模块解析CSV数据
• 批量数据处理：读取CSV后解析为多行数据，通过SQL输出绑定批量插入

完整代码实现 #

import azure.functions as func
import logging
from azure.functions.decorators.core import DataType
import csv
import io
import uuid

app = func.FunctionApp()

@app.function_name(name="BlobToSqlBatchInsert")
@app.blob_trigger(arg_name="myblob", path="your-container-name/{name}", 
                  connection="AzureWebJobsStorage")
@app.generic_output_binding(arg_name="toDoItems", type="sql", 
                            CommandText="dbo.ToDo", 
                            ConnectionStringSetting="SqlConnectionString",
                            data_type=DataType.STRING)
def blob_trigger(myblob: func.InputStream, toDoItems: func.Out[func.SqlRow]) -> None:
    logging.info(f"Processing blob\nName: {myblob.name}\nSize: {myblob.length} bytes")

    # 读取Blob中的CSV内容
    blob_content = myblob.read().decode('utf-8')
    csv_file = io.StringIO(blob_content)
    csv_reader = csv.DictReader(csv_file)

    # 解析CSV为SqlRow列表
    sql_rows = []
    for row in csv_reader:
        # 自动生成UUID作为Id，CSV列需对应表字段：title、completed、url
        item_id = str(uuid.uuid4())
        sql_row = func.SqlRow({
            "Id": item_id,
            "title": row.get("title", ""),
            "completed": row.get("completed", "False").lower() == "true",
            "url": row.get("url", "")
        })
        sql_rows.append(sql_row)

    # 批量插入到SQL数据库
    if sql_rows:
        toDoItems.set(sql_rows)
        logging.info(f"Successfully inserted {len(sql_rows)} rows into dbo.ToDo")
    else:
        logging.warning("No valid rows found in the CSV file")

配置说明 #

1. 应用设置配置：
   • AzureWebJobsStorage：Azure存储账户的连接字符串，用于Blob触发器的权限验证
   • SqlConnectionString：Azure SQL数据库的连接字符串，需包含插入权限
2. Blob触发器路径：修改path="your-container-name/{name}"中的your-container-name为实际Blob容器名，可添加通配符（如your-container-name/*.csv）仅监听CSV文件
3. 依赖安装：在Function项目根目录创建requirements.txt，添加以下内容：

azure-functions
azure-storage-blob

注意事项 #

• CSV列名需和dbo.ToDo表的字段对应，若列名不匹配需在解析时手动映射字段
• 处理超大CSV时，可改用pandas的分块读取（chunksize参数）避免内存溢出
• 确保Azure Function拥有Blob容器的读取权限和SQL数据库的插入权限

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