使用Python Azure函数存储在BLOB存储中的Process CSV文件

发布于 2025-02-11 05:57:36 字数 151 浏览 1 评论 0原文

我有一个CSV文件，其中存储在Blob存储中的记录数。我想阅读要添加到CSV文件中的新记录，对其进行处理，然后将其存储回BLOB存储中的另一个容器。我想使用Python Azure函数实现此流程。我无法在Python Azure功能中编写用于入站和出站的代码。

请帮忙。谢谢

原文

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

煞人兵器 2025-02-18 05:57:36

以下是对我有用的代码。我正在使用HTTP触发器来满足您的要求。

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "blob",
      "direction":"in",
      "name": "inputblob",
      "path": "input/input.csv",
      "connection": "storageacc_STORAGE"
    },
    {
      "type": "blob",
      "direction":"out",
      "name": "outputblob",
      "path": "output/output.csv",
      "connection": "storageacc_STORAGE"
    }
  ]
}

init .py

import csv
import logging

import azure.functions as func


def main(req: func.HttpRequest, inputblob: func.InputStream,outputblob: func.Out[bytes]) -> func.InputStream:
    
    # Reading from the input binding
    input_file = inputblob.read()
    
    # Processing the csv file

    # Writing to output binding
    outputblob.set(input_file)

local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "storageacc_STORAGE": "<Your_Connection_String>"
  }
}

结果：

” vue7w.png“ rel =” nofollow noreferrer“>

Below is the code that worked for me. I'm using HTTP Trigger to achieve your requirement.

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "blob",
      "direction":"in",
      "name": "inputblob",
      "path": "input/input.csv",
      "connection": "storageacc_STORAGE"
    },
    {
      "type": "blob",
      "direction":"out",
      "name": "outputblob",
      "path": "output/output.csv",
      "connection": "storageacc_STORAGE"
    }
  ]
}

init.py

import csv
import logging

import azure.functions as func


def main(req: func.HttpRequest, inputblob: func.InputStream,outputblob: func.Out[bytes]) -> func.InputStream:
    
    # Reading from the input binding
    input_file = inputblob.read()
    
    # Processing the csv file

    # Writing to output binding
    outputblob.set(input_file)

local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "storageacc_STORAGE": "<Your_Connection_String>"
  }
}

RESULTS: