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

## 概要

Azure IoT Hub から Azure Event Hubs にメッセージルーティンしたデバイス ツイン変更イベント（`TwinChangeEvents`データソース）を Databricks で参照する方法を紹介します。

## 事前準備

### 環境構築

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

### Evnet hub の作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2e59a836-51ae-4ddf-9a1f-149652ffe507.png)

### Azure Event Hubs の接続文字列を取得

Azure Event Hubs のリソース画面にて、`共有アクセス ポリシー`タブ -> `追加`ポリシーを選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7a2ca188-1911-409a-808b-ca6f3bfe9acd.png)

ポリシー名に任意の名前を入力後、`リッスン`をチェックし、`作成`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0bd389c7-ae1d-4de5-b88e-4f8167b3afe6.png)

`プライマリ接続文字列`の値をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/66693071-1686-4ed5-b36c-dc7564887653.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/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/51ce9cd5-6561-4729-bd7c-2e6fe4d5dcff.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)


### Azure IoT Hubs にてレコードを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4ab22746-a970-4b34-843b-e3e4a70d4cad.png)

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

### Azure Event Hubs から Kafka エンドポイントによりデータを取得

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

```python
eh_namespace_name = "kafka-test-01"
eh_connection_string = "Endpoint=sb://kafka-test-01.servicebus.windows.net/;SharedAccessKeyName=qiita-sas-01;SharedAccessKey=fiKmMqJtmn2dGAeV6oj1dWWxxxx="
topic_name = "iot-message-01"

# Databricks から Azure Event Hubs に設定する際の設定をセット
eh_sasl = (
    "kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule "
    f'required username="$ConnectionString" password="{eh_connection_string}";'
)
bootstrap_servers = f"{eh_namespace_name}.servicebus.windows.net:9093"
kafka_options = {
    "kafka.bootstrap.servers": bootstrap_servers,
    "kafka.sasl.mechanism": "PLAIN",
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": eh_sasl,
}
```

```python
# Azure Event Hubs からデータを取得
from pyspark.sql.functions import col

src_df = (
    spark.readStream.format("kafka")
    .options(**kafka_options)
    .option("subscribe", topic_name)
    .option("startingOffsets", "earliest")  # 検証目的であるため earliest に設定
    .option("includeHeaders", "true")
    .load()
)
```

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

```text
root
 |-- key: binary (nullable = true)
 |-- value: binary (nullable = true)
 |-- topic: string (nullable = true)
 |-- partition: integer (nullable = true)
 |-- offset: long (nullable = true)
 |-- timestamp: timestamp (nullable = true)
 |-- timestampType: integer (nullable = true)
 |-- headers: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- key: string (nullable = true)
 |    |    |-- value: binary (nullable = true)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b06a4367-8bc4-44a6-aaf0-9058ae63b8a5.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3d2c2a29-010c-45a7-8c6b-4d8d1ef5a6bb.png)

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

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

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

```json
{
	"version": 3,
	"properties": {
		"desired": {
			"temp": "100",
			"$metadata": {
				"$lastUpdated": "2025-05-26T07:20:59.6106683Z",
				"$lastUpdatedVersion": 2,
				"temp": {
					"$lastUpdated": "2025-05-26T07:20:59.6106683Z",
					"$lastUpdatedVersion": 2
				}
			},
			"$version": 2
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5441350d-7dd3-422c-a803-9f1ffe57211f.png)

### `header` 列からデータの取得

<details><summary>header 列から値を取得する関数はこちら</summary>

```python
"""IoT Hub AMQP ヘッダー列を PySpark DataFrame に展開するユーティリティ.

IoT Hub から取得した AMQP 形式のヘッダーは Binary 型で格納される。
本モジュールでは以下を行う。

* 各バイト列を適切な型 (string / timestamp) にデコード
* 配列形式のヘッダー (`array<struct<key:string,value:binary>>`) を
  一括で構造化 (`struct`) する

Google Style Docstring を採用し、日本語で詳細を記述している。

Note:
    Databricks Runtime 14.3 / Spark 3.5.0 で動作確認。
"""
from __future__ import annotations

from functools import reduce
import operator as op
from typing import Dict

from pyspark.sql import DataFrame
from pyspark.sql.functions import (
    col,
    conv,
    element_at,
    expr,
    map_from_entries,
    substring,
    timestamp_millis,
)

# --------------------------------------------------------------------------- #
# 単体デコーダー
# --------------------------------------------------------------------------- #

def decode_string(df: DataFrame, col_name: str, char_set: str = "UTF-8") -> DataFrame:
    """BINARY → STRING 変換を行う.

    Args:
        df: 対象 DataFrame.
        col_name: 変換対象の列名 (binary).
        char_set: デコードに用いる文字セット. デフォルトは ``"UTF-8"``.

    Returns:
        ``col_name`` が string 型へ置換された DataFrame.
    """
    return df.withColumn(col_name, expr(f"decode(`{col_name}`, '{char_set}')"))


def decode_string_with_str8utf8(
    df: DataFrame,
    col_name: str,
    prefix: int = 2,
    char_set: str = "UTF-8",
) -> DataFrame:
    """AMQP ``str8-utf8`` 形式 (prefix+string) をデコードする.

    Args:
        df: 対象 DataFrame.
        col_name: 変換対象の列名 (binary).
        prefix: 先頭の prefix バイト数. デフォルト ``2``.
        char_set: デコードに用いる文字セット.

    Returns:
        ``col_name`` が string 型へ置換された DataFrame.
    """
    sliced = expr(
        f"decode(substring(`{col_name}`, {prefix} + 1, length(`{col_name}`) - {prefix}), '{char_set}')"
    )
    return df.withColumn(col_name, sliced)


def decode_timestamp_str8_utf8(
    df: DataFrame,
    col_name: str,
    prefix: int = 2,
    char_set: str = "UTF-8",
    ts_fmt="yyyy-MM-dd\\'T\\'HH:mm:ss.SSSSSSSX",
) -> DataFrame:
    """AMQP ``timestamp:str8-utf8`` を Spark の ``timestamp`` へ変換する.

    流れ:

    1. ``str8-utf8`` を文字列へデコード
    2. ``TRY_TO_TIMESTAMP`` で timestamp へ変換

    Args:
        df: 対象 DataFrame.
        col_name: 変換対象の列名 (binary).
        prefix: 先頭の prefix バイト数.
        char_set: デコードに用いる文字セット.
        ts_fmt: Spark の日時書式 (``SimpleDateFormat``).

    Returns:
        ``col_name`` が timestamp 型へ置換された DataFrame.
    """
    df = decode_string_with_str8utf8(df, col_name, prefix, char_set)
    return df.withColumn(col_name, expr(f"TRY_TO_TIMESTAMP(`{col_name}`, '{ts_fmt}')"))


def decode_timestamp_with_ms64(df: DataFrame, col_name: str) -> DataFrame:
    """AMQP ``timestamp:ms64`` を Spark の ``timestamp`` へ変換する.

    AMQP ms64 は「符号付き 64bit big-endian 整数 (ミリ秒)」で表される。

    流れ:

    1. 先頭 1 バイトのフォーマットコードを除去
    2. 8 バイトを 16 進文字列へ変換
    3. 各バイトを ``long`` へ展開しビットシフトで合算
    4. ``timestamp_millis`` で ``timestamp`` へ変換

    Args:
        df: 対象 DataFrame.
        col_name: 変換対象の列名 (binary).

    Returns:
        ``col_name`` が timestamp 型へ置換された DataFrame.
    """
    # 1) 余計な 1 バイトを除去し 8 バイトを抽出
    df = df.withColumn(col_name, substring(col(col_name), 2, 8))
    # 2) BINARY → HEX string
    df = df.withColumn(col_name, expr(f"hex(`{col_name}`)"))

    # 3) HEX → signed 64bit integer
    byte_exprs = [
        conv(substring(col(col_name), i * 2 + 1, 2), 16, 10).cast("long") * (1 << (56 - 8 * i))
        for i in range(8)
    ]
    df = df.withColumn(col_name, reduce(op.add, byte_exprs))

    # 4) epoch milli → timestamp
    return df.withColumn(col_name, timestamp_millis(col(col_name)))


# --------------------------------------------------------------------------- #
# ヘッダー展開ユーティリティ
# --------------------------------------------------------------------------- #

def add_cols_from_iot_hub_headers(
    df: DataFrame,
    header_col_name: str = "headers",
    drop_temp_col: bool = True,
) -> DataFrame:
    """ヘッダー配列列を各フィールド列へ展開する.

    IoT Hub ではヘッダーが ``array<struct<key:string,value:binary>>`` として
    送出される。本関数は以下を実施する。

    * ``header_col_name`` を ``map<string,binary>`` へ変換
    * :pydata:`AMQP_HEADER_SPEC` に従い列を追加し型変換
        * ``string``             → :func:`decode_string`
        * ``string:str8-utf8``   → :func:`decode_string_with_str8utf8`
        * ``timestamp:str8-utf8``→ :func:`decode_timestamp_str8_utf8`
        * ``timestamp:ms64``     → :func:`decode_timestamp_with_ms64`

    Args:
        df: 対象 DataFrame.
        header_col_name: AMQP ヘッダーを格納する列名.
        drop_temp_col: 一時列を削除するかどうか.

    Returns:
        追加済み列を含む DataFrame.

    Raises:
        ValueError: :pydata:`AMQP_HEADER_SPEC` に未知の型が含まれる場合.
    """
    # 1) ARRAY<struct<key,value>> → MAP<key,value>
    temp_map_col = f"_amqp_header_map_{header_col_name}"
    df = df.withColumn(temp_map_col, map_from_entries(col(header_col_name)))

    # 2) 型仕様に従い列を追加
    for key, col_type in AMQP_HEADER_SPEC.items():
        df = df.withColumn(key, element_at(col(temp_map_col), key))

        col_type_lc = col_type.lower()
        if col_type_lc == "string":
            df = decode_string(df, key)
        elif col_type_lc == "string:str8-utf8":
            df = decode_string_with_str8utf8(df, key)
        elif col_type_lc == "timestamp:str8-utf8":
            df = decode_timestamp_str8_utf8(df, key)
        elif col_type_lc == "timestamp:ms64":
            df = decode_timestamp_with_ms64(df, key)
        else:
            raise ValueError(f"Unknown AMQP header type: {col_type}")

    if drop_temp_col:
        df = df.drop(temp_map_col)
    return df

```

</details>

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/850a22d3-9ac5-4d4c-8997-2c0e7b2c913f.png)


```python
AMQP_HEADER_SPEC = {
    "user-id": "string",
    "content-type": "string",
    "content-encoding": "string",
    "iothub-connection-device-id": "string:str8-utf8",
    "iothub-connection-module-id": "string:str8-utf8",
    "iothub-message-schema": "string:str8-utf8",
    "iothub-enqueuedtime": "timestamp:ms64",
    "iothub-message-source": "string:str8-utf8",
    "operationTimestamp": "timestamp:str8-utf8",
    "opType": "string:str8-utf8",
    "hubName": "string:str8-utf8",
    "deviceId": "string:str8-utf8",
    "moduleId": "string:str8-utf8",
}

header_col_name = "headers"

df = add_cols_from_iot_hub_headers(src_df, header_col_name)

df.printSchema()
df.display()
```

| key                         | value                             |
| --------------------------- | --------------------------------- |
| user-id                     | iot-hub-test-0001             |
| content-type                | application/json              |
| content-encoding            | utf-8                         |
| iothub-connection-device-id | qiita_test_device_01          |
| iothub-connection-module-id | qiita_test_module_01          |
| iothub-message-schema       | twinChangeNotification        |
| iothub-enqueuedtime         | 2025-05-26T07:22:06.101+00:00 |
| iothub-message-source       | twinChangeEvents              |
| operationTimestamp          | 2025-05-26T07:22:06.096+00:00 |
| opType                      | updateTwin                    |
| hubName                     | iot-hub-test-0001             |
| deviceId                    | qiita_test_device_01          |
| moduleId                    | qiita_test_module_01          |

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cd4a4e7c-503f-45b7-ab4f-5882d653848c.png)


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

### データの確認方法

Azure Iot Hubs における Data Explorer にて確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6e3e485d-a773-464b-a6a6-1173d4ccffff.png)


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

```json
{
	"version": 3,
	"properties": {
		"desired": {
			"temp": "100",
			"$metadata": {
				"$lastUpdated": "2025-05-26T07:20:59.6106683Z",
				"$lastUpdatedVersion": 2,
				"temp": {
					"$lastUpdated": "2025-05-26T07:20:59.6106683Z",
					"$lastUpdatedVersion": 2
				}
			},
			"$version": 2
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/110bbb15-65ad-41b7-97fa-14d3b13b6ee1.png)

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


```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-26T07:21:20.6581312Z",
				"$lastUpdatedVersion": 3,
				"key_3": {
					"$lastUpdated": "2025-05-26T07:21:20.6581312Z",
					"$lastUpdatedVersion": 3
				},
				"key_1": {
					"$lastUpdated": "2025-05-26T07:21:20.6581312Z",
					"$lastUpdatedVersion": 3
				},
				"key_2": {
					"$lastUpdated": "2025-05-26T07:21:20.6581312Z",
					"$lastUpdatedVersion": 3
				}
			},
			"$version": 3
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "2025-05-26T07:18:02.3713524Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fcaaaafa-fd49-4964-9381-01a79711b972.png)

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


```json
{
	"version": 3,
	"properties": {
		"desired": {
			"temp": 100,
			"$metadata": {
				"$lastUpdated": "2025-05-26T07:22:06.0969398Z",
				"$lastUpdatedVersion": 2,
				"temp": {
					"$lastUpdated": "2025-05-26T07:22:06.0969398Z",
					"$lastUpdatedVersion": 2
				}
			},
			"$version": 2
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/379007e4-4fa2-4a0f-90c6-25ed9ca7a478.png)

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


```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-26T07:22:11.5189383Z",
				"$lastUpdatedVersion": 3,
				"key_3": {
					"$lastUpdated": "2025-05-26T07:22:11.5189383Z",
					"$lastUpdatedVersion": 3
				},
				"key_1": {
					"$lastUpdated": "2025-05-26T07:22:11.5189383Z",
					"$lastUpdatedVersion": 3
				},
				"key_2": {
					"$lastUpdated": "2025-05-26T07:22:11.5189383Z",
					"$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/3658be5b-3a2f-4a0b-8bfa-b1f6380bd4d3.png)