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

## 概要

Azure IoT Hub から Azure Event Hubs にメッセージルーティンしたデバイス ライフサイクル イベント（`DeviceLifecycleEvents`データソース）を 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/1f17cfd5-fd5d-4a38-8eef-40a5e611aa97.png)


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e643c1f2-7d70-4a6d-865e-879065c3425c.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/faaec6e5-b13c-4ca4-bc52-f846baebcdb8.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 IoT Hubs にてレコードを確認

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c8e39078-e763-4d85-a4ee-ff42711b5596.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)
```

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

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

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

```json
{
	"deviceId": "qiita-test-device-01",
	"etag": "AAAAAAAAAAE=",
	"version": 2,
	"properties": {
		"desired": {
			"$metadata": {
				"$lastUpdated": "2025-05-26T04:37:50.7967261Z"
			},
			"$version": 1
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "2025-05-26T04:37:50.7967261Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1a26daf0-998b-40e5-a59f-6d68563c55fa.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",
    "operationTimestamp": "timestamp:str8-utf8",
    "iothub-message-schema": "string:str8-utf8",
    "iothub-enqueuedtime": "timestamp:ms64",
    "opType": "string:str8-utf8",
    "deviceId": "string:str8-utf8",
    "hubName": "string:str8-utf8",
    "iothub-message-source": "string:str8-utf8",
}

header_col_name = "headers"

df = add_cols_from_iot_hub_headers(df, header_col_name)

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


| key                         | value                         |
| --------------------------- | ----------------------------- |
| user-id                     | iot-hub-test-001              |
| content-type                | application/json              |
| content-encoding            | utf-8                         |
| iothub-connection-device-id | qiita-test-device-01          |
| operationTimestamp          | 2025-05-26T04:37:50.796+00:00 |
| iothub-message-schema       | deviceLifecycleNotification   |
| iothub-enqueuedtime         | 2025-05-26T04:37:50.929+00:00 |
| opType                      | createDeviceIdentity          |
| deviceId                    | qiita-test-device-01          |
| hubName                     | iot-hub-test-001              |
| iothub-message-source       | deviceLifecycleEvents         |

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f3f297d9-bf17-4de3-8778-aa348009c0bc.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)


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

```json
{
	"deviceId": "qiita-test-device-01",
	"etag": "AAAAAAAAAAE=",
	"version": 2,
	"properties": {
		"desired": {
			"$metadata": {
				"$lastUpdated": "2025-05-26T04:37:50.7967261Z"
			},
			"$version": 1
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "2025-05-26T04:37:50.7967261Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b767acf2-b5bd-4bae-a22c-6ed347bfa522.png)


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

```json
{
	"deviceId": "qiita-test-device-01",
	"etag": "AAAAAAAAAAE=",
	"version": 2,
	"properties": {
		"desired": {
			"$metadata": {
				"$lastUpdated": "2025-05-26T04:37:50.7967261Z"
			},
			"$version": 1
		},
		"reported": {
			"$metadata": {
				"$lastUpdated": "2025-05-26T04:37:50.7967261Z"
			},
			"$version": 1
		}
	}
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8280fc49-c821-41cc-8f3c-f8abe95c0427.png)


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

```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/311e8a46-1df7-45e7-80e4-782a92ad6e57.png)

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

```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/72777d7a-1c2d-4a3f-8af9-62361b686818.png)