python-3.x - Azure 组合器函数，用于使用 Python 接收数据并将其写入 Azure Blob 存储

Question

我想使用 Python 创建一个 Azure 函数，该函数将从 Azure 事件中心读取数据。幸运的是，Visual Studio Code 提供了一种创建 Azure 函数骨架的方法。可以根据需要进行编辑。我能够在 Microsoft 文档的帮助下创建演示 HTTP 触发器 Azure 函数，但我不知道应该在以下函数中进行哪些更改，以便它可以从事件中心读取数据并将其写入 Azure Blob 存储。另外，如果有人可以引用建议任何博客来获取有关 azure 功能和标准实践的更多详细信息。

更新:

我尝试根据@Stanley的建议更新我的代码，但可能需要更新代码。我在我的 Azure 函数中编写了以下代码。

local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "Storage account connection string",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "EventHub_ReceiverConnectionString": "Endpoint Connection String of the EventHubNamespace",
    "Blob_StorageConnectionString": "Storage account connection string"
  }
}

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "function",
      "type": "eventHubTrigger",
      "direction": "in",
      "name": "event",
      "eventHubName": "pwo-events",
      "connection": "EventHub_ReceiverConnectionString",
      "cardinality": "many",
      "consumerGroup": "$Default",
      "dataType": "binary"
    }
  ]
}

init.py

import logging
import azure.functions as func
from azure.storage.blob import BlobClient

storage_connection_string='Storage account connection string'
container_name = ''


def main(event: func.EventHubEvent):
    logging.info(f'Function triggered to process a message: {event.get_body().decode()}')
    logging.info(f'  SequenceNumber = {event.sequence_number}')
    logging.info(f'  Offset = {event.offset}')
    blob_client = BlobClient.from_connection_string(storage_connection_string,container_name,str(event.sequence_number) + ".txt")
    blob_client.upload_blob(event.get_body().decode())

以下是我的 blob 容器的屏幕截图:

执行上面的代码后，某些内容被写入 blob 容器中。但它不是以 txt 文件保存，而是以其他格式保存。另外，如果我多次触发 azure 函数，那么文件将被覆盖。我想执行追加操作而不是覆盖。另外，我想将我的文件保存在用户定义的位置。示例:容器/年=/月=/日期=谢谢!!

Answer 1

如果要从 Azure 事件中心读取数据，请使用 event hub trigger会容易得多，这是我的测试代码(读取数据并写入存储):

import logging
import azure.functions as func
from azure.storage.blob import BlobClient
import datetime

storage_connection_string=''
container_name = ''

today = datetime.datetime.today()


def main(event: func.EventHubEvent):
    logging.info(f'Function triggered to process a message: {event.get_body().decode()}')
    logging.info(f'  SequenceNumber = {event.sequence_number}')
    logging.info(f'  Offset = {event.offset}')

    blob_client =  BlobClient.from_connection_string(
        storage_connection_string,container_name,
    str(today.year) +"/" + str(today.month) + "/" + str(today.day) + ".txt")

blob_client.upload_blob(event.get_body().decode(),blob_type="AppendBlob")

我使用下面的代码将事件发送到事件中心:

import asyncio
from azure.eventhub.aio import EventHubProducerClient
from azure.eventhub import EventData

async def run():
    # Create a producer client to send messages to the event hub.
    # Specify a connection string to your event hubs namespace and
    # the event hub name.
    producer = EventHubProducerClient.from_connection_string(conn_str="<conn string>", eventhub_name="<hub name>")
    async with producer:
        # Create a batch.
        event_data_batch = await producer.create_batch()

        # Add events to the batch.
        event_data_batch.add(EventData('First event '))
        event_data_batch.add(EventData('Second event'))
        event_data_batch.add(EventData('Third event'))

        # Send the batch of events to the event hub.
        await producer.send_batch(event_data_batch)

loop = asyncio.get_event_loop()
loop.run_until_complete(run())

我的local.settings.json:

{
    "IsEncrypted": false,
    "Values": {
        "AzureWebJobsStorage": "<storage account conn str>",
        "FUNCTIONS_WORKER_RUNTIME": "python",
        "testhubname0123_test_EVENTHUB": "<event hub conn str>"
    }
}

我的function.json 就像 this doc表示:

{
    "scriptFile": "__init__.py",
    "bindings": [{
        "type": "eventHubTrigger",
        "name": "event",
        "direction": "in",
        "eventHubName": "test01（this is my hubname, pls palce yours here）",
        "connection": "testhubname0123_test_EVENTHUB"
    }]
}

结果

使用上面的代码运行该函数并将数据发送到事件中心:

数据已成功存入存储:

下载.txt并查看其内容可以看到写入了3个事件内容: