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

## 概要

Azure IoT Hub から Azure Storage にメッセージルーティンしたデバイス ツイン変更イベント（`TwinChangeEvents`データソース）を 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/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/146174b7-cc0f-4133-8c8d-8bfadb4c5067.png)

### デバイスとモジュールの作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/51d6c229-a07a-46c6-87ca-c54d08d46c18.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0d759439-ca1c-4088-9f2d-8928608e1219.png)


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

### 事前準備

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

```python
iot_conn_str = "HostName=iot-hub-test-0001.azure-devices.net;SharedAccessKeyName=iothubowner;SharedAccessKey=V/HbzmQCQJHBScUBCuXXXX="
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7e2f9344-642f-4b7b-b207-a60ea51cdd3a.png)

### デバイス twin に対する update_twin と replace_twin の実施

```python
device_id = "qiita_test_device_01"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9c593751-7e51-4483-bf8c-e3def303aac9.png)



```python
from azure.iot.hub import IoTHubRegistryManager
from azure.iot.hub.models import Twin, TwinProperties

# クライアント初期化
registry_manager = IoTHubRegistryManager(iot_conn_str)

try:
    # 現在のツインを取得
    twin = registry_manager.get_twin(device_id)
    print("変更前の Twin:", twin.as_dict()["properties"]["desired"].get("temp", None))

    # 更新内容を定義
    desired = {"temp": "100"}
    twin_patch = Twin(properties=TwinProperties(desired=desired))

    # 部分更新を実行（ETag を指定）
    updated_twin = registry_manager.update_twin(device_id, twin_patch, twin.etag)
    print("変更後の Twin:", updated_twin.as_dict()["properties"]["desired"].get("temp", None))
except Exception as ex:
    print("エラー:", ex)

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5e8627e4-87d4-46be-acd8-652862266efa.png)


```python
from azure.iot.hub import IoTHubRegistryManager
from azure.iot.hub.models import Twin, TwinProperties

registry_manager = IoTHubRegistryManager(iot_conn_str)

# 現在のツインを取得
twin = registry_manager.get_twin(device_id)
print("変更前の Twin:", twin.as_dict()["properties"])

# 新しいツイン情報を構築
new_twin = Twin(
    properties=TwinProperties(
        desired={
            "key_1": "a",
            "key_2": "b",
            "key_3": "c",
        },
        reported=None  # reported プロパティは変更しない場合 None
    )
)
# ツイン全体を置換
replaced_twin = registry_manager.replace_twin(device_id, new_twin)
print("")
print("変更後の Twin:", replaced_twin.as_dict()["properties"])

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/24641925-1daf-4d55-8a4f-a2a8de74d7ad.png)


### モジュール twin に対する update_twin と replace_twin の実施

```python
device_id = "qiita_test_device_01"
module_id = "qiita_test_module_01"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/15f72441-18bf-479f-a67f-3214a23e874d.png)


```python
from azure.iot.hub import IoTHubRegistryManager
from azure.iot.hub.models import Twin, TwinProperties

# クライアント初期化
registry_manager = IoTHubRegistryManager(iot_conn_str)

try:
    # ① Twin を取得して現在値と etag を確認
    twin = registry_manager.get_module_twin(device_id, module_id)
    print("Before:", twin.properties.desired.get("temp"))

    # ② 差分だけを Twin Patch に詰める
    patch = Twin(properties=TwinProperties(desired={"temp": 100}))

    # ③ 部分更新（競合無視なら etag='*'）
    updated = registry_manager.update_module_twin(
        device_id, module_id, patch, etag=twin.etag         # or etag="*"
    )

    print("After :", updated.properties.desired.get("temp"))
except Exception as ex:
    print("エラー:", ex)

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b9400244-79f6-43fe-ab97-e44023b04d5f.png)

```python
from azure.iot.hub import IoTHubRegistryManager
from azure.iot.hub.models import Twin, TwinProperties

registry_manager = IoTHubRegistryManager(iot_conn_str)

# ① 現在のモジュール Twin を取得 (etag も受け取る)
twin = registry_manager.get_module_twin(device_id, module_id)  # ← module 用 API
print("置換前:", twin.properties.desired)

# ② Twin オブジェクトを丸ごと構築
# ※ Service 側からは tags と desired しか書き込めません
new_twin = Twin(
    properties=TwinProperties(
        desired={
            "key_1": "a",
            "key_2": "b",
            "key_3": "c",
        }
    )
)

# ③ etag を使って全置換 (競合無視なら etag="*")
replaced = registry_manager.replace_module_twin(
    device_id, module_id, new_twin, etag=twin.etag        # or etag="*"
)

print("置換後:", replaced.properties.desired)

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c9af65a6-5951-4b2d-9a7a-d68bf1c15427.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/d1e9abfc-1745-4fff-8e83-8469a2d440b4.png)


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

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

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

```json
{
	"version": 3,
	"properties": {
		"desired": {
			"temp": "100",
			"$metadata": {
				"$lastUpdated": "2025-05-27T04:08:48.9093199Z",
				"$lastUpdatedVersion": 2,
				"temp": {
					"$lastUpdated": "2025-05-27T04:08:48.9093199Z",
					"$lastUpdatedVersion": 2
				}
			},
			"$version": 2
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/82368dac-2543-42be-bd4d-b336fad640ee.png)


:::note info

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5c1a1067-eac0-4bf4-be60-0fb66a5d73d4.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/94ff8ecd-ba2c-4ba9-80a6-fd0f22e9bafc.png)


:::



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

### デバイス twin に対する update_twin のデータ

#### Properties

```text
operationTimestamp: "2025-05-27T04:08:48.9093199Z"
iothub-message-schema: "twinChangeNotification"
opType: "updateTwin"
deviceId: "qiita_test_device_01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f5483648-a9a1-4fff-9cca-4aa8fed21b24.png)


#### SystemProperties

```text
twin-version: "3"
connectionDeviceId: "qiita_test_device_01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "2eaf33f41d3"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T04:08:49.0610000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/85fe6402-e8bd-4b63-9c00-e3f4dca65bcb.png)

#### body

```json
{
	"version": 3,
	"properties": {
		"desired": {
			"temp": "100",
			"$metadata": {
				"$lastUpdated": "2025-05-27T04:08:48.9093199Z",
				"$lastUpdatedVersion": 2,
				"temp": {
					"$lastUpdated": "2025-05-27T04:08:48.9093199Z",
					"$lastUpdatedVersion": 2
				}
			},
			"$version": 2
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d672630f-1465-4177-8797-b924eca601f1.png)


### デバイス twin に対する replace_twin のデータ

#### Properties

```text
operationTimestamp: "2025-05-27T04:08:51.5656567Z"
iothub-message-schema: "twinChangeNotification"
opType: "replaceTwin"
deviceId: "qiita_test_device_01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0993a556-b8ed-4053-b852-f613cb45f41d.png)


#### SystemProperties

```text
twin-version: "4"
connectionDeviceId: "qiita_test_device_01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "2eaf4d2d8d1"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T04:08:51.5770000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/554059ef-b1e1-4adb-9ffd-38deb90b9fe3.png)

#### body

```json
{
	"deviceId": "qiita_test_device_01",
	"etag": "AAAAAAAAAAM=",
	"version": 4,
	"properties": {
		"desired": {
			"key_3": "c",
			"key_1": "a",
			"key_2": "b",
			"$metadata": {
				"$lastUpdated": "2025-05-27T04:08:51.5656567Z",
				"$lastUpdatedVersion": 3,
				"key_3": {
					"$lastUpdated": "2025-05-27T04:08:51.5656567Z",
					"$lastUpdatedVersion": 3
				},
				"key_1": {
					"$lastUpdated": "2025-05-27T04:08:51.5656567Z",
					"$lastUpdatedVersion": 3
				},
				"key_2": {
					"$lastUpdated": "2025-05-27T04:08:51.5656567Z",
					"$lastUpdatedVersion": 3
				}
			},
			"$version": 3
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "2025-05-27T04:08:31.3307051Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b1199893-35ca-4ece-9ea1-6ced446663de.png)


### モジュール twin に対する update_twin のデータ


#### Properties

```text
moduleId: "qiita_test_module_01"
operationTimestamp: "2025-05-27T04:08:56.0189103Z"
iothub-message-schema: "twinChangeNotification"
opType: "updateTwin"
deviceId: "qiita_test_device_01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ed30ec6e-0bdf-4045-9627-e90d1fc11830.png)


#### SystemProperties

```text
twin-version: "3"
connectionModuleId: "qiita_test_module_01"
connectionDeviceId: "qiita_test_device_01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "2eaf77b7232"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T04:08:56.0300000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/db1f2f80-ed97-4b12-a14c-bfe63ea1d0db.png)

#### body

```json
{
	"version": 3,
	"properties": {
		"desired": {
			"temp": 100,
			"$metadata": {
				"$lastUpdated": "2025-05-27T04:08:56.0189103Z",
				"$lastUpdatedVersion": 2,
				"temp": {
					"$lastUpdated": "2025-05-27T04:08:56.0189103Z",
					"$lastUpdatedVersion": 2
				}
			},
			"$version": 2
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3e72dc99-d956-46c9-8c5b-33e33675297d.png)

### モジュール twin に対する replace_twin の実データ

#### Properties

```text
moduleId: "qiita_test_module_01"
operationTimestamp: "2025-05-27T04:08:57.6127052Z"
iothub-message-schema: "twinChangeNotification"
opType: "replaceTwin"
deviceId: "qiita_test_device_01"
hubName: "iot-hub-test-0001"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/74715090-59e4-4fb8-a8d8-2efee5252737.png)


#### SystemProperties

```text
twin-version: "4"
connectionModuleId: "qiita_test_module_01"
connectionDeviceId: "qiita_test_device_01"
contentType: "application/json"
contentEncoding: "utf-8"
correlationId: "2eaf86e9cf5"
userId: "aW90LWh1Yi10ZXN0LTAwMDE="
enqueuedTime: "2025-05-27T04:08:57.6400000Z"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/335c8665-2e9f-4c9c-8f3b-4185b6ee44db.png)

#### body

```json
{
	"deviceId": "qiita_test_device_01",
	"moduleId": "qiita_test_module_01",
	"etag": "AAAAAAAAAAM=",
	"version": 4,
	"properties": {
		"desired": {
			"key_3": "c",
			"key_1": "a",
			"key_2": "b",
			"$metadata": {
				"$lastUpdated": "2025-05-27T04:08:57.6127052Z",
				"$lastUpdatedVersion": 3,
				"key_3": {
					"$lastUpdated": "2025-05-27T04:08:57.6127052Z",
					"$lastUpdatedVersion": 3
				},
				"key_1": {
					"$lastUpdated": "2025-05-27T04:08:57.6127052Z",
					"$lastUpdatedVersion": 3
				},
				"key_2": {
					"$lastUpdated": "2025-05-27T04:08:57.6127052Z",
					"$lastUpdatedVersion": 3
				}
			},
			"$version": 3
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "0001-01-01T00:00:00Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5def311a-1cf6-4ab9-8aa8-b2e457edcc64.png)