Azure IoT Hub から Azure Storage にメッセージルーティンしたデバイス接続状態イベントを Databricks で参照する方法

## 概要

Azure IoT Hub から Azure Storage にメッセージルーティンしたデバイス接続状態イベント（`DeviceConnectionStateEvents`データソース）を 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 にてカスタムエンドポイントを作成

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


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9b85923b-2d3c-4d36-9641-3dae811eb378.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/27f47cc1-8c51-41f6-8e64-0c768e2a3811.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)

#### Databricks にてデバイスの接続と切断の実施

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

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

```python
# クライアントのインスタンス化
from azure.iot.device import IoTHubDeviceClient

device_client = IoTHubDeviceClient.create_from_connection_string(
    device_conn_str,
    websockets=True,
)
```

```python
# デバイスでの接続

device_client.connect()
```

```python
import time
time.sleep(30)

# デバイスでの切断
device_client.shutdown()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b04e71d4-2494-403a-8b7f-7d2982e090c2.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)


### Azure Storage にてファイルを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/046168a2-dcef-441d-91c4-e4be6c574836.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/c19c451c-43da-436b-8091-7623f116eb23.png)


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

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

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

```json
{"sequenceNumber":"000000000000000001DBCE0D753FFC4B0000000200000000000000000000000A"}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/edade970-9b7c-4033-ac82-90f6b843ac84.png)


:::note info

SystemProperties の userId がエンコードされたまま連携されていることを確認しました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/818bc702-af2c-4470-97cd-327a3b3ccbb2.png)

下記のコードでデコードできます。

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

df = df.withColumn(
    "userId",
    expr("decode(unbase64(SystemProperties['userId']), 'UTF-8')"),
)
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e7611751-f3e6-43ec-bed7-bbc1363c7867.png)

:::



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

### デバイス接続時のデータ

#### Properties

```text
operationTimestamp: "2025-05-27T03:45:13.6146405Z"
iothub-message-schema: "deviceConnectionStateNotification"
opType: "deviceConnected"
deviceId: "qiita_test_device_id_01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0988cba1-c0a0-4c0a-b56a-32bad9146849.png)

#### SystemProperties

```text
connectionDeviceId: "qiita_test_device_id_01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "90a3dd21-7d54-946a-15ff-f878d916d59d"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T03:45:19.2990000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f3b225ed-a9ef-49f8-ac25-0557363cb468.png)

#### body

```json
{"sequenceNumber":"000000000000000001DBCE0D753FFC4B0000000200000000000000000000000A"}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a86bb15e-ab01-48d3-a4a9-ff4597024f8f.png)

### デバイス切断時のデータ

#### Properties

```text
operationTimestamp: "2025-05-27T03:45:54.4356635Z"
iothub-message-schema: "deviceConnectionStateNotification"
opType: "deviceDisconnected"
deviceId: "qiita_test_device_id_01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/511a6be6-d56e-4236-908d-84bb20c75d71.png)

#### SystemProperties

```text
connectionDeviceId: "qiita_test_device_id_01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "7aae80e5-7f80-8a62-f9f4-be414b2251e5"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T03:46:18.3660000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/442cca61-7381-48c9-8695-07357df225f7.png)

#### body

```json
{"sequenceNumber":"000000000000000001DBCE0D753FFC4B0000000200000000000000000000000B"}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/04cc5c15-1c74-4d78-af32-9fe6c98d9644.png)

### モジュール接続時のデータ

#### Properties

```text
moduleId: "qiita_test_module_01"
operationTimestamp: "2025-05-27T03:46:04.6701497Z"
iothub-message-schema: "deviceConnectionStateNotification"
opType: "deviceConnected"
deviceId: "qiita_test_device_id_01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/53f90e56-4a26-442c-8560-766e4b79f45a.png)

#### SystemProperties

```text
connectionModuleId: "qiita_test_module_01"
connectionDeviceId: "qiita_test_device_id_01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "d56c53b0-b997-99be-71ca-1a2262e399c8"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T03:46:18.3500000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f5a4fb22-a450-4662-9eb4-ff29789e9ad3.png)

#### body

```json
{"sequenceNumber":"000000000000000001DBCE0D753FFC4B0000000200000000000000000000000C"}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bf6c25dd-2e33-4411-a372-2786d5729873.png)


### モジュール切断時のデータ

#### Properties

```text
moduleId: "qiita_test_module_01"
operationTimestamp: "2025-05-27T03:46:37.9046815Z"
iothub-message-schema: "deviceConnectionStateNotification"
opType: "deviceDisconnected"
deviceId: "qiita_test_device_id_01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2af160c5-2692-418d-946a-138a4eae06d2.png)

#### SystemProperties

```text
connectionModuleId: "qiita_test_module_01"
connectionDeviceId: "qiita_test_device_id_01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "fec1cfa2-c409-de65-01ec-38b20a125935"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T03:47:18.4030000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c17bb493-d0f3-4de5-8fb3-abade4af17e3.png)


#### body

```json
{"sequenceNumber":"000000000000000001DBCE0D753FFC4B0000000200000000000000000000000D"}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b4674276-c369-4ce7-ad33-e5653cba4006.png)