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

## 概要

Azure IoT Hub から Azure Storage にメッセージルーティンしたデバイス ライフサイクル イベント（`DeviceLifecycleEvents`データソース）を 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/c0079908-a15f-4774-8285-21213363b435.png)


## デバイス ライフサイクル イベントのメッセージルーティンの実施

### Device の作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6f088738-2826-46fa-989d-0785ddde6005.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/faaf0f54-6705-4d5e-b8ee-2e194e5f49fc.png)

### Module の作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3ae668d3-5ad3-4890-a4ca-48e1622972da.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3d31a745-0c68-4350-8531-a92e3a85c35f.png)

### Module の削除

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/14a53e62-8b46-47d1-92f4-04bff59588bf.png)

### Device の削除

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f6be2fff-c6bd-4f09-b272-0fbcbf0784c0.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/e38624fc-ffcf-4bbd-b1af-2b8c53f8a34d.png)


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

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

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

```json
{
	"deviceId": "qiita-test-device-01",
	"etag": "AAAAAAAAAAE=",
	"version": 2,
	"properties": {
		"desired": {
			"$metadata": {
				"$lastUpdated": "2025-05-27T01:40:32.8491207Z"
			},
			"$version": 1
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "2025-05-27T01:40:32.8491207Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4f5ecc48-96cb-4865-ac4d-063f4993fc67.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/555a7bc1-8dd6-4f1d-b90f-2ad0145eaa97.png)


:::



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

### デバイス作成時のデータ

#### Properties

```text
operationTimestamp: "2025-05-27T01:40:32.8491207Z"
iothub-message-schema: "deviceLifecycleNotification"
opType: "createDeviceIdentity"
deviceId: "qiita-test-device-01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/37f8ebaf-48f1-4120-9d4b-854212befa99.png)

#### SystemProperties

```text
twin-version: "2"
connectionDeviceId: "qiita-test-device-01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "2d63cf127e9"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T01:40:32.9790000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e1aac37c-258c-4ef2-8cf2-be4bc5587918.png)

#### body

```json
{
	"deviceId": "qiita-test-device-01",
	"etag": "AAAAAAAAAAE=",
	"version": 2,
	"properties": {
		"desired": {
			"$metadata": {
				"$lastUpdated": "2025-05-27T01:40:32.8491207Z"
			},
			"$version": 1
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "2025-05-27T01:40:32.8491207Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8c52ab25-23e5-49cc-986a-d1cf5e3e8994.png)


### デバイス削除時のデータ


#### Properties

```text
operationTimestamp: "2025-05-27T01:40:53.2559728Z"
iothub-message-schema: "deviceLifecycleNotification"
opType: "deleteDeviceIdentity"
deviceId: "qiita-test-device-01"
hubName: "iot-hub-test-0001"
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/28b735e6-9010-4d8a-bded-dbb8fa1792a2.png)

#### SystemProperties

```text
twin-version: "2"
connectionDeviceId: "qiita-test-device-01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "2d649198c20"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T01:40:53.2760000Z"
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e0394274-d55f-49c2-b680-785e3a2b01d7.png)

#### body

```json
{
	"deviceId": "qiita-test-device-01",
	"etag": "AAAAAAAAAAE=",
	"version": 2,
	"properties": {
		"desired": {
			"$metadata": {
				"$lastUpdated": "2025-05-27T01:40:32.8491207Z"
			},
			"$version": 1
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "2025-05-27T01:40:32.8491207Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/48c9657d-34b6-4d80-8970-41d870cf0d17.png)


### モジュール作成時のデータ


#### Properties

```text
moduleId: "qiita-test-module-01"
operationTimestamp: "2025-05-27T01:40:42.4587483Z"
iothub-message-schema: "moduleLifecycleNotification"
opType: "createModuleIdentity"
deviceId: "qiita-test-device-01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/18d294ce-e9a9-4bda-b4ca-a6016435097d.png)

#### SystemProperties

```text
twin-version: "2"
connectionModuleId: "qiita-test-module-01"
connectionDeviceId: "qiita-test-device-01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "2d642ae98ce"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T01:40:42.5100000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/15a1c235-4a85-440a-aabb-82b820fcf47b.png)

#### body

```json
{
	"deviceId": "qiita-test-device-01",
	"moduleId": "qiita-test-module-01",
	"etag": "AAAAAAAAAAE=",
	"version": 2,
	"properties": {
		"desired": {
			"$metadata": {
				"$lastUpdated": "0001-01-01T00:00:00Z"
			},
			"$version": 1
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "0001-01-01T00:00:00Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1ccbc01a-505a-4f04-806b-4e81a29e4bdf.png)

### モジュール削除時のデータ


#### Properties

```text
moduleId: "qiita-test-module-01"
operationTimestamp: "2025-05-27T01:40:46.7869929Z"
iothub-message-schema: "moduleLifecycleNotification"
opType: "deleteModuleIdentity"
deviceId: "qiita-test-device-01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7522040f-5d38-4aa6-8a22-1dd2bfca5db5.png)

#### SystemProperties

```text
twin-version: "2"
connectionModuleId: "qiita-test-module-01"
connectionDeviceId: "qiita-test-device-01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "2d6453de53e"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T01:40:46.8070000Z"
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dc6f3628-39b4-487c-a133-8212357e6b8d.png)

#### body

```json
{
	"deviceId": "qiita-test-device-01",
	"moduleId": "qiita-test-module-01",
	"etag": "AAAAAAAAAAE=",
	"version": 2,
	"properties": {
		"desired": {
			"$metadata": {
				"$lastUpdated": "0001-01-01T00:00:00Z"
			},
			"$version": 1
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "0001-01-01T00:00:00Z"
			},
			"$version": 1
		}
	}
}
```


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0117bfae-fd65-4850-be5a-eb0e1d095a4d.png)