前回はIoT HubへのメッセージをAzure Stream Analyticsで受ける方法を調べていきましたが、今回はAzure Functionsで受ける方法を調べていきます。
環境
Windows11, WSL2
❯ cat /etc/lsb-release
DISTRIB_ID=Ubuntu
DISTRIB_RELEASE=22.04
DISTRIB_CODENAME=jammy
DISTRIB_DESCRIPTION="Ubuntu 22.04.2 LTS"
おおまかな流れ
- Azure Functions Core Toolsのインストール
- Azure Functions リソースの用意
- funcコマンドでFunctionプロジェクトの作成
- 出来上がったプロジェクトの設定
- Functionsのローカル起動
Azure Functions Core Toolsのインストール
まずはAzure Functions Core Toolsが必要なので、公式サイトに従ってインストールしましょう。
私の環境の場合はWSL2/Ubuntu なので、以下のリンクの手順でインストールしました。
（マイクロソフトのgpgキーとリポジトリをaptに追加したあとにパッケージインストール）
これでシステム上で func コマンドが使えるようになります。
❯ func --version
4.0.5030
Azure Functions リソースの用意
以前も紹介しました以下の記事の抜粋になります
下記コードでFunctionsリソースとそれに対応するストレージを作成します。
コマンドを実行する前に az login しておくことを忘れずに
#!/bin/sh
STORAGE_NAME="storagename"
RESOURCE_GROUP_NAME="reouce-group-name"
LOCATION="japaneast"
APP_NAME="func-name"
az storage account create --name $STORAGE_NAME --resource-group $RESOURCE_GROUP_NAME
az functionapp create --consumption-plan-location $LOCATION --runtime python --runtime-version 3.10 --functions-version 4 --name $APP_NAME --os-type linux --storage-account $STORAGE_NAME --resource-group $RESOURCE_GROUP_NAME
すると、Functionsリソースが指定のリソースグループに作成されています。
概要を見て、これからAzure Event Hub triggerに対応した関数を作っていくのですが、これはコマンドとエディターで作っていくということで、やり方を確認するのは以下の画面でできます。
funcコマンドでFunctionプロジェクトの作成
上で確認した方法をなぞっていくことになりますが、私の場合はすこしやり方が違う場合もあります（やり方はいくつもあると思いますので、ご自分のやりやすい方法で解決されてください）
適当なディレクトリで func init を実行します。
実際には、func init test-func --python というコマンドを使いました。
これによって、PythonをつかったFunctionsプロジェクトを test-funcというディレクトリに作成することができます。
次に cd test-func した後に、func newします。
すると、下図のように、templateを選ぶ選択肢が出てきますので、Azure Event Hub trigger(8)を選択します。
次に名前の入力を求められるのでiot-hub-conと名付けてみます。
これで、Functionプロジェクトの作成が完了しました。
加えて、pip install azure-functionsでPython側に Azure functions core tools のSDKをインストールしておきましょう。
私の場合はPythonプロジェクトをpoetryで管理していますので、poetry add azure-functionsでインストールを完了しました。
出来上がったプロジェクトの設定
次に作成したプロジェクトの設定ファイルを編集し、実際にIoT Hubと接続してゆきます。
test-func/local.settings.jsonの編集
まずはtest-func/local.settings.jsonを編集していきます
{
"IsEncrypted": false,
"Values": {
"AzureWebJobsStorage": <ストレージ接続文字列>,
"FUNCTIONS_WORKER_RUNTIME": "python",
"IoTHubConnectionString": <イベントハブ互換エンドポイント>
}
}
上記の通り、ここではストレージ接続文字列とイベントハブ互換エンドポイントが必要です。
これらを取得する方法を以下に説明します。
ストレージ接続文字列の取得
ストレージ接続文字列は、Functionsリソースを作るときにFunctionsリソースに紐付けるために作ったあのストレージの接続文字列です。
以下の場所で取得できます。
イベントハブ互換エンドポイントの取得
イベントハブ互換エンドポイントは、IoT Hubの組み込みエンドポイントから取得できます。
下図の通りです。
test-func/iot-hub-con/function.jsonの編集
{
"scriptFile": "__init__.py",
"bindings": [
{
"type": "eventHubTrigger",
"name": "events",
"direction": "in",
"eventHubName": <Iot Hubリソースの名前>,
"connection": "IoTHubConnectionString",
"cardinality": "many",
"consumerGroup": "$Default"
}
]
}
ここで変更が必要なものはeventHubNameとconnectionの2つです。
eventHubNameは作成したIoT Hubのリソースの名前をそのまま入力します。
connectionの"IoTHubConnectionString"という文字列は、local.settings.jsonの
"IoTHubConnectionString": <イベントハブ互換エンドポイント> 部分を設定項目名を使って参照する形になっています。
Functionsのローカル起動
ここまで終わったら func start で作成したアプリを実行します。
poetryの場合はpoetry run func startです。
するとローカルのアプリがIoT Hubと接続されて、IoT Hubデータの待ち受けが開始されます。
この状態で、IoT Hubに対してmessageを送信すると、__init__.pyの内容どおり、それを出力する内容が表示されるはずです。
import logging
from azure.functions import EventHubEvent
from typing import List
def main(events: List[EventHubEvent]):
for event in events:
logging.info('Python EventHub trigger processed an event: %s',
event.get_body().decode('utf-8'))
無事メッセージを受け取ると下記のような表示が得られます。
