如何在Azure Function中将pandas DataFrame写入Azure Blob Storage

步骤1：配置输出Blob绑定 #

你需要先修改函数目录下的function.json文件，新增Blob输出绑定配置，示例配置如下：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "inputBlob",
      "type": "blobTrigger",
      "direction": "in",
      "path": "你的输入容器名/{name}",
      "connection": "你的存储连接字符串配置名"
    },
    // 新增以下输出绑定配置
    {
      "name": "outputBlob",
      "type": "blob",
      "direction": "out",
      "path": "你的输出容器名/processed_{name}", // 可自定义输出路径和文件名
      "connection": "你的存储连接字符串配置名"
    }
  ]
}

参数说明：name对应Python函数里的输出参数名，path是输出Blob的存储路径，connection填你Azure Function配置里的存储连接字符串的键名，和输入绑定使用同一个配置也可以。

步骤2：修改Python函数代码 #

修改后的完整代码如下：

import logging
import pandas as pd
import io
import azure.functions as func

# 新增outputBlob参数，和function.json里的输出绑定name对应
def main(inputBlob: func.InputStream, outputBlob: func.OutputStream):
    logging.info(f"Python blob trigger function processed blob \n"
                 f"Name: {inputBlob.name}\n"
                 f"Blob Size: {inputBlob.length} bytes")

    df = pd.read_csv(io.BytesIO(inputBlob.read()), sep='#', encoding='unicode_escape', header=None, names=range(16))
    logging.info(df.head())

    # 将DataFrame转为CSV格式的字节流
    output = io.StringIO()
    df.to_csv(output, index=False, encoding='utf-8') # index=False表示不写入行索引列
    output_bytes = output.getvalue().encode('utf-8')

    # 写入到输出Blob
    outputBlob.set(output_bytes)
    logging.info(f"处理完成，结果已写入输出Blob")

关键说明 #

• 不需要额外安装Azure存储SDK，输出绑定已经封装了写入逻辑，直接操作outputBlob参数即可
• 如果需要保留原文件的分隔符、编码格式，调整df.to_csv的sep、encoding参数即可
• 输出文件名可以在function.json的path里自定义规则，支持动态参数，比如用输入Blob的名称拼接前缀后缀

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