企业官网建设流程全解析

探究 Azure Functions 函数绑定：为 IoT 定位数据配置 Blob 存储输出绑定（IoT-For-Beginners 运输项目第 2 课作业）

【免费下载链接】IoT-For-Beginners12 Weeks, 24 Lessons, IoT for All!项目地址: https://gitcode.com/GitHub_Trending/io/IoT-For-Beginners

本篇技术指南以《IoT-For-Beginners》开源课程「运输（Transport）」项目第 2 课《存储定位数据》的课后作业《探究函数绑定（Investigate function bindings）》为核心展开。该作业要求学习者理解 Azure Functions 中「触发器 + 输入/输出绑定」的声明式机制，并尝试在function.json中为函数的main返回值配置 Blob 存储输出绑定，使 GPS 遥测数据能自动落盘到 Azure Blob 存储。读完本文，你将掌握函数绑定的概念体系、仓库中 Event Hub 触发器的真实配置写法、输出绑定的配置要点，以及它与本课实际采用的 Python SDK 写法之间的差异与取舍。

作业背景：从 IoT Hub 事件到 Blob 存储的"最后一公里"

在本课主文档 3-transport/lessons/2-store-location-data/README.md 中，完整的数据链路是这样的：

1. 物联网设备（Wio Terminal / Raspberry Pi / 虚拟设备）每分钟读取一次 GPS 坐标，以 JSON 格式发送到 Azure IoT Hub；
2. 一个基于 Python 运行时、名为gps-trigger的 Azure Functions 应用，通过Event Hub 兼容端点触发器监听 IoT Hub 中的遥测事件；
3. 函数代码将每条事件加工后写入Blob 存储（容器名为gps-data），供下一课在地图上可视化车辆行驶轨迹。

{ "gps" : { "lat" : <latitude>, "lon" : <longitude> } }

本课主文档使用的是「在函数代码内部调用 Python SDK 显式上传 blob」的方案；而本次作业则反其道而行之：要求学习者去阅读 Azure Functions 绑定文档，搞清楚如何利用输出绑定（output binding），让函数直接把main的返回值交给 Azure Functions 运行时自动写入 Blob 存储——不写一行存储调用代码。

函数绑定：从"函数参数即配置"理解 Azure Functions

函数绑定（function bindings）是 Azure Functions 的声明式数据接入/接出机制，核心思想是：函数与外部服务的连接方式不需要写死在代码里，而是以 JSON 声明在function.json文件中。它分为两类：

• 输入绑定（input binding）：把外部数据（如队列消息、IoT Hub 事件、Blob 内容）注入为函数的参数；
• 输出绑定（output binding）：把函数的返回值（或指定的func.Out对象）自动写出到外部服务（如 Blob 存储、队列、Cosmos DB）。

每个绑定都由几个固定属性描述，这也是本次作业需要重点掌握的：

对于 Python 函数而言，输入绑定会作为参数传入main，输出绑定则对应func.Out[type]参数；当输出绑定绑定到函数的返回值时，只需让main返回对应数据即可。

仓库中的真实绑定配置：读懂function.json

课程配套代码的触发器配置就在 code/functions/gps-trigger/iot-hub-trigger/function.json：

{ "scriptFile": "__init__.py", "bindings": [ { "type": "eventHubTrigger", "name": "events", "direction": "in", "eventHubName": "samples-workitems", "connection": "IOT_HUB_CONNECTION_STRING", "cardinality": "many", "consumerGroup": "$Default", "dataType": "binary" } ] }

对照上面的属性表逐项解析这份真实配置：

• type: "eventHubTrigger"——这是一个触发器（特殊的输入绑定），监听 Event Hub 兼容端点。IoT Hub 内置了这样的端点，因此 Azure Functions 可以直接订阅 IoT 设备的遥测消息；
• name: "events"——事件以events参数注入函数，对应init.py 中的签名def main(events: List[func.EventHubEvent])；
• direction: "in"——数据流入方向；
• connection: "IOT_HUB_CONNECTION_STRING"——注意这里存放的是键名而非连接字符串本身，真实值保存在 local.settings.json 的Values.IOT_HUB_CONNECTION_STRING中，部署到云端后则放到函数的 Application Settings；
• cardinality: "many"——一次批量接收多条事件，因此main的参数类型是List[func.EventHubEvent]；
• consumerGroup: "$Default"——使用 IoT Hub 默认的消费组；
• dataType: "binary"——事件体按二进制处理，所以代码中需要用event.get_body().decode('utf-8')还原为文本。

对应的main函数签名与事件处理代码：

def main(events: List[func.EventHubEvent]): for event in events: logging.info('Python EventHub trigger processed an event: %s', event.get_body().decode('utf-8')) device_id = event.iothub_metadata['connection-device-id'] blob_name = f'{device_id}/{str(uuid.uuid1())}.json' # ...

从源码结构可以看到：当前实现并未使用输出绑定，而是借助事件元数据（iothub_metadata）拿到设备 ID，再构造出设备ID/随机UUID.json的 blob 路径。这正是本课作业引导你对比的两种方案之一。

本次作业核心：为返回值配置 Blob 存储输出绑定

作业要求原文指出："函数绑定允许你的代码通过从main函数返回值的方式，把 blob 保存到 Blob 存储；Azure 存储账户、容器（collection）以及其他细节均在function.json中配置。"

因此在function.json的bindings数组中追加一个direction: "out"、type: "blob"的绑定，即可实现"返回即落盘"。一个符合作业要求的典型配置形如：

{ "scriptFile": "__init__.py", "bindings": [ { "type": "eventHubTrigger", "name": "events", "direction": "in", "eventHubName": "samples-workitems", "connection": "IOT_HUB_CONNECTION_STRING", "cardinality": "many", "consumerGroup": "$Default", "dataType": "binary" }, { "type": "blob", "name": "outputBlob", "direction": "out", "path": "gps-data/{rand-guid}.json", "connection": "STORAGE_CONNECTION_STRING" } ] }

配置要点说明：

• type: "blob"+direction: "out"声明这是一个Blob 存储输出绑定；
• path中gps-data是容器名（与本课 SDK 方案使用的容器名保持一致），{rand-guid}是 Azure Functions 运行时自动生成的随机 GUID——这正是本课 SDK 方案中用 Pythonuuid.uuid1()手工生成的等价物；
• connection: "STORAGE_CONNECTION_STRING"指向 local.settings.json 中的存储账户连接字符串条目，作业要求将此键加入本地配置；
• 当输出绑定与函数返回值绑定时，main只需返回要写入的 blob 内容（此处即 JSON 文档），运行时负责上传。

配置完成后，用func azure functionapp fetch-app-settings（云端）或直接运行本地 Functions 应用，即可观察：每条来自 IoT Hub 的 GPS 事件会自动以独立 JSON 文件的形式出现在gps-data容器中，无需在代码里调用任何上传方法。

输出绑定与 SDK 写法的取舍：为什么本课主文档选择后者

本课主文档《存储定位数据》在"连接你的无服务器代码到存储"一节中明确给出了两种方式：

1. 在函数代码内部使用 Blob 存储 Python SDK 写数据——本课采用的方式；
2. 使用输出函数绑定，将函数返回值绑定到 Blob 存储，由运行时自动保存——本次作业要探究的方式。

两种方式在仓库中都有迹可循。SDK 方式的核心逻辑在 code/functions/gps-trigger/iot-hub-trigger/init.py：

def get_or_create_container(name): connection_str = os.environ['STORAGE_CONNECTION_STRING'] blob_service_client = BlobServiceClient.from_connection_string(connection_str) for container in blob_service_client.list_containers(): if container.name == name: return blob_service_client.get_container_client(container.name) return blob_service_client.create_container(name, public_access=PublicAccess.Container)

这段代码值得细读：

• os.environ['STORAGE_CONNECTION_STRING']从环境变量读取连接字符串——本地运行时来自local.settings.json的Values，云端则来自 Application Settings；
• Python Blob SDK没有"不存在则创建"的容器辅助方法，因此这里手动遍历list_containers()判断容器gps-data是否存在，存在则复用其ContainerClient，不存在则调用create_container(name, public_access=PublicAccess.Container)新建；
• PublicAccess.Container授予容器级别的公共读权限——课程特意为之，以便下一课在地图上直接读取 blob 中的 GPS 数据完成可视化。

随后的上传逻辑把设备 ID、入队时间与 GPS 坐标组装成与输出绑定等价的 JSON 文档：

event_body = json.loads(event.get_body().decode('utf-8')) blob_body = { 'device_id' : device_id, 'timestamp' : event.iothub_metadata['enqueuedtime'], 'gps': event_body['gps'] } logging.info(f'Writing blob to {blob_name} - {blob_body}') blob.upload_blob(json.dumps(blob_body).encode('utf-8'))

为什么课程采用 SDK 而作业要求你探究输出绑定？从源码对比中可以看出各自的价值：SDK 方案控制力最强，能自由地遍历容器、按需设置访问权限、拼接任意 blob 路径（本课用设备 ID 作为文件夹前缀）；而输出绑定方案更"声明式"，配置集中在function.json，函数体零存储代码，特别适合"一条事件对应一个 blob"的简单场景，也更容易被复用和维护。理解两者的边界，正是本次作业的设计意图。

需要特别强调的是：时间戳应使用消息的入队时间（enqueuedtime）而非函数执行时的当前时间。主文档中特别提示，如果 Functions 应用没有持续运行，消息可能在 IoT Hub 中滞留一段时间才被拾取——用入队时间才能还原设备真实的遥测时刻。这一点无论采用哪种写入方式都成立，也是验证作业成果时容易被忽略的细节。

配套配置：让本地运行与云端部署都走得通

无论选择输出绑定还是 SDK 方案，以下配套文件都是本地联调的关键，全部收录在仓库 code/functions/gps-trigger/ 目录下：

1.local.settings.json—— 本地密钥与运行环境

{ "IsEncrypted": false, "Values": { "FUNCTIONS_WORKER_RUNTIME": "python", "AzureWebJobsStorage": "UseDevelopmentStorage=true", "IOT_HUB_CONNECTION_STRING": "<connection string>", "STORAGE_CONNECTION_STRING": "<connection string>" } }

其中AzureWebJobsStorage: "UseDevelopmentStorage=true"配合Azurite（本地存储模拟器）使用，让函数在本地即可读写存储；IOT_HUB_CONNECTION_STRING指向 IoT Hub 的 Event Hub 兼容端点（带Endpoint=...;SharedAccessKeyName=...的完整连接字符串）；STORAGE_CONNECTION_STRING则是用az storage account show-connection-string --output table --name <storage_name>获取的存储账户连接字符串。部署到云端后，这两个值需要原样上传为 Functions 应用的 Application Settings。

2.requirements.txt—— 依赖清单

azure-functions azure-storage-blob

使用 SDK 方案必须安装azure-storage-blob（本次作业探究的输出绑定方案在 Python 3.x 上则主要依赖azure-functions与扩展包）。若安装报错，可先pip install --upgrade pip升级 pip 再重试。

3.host.json—— 运行时与扩展

{ "version": "2.0", "extensionBundle": { "id": "Microsoft.Azure.Functions.ExtensionBundle", "version": "[2.*, 3.0.0)" } }

Blob 存储绑定依赖扩展包（Extension Bundle）提供，[2.*, 3.0.0)声明使用 2.x 系列的扩展捆绑，本地开发时由 Azure Functions Core Tools 自动解析，无需手工安装扩展。

评测量表：如何检验你的作业完成度

作业附带的评测标准（Rubric）把完成度分为三档，正好可以当作自测清单：

对照此表自查的三步法：

1. 能否配置——function.json中direction: "out"的 blob 绑定是否正确加入、connection键是否与local.settings.json中的STORAGE_CONNECTION_STRING对应；
2. 能否返回——main函数的返回路径是否与绑定参数（name）对齐，返回内容是否为合法的 JSON 文档；
3. 能否落盘——用az storage blob list --container-name gps-data --account-name <storage_name> --account-key <key1> --output table能否在容器中看到新增的 JSON blob，且内容结构符合{"device_id": ..., "timestamp": ..., "gps": {"lat": ..., "lon": ...}}。

需要提醒的是，验证时不要同时运行az iot hub monitor-events监控命令——它会与函数应用争抢 IoT Hub 的消费组事件，导致函数收不到消息。

小结

本次作业表面上只是"读文档、配绑定"，实质上让你打通了 Azure Functions 的两条数据通路：一条是仓库中已经实现的SDK 显式写入（__init__.py中BlobServiceClient手动建容器、拼路径、上传 blob），另一条是作业要求探究的声明式输出绑定（在function.json中声明blob+direction: "out"，main返回即自动落盘）。把仓库里的function.json、local.settings.json、host.json与__init__.py放在一起对照阅读，你就能同时掌握"触发器如何把 IoT Hub 事件注入函数"与"输出绑定如何把函数返回值写出到存储"这两块核心拼图，为后续将 Functions 应用部署到云端、以及在地图上可视化 GPS 轨迹打下坚实基础。

【免费下载链接】IoT-For-Beginners12 Weeks, 24 Lessons, IoT for All!项目地址: https://gitcode.com/GitHub_Trending/io/IoT-For-Beginners