如何用Azure Function编辑Azure Blob Storage中的CSV文件（替代DataFactory）

我完全懂你的困扰——DataFactory后台启动集群来做这么简单的CSV处理确实大材小用，用Azure Function做轻量处理才是更高效的选择。你当前的函数只是做了Blob复制，接下来咱们把它改成能精准筛选指定列、按自定义顺序输出的版本。

完整实现代码 #

直接替换你的__init__.py内容即可：

import logging
import azure.functions as func
import csv
from io import StringIO

def main(myblob: func.InputStream, outputblob: func.Out[func.InputStream]):
    logging.info(f"Python blob trigger function processed blob \n"
                f"Name: {myblob.name}\n"
                f"Blob Size: {myblob.length} bytes")

    # 读取Blob中的CSV内容，处理UTF-8编码（含BOM的文件用utf-8-sig解码）
    csv_content = myblob.read().decode('utf-8-sig')
    input_io = StringIO(csv_content)

    # --------------------------
    # 这里替换成你实际需要的列名，按目标顺序排列
    target_columns = ["user_id", "order_date", "total_amount"]
    # --------------------------

    # 初始化CSV读写器
    reader = csv.DictReader(input_io)
    output_io = StringIO()
    writer = csv.DictWriter(output_io, fieldnames=target_columns)

    # 写入表头和筛选后的行数据
    writer.writeheader()
    for row in reader:
        # 只保留目标列的内容，自动按设定顺序排列
        filtered_row = {col: row[col] for col in target_columns if col in row}
        writer.writerow(filtered_row)

    # 将处理后的内容写入输出Blob
    outputblob.set(output_io.getvalue().encode('utf-8'))

关键细节说明 #

• 流处理适配：Azure Blob的InputStream不能直接传给csv模块，所以用StringIO把字节流转换成内存中的字符串流，模拟本地文件操作，这是解决你之前无法访问编辑数据的核心。
• 编码兼容：用utf-8-sig解码是为了兼容Windows生成的带BOM的CSV文件，避免表头出现\ufeff这类奇怪字符；输出时再转成标准UTF-8字节流写入Blob。
• 列控制：通过target_columns列表定义你需要保留的列和顺序，DictReader按列名读取数据，DictWriter按指定顺序写入，完美实现你的需求。
• 容错处理：if col in row的判断可以避免源CSV中缺失目标列时抛出错误，缺失的列会自动留空。

本地调试提示 #

在VS Code中调试时，你可以在本地项目的input-raw文件夹下放置测试CSV文件，启动Azure Function本地调试，处理后的文件会生成在同目录下（带-copy后缀），方便你验证结果。

你的现有Blob绑定配置不需要修改，保持当前的bindings设置即可正常触发和输出。

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