Azure IoT Hub から Azure Storage にメッセージルーティンした device-to-cloud メッセージ(テレメトリー)を Databricks で参照する方法


## 概要

Azure IoT Hub から Azure Storage にメッセージルーティンした device-to-cloud メッセージ(`DeviceMessages`データソース)を Databricks で参照する方法を紹介します。

## 事前準備

### 環境構築

- Azure IoT Hub の構築 （本手順は Free tier で実施）
- Azure Storage の構築（Databricks 環境から接続できるように設定）
- Databricks （本手順は Serveless ではなく汎用コンピュートで実行）

### Azure IoT Hub のマネージド ID に対して Azure Storage の`ストレージ BLOB データ共同作成者`権限を付与

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/47ce17a1-6416-4587-a9b9-813e11262efa.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2d220b77-128b-4003-9d1d-c13c432b6b3a.png)

### Azure IoT Hub のマネージド ID に対して Azure Event Hubs の`Azure Event Hubs のデータ送信者`権限を付与

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/47ce17a1-6416-4587-a9b9-813e11262efa.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/78dcb6a2-a1b5-47ef-ae0c-cffd006c43ad.png)

### Azure IoT Hub にてカスタムエンドポイントを作成

Azure portal では作成する画面はないため、メッセージ ルーティングを作成するに同時に作成できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/97c1802c-81a2-49c6-8fea-9ece7064b8d8.png)

### Azure IoT Hubs にてメッセージルーティンを設定

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1f17cfd5-fd5d-4a38-8eef-40a5e611aa97.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/24e9acb5-631a-4dd6-9174-f9ae85fc110c.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1e972d06-bc46-4d6b-b641-1b0c4a17e46e.png)

### Azure IoT Hub にてカスタムエンドポイントを作成

Azure portal では作成する画面はないため、メッセージ ルーティングを作成するに同時に作成できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/97c1802c-81a2-49c6-8fea-9ece7064b8d8.png)

### Azure IoT Hubs にてメッセージルーティンを設定

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c9eaf6a9-4b38-4cce-86c1-5a0fd09e279a.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bc73f5ee-99b8-4f14-abdb-c1ecca6aef06.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3107fd26-1634-44b2-bddf-a225de0eb490.png)

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/68298e55-1012-42d5-8bdf-cddf36ca0138.png)


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/454f5851-6312-480d-84e5-347244dd4186.png)


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b300bb26-0593-4a93-a1d8-17a3c4cf957c.png)

#### モジュールの作成と接続文字列の取得

作成済みのデバイスの画面に、`+ モジュール ID の追加`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dce6cec7-23da-4660-be75-eeebeb3a24d6.png)

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/17c8b90e-239d-4222-b0c9-24a773bd5be4.png)

作成したモジュールを表示して、`プライマリ接続文字列`値をコピーします。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3e71649c-afee-43c9-9118-f0015c17764c.png)


## メッセージルーティンの実施

### Databricks にてライブラリのインストール

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

### デバイスから device-to-cloud メッセージ(テレメトリー)の送信

```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/2091debc-8879-4ba8-ab33-28480f37bb21.png)


### モジュールから device-to-cloud メッセージ(テレメトリー)の送信


```python
module_conn_str = "HostName=iot-hub-test-001.azure-devices.net;DeviceId=qiita_test_01;ModuleId=qiita_test_module_01;SharedAccessKey=s5svMAgEglFjcr923J+GfDcXXXXX="

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)

# クライアントのインスタンス化
module_client = IoTHubDeviceClient.create_from_connection_string(
    module_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"

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

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7bbddad9-da77-4bda-bea1-e26434eea3b6.png)

## Databricks におけるデータの読み込み

### Azure Storage からデータを取得

Azure Storage からデータを取得します。

```python
storage_account = "iothubstorage123"
container       = "qiitatest"

abfs_path = f"abfss://{container}@{storage_account}.dfs.core.windows.net/"
abfs_path += "iot-hub-test-0001/*/*/*/*/*/*"
```

```python
src_df = (
    spark.read
         .format("avro")
         .load(abfs_path)
)
src_df.printSchema()
src_df.display()
```

```python
src_df.printSchema()
src_df.display()
```


```text
root
 |-- EnqueuedTimeUtc: string (nullable = true)
 |-- Properties: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)
 |-- SystemProperties: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)
 |-- Body: binary (nullable = true)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5933c1cb-a775-4963-bdb9-b54404e48f89.png)


### `body`列に対する decode の実施

```python
from pyspark.sql.functions import expr

df = src_df.withColumn(
    "Body",
    expr("decode(Body, 'UTF-8')")
)
df.display()
```

```json
{
	"temperature": 21.82,
	"timestamp": "2025-05-27T04:22:28.003265+00:00"
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b8b5c54d-f614-4745-b131-a9bfc2ad751a.png)


## ＜補足＞Azure Storage に連携されるデータ

### デバイスによる device-to-cloud メッセージ(テレメトリー)のデータ

#### Properties

```text
sensor-location: "jp"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c71bdccc-fdc7-465a-98ba-ed1a8a5a7911.png)


#### SystemProperties

```text
connectionDeviceGenerationId: "638839157113307051"
connectionAuthMethod: "{\"scope\":\"device\",\"type\":\"sas\",\"issuer\":\"iothub\"}"
connectionDeviceId: "qiita_test_device_01"
contentType: "application/json"
messageId: "2ad60f28-54ba-4b42-bf4e-19e580ddb6e0"
contentEncoding: "utf-8"
enqueuedTime: "2025-05-27T04:22:28.2750000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0093ca24-9d20-400c-85c3-ab39921f3e5b.png)


#### body

```json
{
	"temperature": 21.82,
	"timestamp": "2025-05-27T04:22:28.003265+00:00"
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f7bb6483-49c9-4b20-a858-839d53cbce07.png)


### モジュールによる device-to-cloud メッセージ(テレメトリー)のデータ

#### Properties
```json
sensor-location: "jp"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e8f9caa5-3c00-47ad-98ba-c19c758dd41f.png)

#### SystemProperties

```text
connectionModuleId: "qiita_test_module_01"
connectionDeviceGenerationId: "638839157225497198"
connectionAuthMethod: "{\"scope\":\"module\",\"type\":\"sas\",\"issuer\":\"iothub\"}"
connectionDeviceId: "qiita_test_device_01"
contentType: "application/json"
messageId: "f4ccdf19-01a3-4df7-912f-7aa4cfbfe4e7"
contentEncoding: "utf-8"
enqueuedTime: "2025-05-27T04:22:33.9320000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8a9f856b-dcfb-4eb9-ac4b-f1add44a147c.png)


#### body

```json
{
	"temperature": 24.08,
	"timestamp": "2025-05-27T04:22:33.702755+00:00"
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8f29e72c-9358-47bf-8baf-8824fac70032.png)