Databricks にて Azure IoT Hub の組み込みエンドポイントからデータ取得する方法


## 概要

Databricks にて Azure IoT SDK により Azure IoT Hub の組み込みエンドポイントからのデータ取得する方法を紹介します。本記事で詳細する方法は、検証目的で実行する方法です。

## 事前準備

### 環境構築

- Azure IoT Hub の構築 （本手順は Free tier で実施）
- Databricks （本手順は Serveless ではなく汎用コンピュートで実行）

### デバイスの作成と接続文字列の取得

Azure IoT Hub のリソース画面にて、`デバイス`タブを選択後、`+ デバイスの追加`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/32708e89-0264-4b64-b5f2-494be974a7d4.png)

`デバイス ID`に任意の値を入力し、`保存`を選択してデバイスを作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/37ac238e-cbd3-48d6-82f5-68cf9038354e.png)

作成したデバイスを表示して、`プライマリ接続文字列`の`コピー済み`を選択して値をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4f8d5549-6a72-4ff4-9e10-0e23214201f6.png)

### 組み込みのエンドポイントの接続文字列を取得

Azure IoT Hub のリソース画面にて、`組み込みのエンドポイント`タブを選択して、`イベントハブ互換エンドポイント`の値をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0d7b0ebb-998d-473d-8ee6-cbee3af3bbed.png)

### テレメトリーを送信する方法

Databricks にて azure-iot-device をインストールします。

```python
%pip install azure-iot-device -q
dbutils.library.restartPython()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/60d08c38-421b-4437-b86a-294846eac223.png)

デバイスの接続文字列を変数にセットします。

```python
device_conn_str = "HostName=iot-hub-test-001.azure-devices.net;DeviceId=qiita_test_01;SharedAccessKey=r2fxXUdn/4w9iNJdhqnIJZ+h0qxxxxx="
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0018e890-76b4-49d8-8f00-38d0ffeea171.png)

Azure IoT Hub にテレメトリーを送信するコードを記述します。コードの実行は、次の手順である Azure にてテレメトリーの監視のコマンドを実行してから実行してください。

```python
import os, json, random, time, uuid
from datetime import datetime, timezone
from azure.iot.device import IoTHubDeviceClient, Message


def get_sample_date():
    date = {
        "temperature": round(random.uniform(20, 25), 2),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }

    return json.dumps(date)

# クライアントのインスタンス化
device_client = IoTHubDeviceClient.create_from_connection_string(
    device_conn_str,
    websockets=True,
)

# データの準備
payload = get_sample_date()
msg = Message(payload)
msg.message_id = uuid.uuid4()
msg.content_encoding = "utf-8"
msg.content_type = "application/json"
msg.custom_properties["sensor-location"] = "jp"

# クライアントに接続
device_client.connect()

# テレメトリーの送信
device_client.send_message(msg)
print("✓ sent:", payload)
time.sleep(5)

# クライアントの終了
device_client.shutdown()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e18b64d1-978e-4be1-bca3-a6271b2ccced.png)


Azure ポータルにて下記のコマンドを実行します。Databricks に戻り、テレメトリーの送信を実施して、テレメトリーが受信できることを確認します。

```bash
IOTHUB_NAME="iot-hub-test-001"
az iot hub monitor-events --hub-name "$IOTHUB_NAME" --props all
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a0fc2b21-46a6-4434-9657-4aaff2c1de34.png)

## 組み込みのエンドポイントから受信する方法

Databricks にて azure-eventhub をインストールします。

```python
%pip install azure-eventhub -q
dbutils.library.restartPython()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5f04097c-6444-4e68-9b13-603e8c7673fb.png)

組み込みのエンドポイントの接続文字列を変数にセットします。また、コンシューマーグループを`$Default`で指定します。

```python
eh_conn_str = "Endpoint=sb://ihsuprodblres095dednamespace.servicebus.windows.net/;SharedAccessKeyName=iothubowner;SharedAccessKey=FUum/tiOgEuhyeXXX=;EntityPath=iothub-ehub-iot-hub-te-65629521-b811712f03"
consumer_group = "$Default"
```

下記のコードを実行して、送信済みのテレメトリーを表示できることを確認します。処理が継続するため、セルの実行を停止することを忘れないようにしてください。

```python
import os

os.environ["EH_CONN"] = eh_conn_str
os.environ["CONSUMER_GROUP"] = consumer_group
```

```python
import asyncio, os, json
from azure.eventhub.aio import EventHubConsumerClient



async def on_event(partition_context, event):
    print(f"{event.body_as_str()}")
    await partition_context.update_checkpoint(event)


async def main():
    eh_conn_str = os.getenv("EH_CONN")
    group = os.getenv("CONSUMER_GROUP")
    client = EventHubConsumerClient.from_connection_string(
        eh_conn_str,
        consumer_group=group,
    )
    async with client:
        await client.receive(
            on_event,
            starting_position="-1",   # 検証目的であるため -1 を指定
        )
```

```python
await main()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4e934aa8-9bb3-456d-8da3-1e10cde8e521.png)