Azure Functionsをつかう その２：Azure Stream Analytics(ASA)と連動する

IoTHubと連動する前に、時系列データの取り扱いが得意なStream AnalyticsとFunctionを連動する方法を調べてみます。

実際のところ

Functionは直接Azure Stream Analytics(ASA)にデータをおくる事ができないので
Azure Event Hubsを咬ませてから実行します。

Azure Event Hubsのキー生成

スクリプト本体

次のような動作をするスクリプトは以下の通り。

• HTTPリクエストに入っているJSONからValue(0~254の整数が入っている）を引っ張り出す
• その時点のタイムスタンプをくっつける
• ASAにEvent Hubs経由でおくる

import logging
import azure.functions as func
import json
import datetime
from azure.eventhub import EventHubProducerClient, EventData
import os
import json

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTPトリガー関数がリクエストを処理しました。')

    # HTTPリクエストからJSONを解析します
    try:
        req_body = req.get_json()
    except ValueError:
        pass
    else:
        # JSONからValueを取得します
        value = req_body.get('Value')

        # タイムスタンプを作成します
        timestamp = datetime.datetime.now().isoformat()

        # Valueとタイムスタンプを統合した単一のJSONを作成します
        data = {
            'Value': value,
            'TimeStamp': timestamp,
        }
        data = json.dumps(data)

        # JSONからEvent Hubsの接続文字列とハブ名を取得します
        with open('config.json') as config_file:
            config = json.load(config_file)
        event_hub_connection_str = config["EventHubsConnectionString"]
        event_hub_name = config["EventHubName"]

        # データをAzure Event Hubsに送信します
        producer = EventHubProducerClient.from_connection_string(event_hub_connection_str, event_hub_name)
        event_data_batch = producer.create_batch()
        event_data_batch.add(EventData(data))
        producer.send_batch(event_data_batch)

        return func.HttpResponse("Event Hubsにデータを送信しました。")

ASA側の受け入れ設定

• Azureポータルにログインし、Stream Analyticsジョブを作成
• "入力"をクリックし、Event Hubsの文字列を指定
• "クエリ"を設定し受け取とった内容の処理を決定。今回は脳死全保存でいくので

SELECT * INTO Output FROM Input

• "出力"を追加し、データの保存先を指定。色々選べる*1が、SQL Databaseにする。

参考もと

learn.microsoft.com

*1:Blob Storage、SQL Database、Cosmos DB、Event Hub、Table Storage、Service Bus Queue、Service Bus Topic