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

## 概要

Azure IoT Hub から Azure Event Hubs にメッセージルーティンしたデバイス接続状態イベント（`DeviceConnectionStateEvents`データソース）を 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/d2322218-a645-4c8c-86e6-a0151d3838c2.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)


#### Databricks にてモジュールの接続と切断の実施


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

```python
module_conn_str = "HostName=iot-hub-test-001.azure-devices.net;DeviceId=qiita_test_device_id_01;ModuleId=qiita_test_module_01;SharedAccessKey=DaHuqr3caI6Edi6UKmdZCoCE+la0S/XXXXX="
```

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

module_client = IoTHubDeviceClient.create_from_connection_string(
    module_conn_str,
    websockets=True,
)
```

```python
# デバイスでの接続
module_client.connect()
```

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c693aa2b-ed7f-4ad2-9a7a-3f22ed1d6b7e.png)


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/779e1bcd-2ebb-4501-ad26-1406c68426ee.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/4df394bb-27e4-4478-80c1-27fef0c9d144.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/129d3f37-48a6-4cc0-8b06-5c3bf0b64502.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a206cf1c-62f6-4971-a2db-261fb3e7a878.png)

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

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

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c59bdc5f-92f7-41fc-bf85-93561b5711bd.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-001                  |
| content-type                | application/json                  |
| content-encoding            | utf-8                             |
| iothub-connection-device-id | qiita_test_device_id_01           |
| iothub-connection-module-id | qiita_test_module_01              |
| iothub-message-schema       | deviceConnectionStateNotification |
| iothub-enqueuedtime         | 2025-05-26T06:35:58.969+00:00     |
| iothub-message-source       | deviceConnectionStateEvents       |
| operationTimestamp          | 2025-05-26T06:35:19.950+00:00     |
| opType                      | deviceConnected                   |
| hubName                     | iot-hub-test-001                  |
| deviceId                    | qiita_test_device_id_01           |
| moduleId                    | qiita_test_module_01              |

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e885490a-1389-461a-8c6b-dfad93e7a749.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
{"sequenceNumber":"000000000000000001DBCE0D753FFC4B00000002000000000000000000000014"}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d94a7bb4-4a23-413b-8a1e-d720675b87ce.png)


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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fc496bf8-247c-4606-927e-383974f9f70a.png)


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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7294965e-c1a5-434e-a2ee-18a6eee4ff83.png)


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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a66fc676-27ff-4a1f-adce-fff2d9f6d1f2.png)