Azure IoT Hub から Azure Event Hubs にメッセージルーティンした device-to-cloud メッセージ(テレメトリー)を Databricks で参照する方法

## 概要

Azure IoT Hub から Azure Event Hubs にメッセージルーティンした device-to-cloud メッセージ(`DeviceMessages`データソース)を 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/3107fd26-1634-44b2-bddf-a225de0eb490.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)

#### モジュールの作成と接続文字列の取得

作成済みのデバイスの画面に、`+ モジュール 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()
```

### デバイスから device-to-cloud メッセージ(テレメトリー)の送信

```python
import os, json, random, time, uuid
from datetime import datetime, timezone
from azure.iot.device import IoTHubDeviceClient, Message


def get_sample_date():
    date = {
        "temperature": round(random.uniform(20, 25), 2),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }

    return json.dumps(date)

# クライアントのインスタンス化
device_client = IoTHubDeviceClient.create_from_connection_string(
    device_conn_str,
    websockets=True,
)

# データの準備
payload = get_sample_date()
msg = Message(payload)
msg.message_id = uuid.uuid4()
msg.content_encoding = "utf-8"
msg.content_type = "application/json"
msg.custom_properties["sensor-location"] = "jp"

# クライアントに接続
device_client.connect()

# テレメトリーの送信
device_client.send_message(msg)
print("✓ sent:", payload)
time.sleep(5)

# クライアントの終了
device_client.shutdown()
```


### モジュールから device-to-cloud メッセージ(テレメトリー)の送信


```python
module_conn_str = "HostName=iot-hub-test-001.azure-devices.net;DeviceId=qiita_test_01;ModuleId=qiita_test_module_01;SharedAccessKey=s5svMAgEglFjcr923J+GfDcXXXXX="

import os, json, random, time, uuid
from datetime import datetime, timezone
from azure.iot.device import IoTHubDeviceClient, Message


def get_sample_date():
    date = {
        "temperature": round(random.uniform(20, 25), 2),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }

    return json.dumps(date)

# クライアントのインスタンス化
module_client = IoTHubDeviceClient.create_from_connection_string(
    module_conn_str,
    websockets=True,
)

# データの準備
payload = get_sample_date()
msg = Message(payload)
msg.message_id = uuid.uuid4()
msg.content_encoding = "utf-8"
msg.content_type = "application/json"
msg.custom_properties["sensor-location"] = "jp"

# クライアントに接続
module_client.connect()

# テレメトリーの送信
module_client.send_message(msg)
print("✓ sent:", payload)
time.sleep(5)

# クライアントの終了
module_client.shutdown()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7bbddad9-da77-4bda-bea1-e26434eea3b6.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: binary (nullable = true)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fdf32d3d-94e5-4260-9666-86cff63ee2da.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c925fec7-ca31-4c6b-8c00-31b72244a1d9.png)


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

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

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

```json
{
	"temperature": 22.1,
	"timestamp": "2025-05-26T07:43:10.822160+00:00"
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/22041e64-6a19-49b7-a375-aa0a83b1b89d.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 = {
    "content-type": "string",
    "content-encoding": "string",
    "message-id": "string",
    "sensor-location": "string:str8-utf8",
    "iothub-message-source": "string:str8-utf8",
    "iothub-connection-device-id": "string:str8-utf8",
    "iothub-connection-module-id": "string:str8-utf8",
    "iothub-connection-auth-method": "string:str8-utf8",
    "iothub-connection-auth-generation-id": "string:str8-utf8",
    "iothub-enqueuedtime": "timestamp:ms64",
}

header_col_name = "headers"

df = add_cols_from_iot_hub_headers(src_df, header_col_name)

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

| key                                  | value                                             |
| ------------------------------------ | ------------------------------------------------- |
| content-type                         | application/json                                  |
| content-encoding                     | utf-8                                             |
| message-id                           | 4c326ad0-4a33-4283-a5fd-b94c9cb8dcea              |
| sensor-location                      | jp                                                |
| iothub-message-source                | Telemetry                                         |
| iothub-connection-device-id          | qiita_test_device_01                              |
| iothub-connection-module-id          | qiita_test_module_01                              |
| iothub-connection-auth-method        | {"scope":"module","type":"sas","issuer":"iothub"} |
| iothub-connection-auth-generation-id | 638838407041844000                                |
| iothub-enqueuedtime                  | 2025-05-26T07:43:57.495+00:00                     |

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7340aa03-a806-4e78-99af-76289f54ad7e.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)


### デバイスによる device-to-cloud メッセージ(テレメトリー)のデータ


```json
{
	"temperature": 22.1,
	"timestamp": "2025-05-26T07:43:10.822160+00:00"
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0a905495-7dcd-46ec-b661-a565bb836aa2.png)

### モジュールによる device-to-cloud メッセージ(テレメトリー)のデータ


```json
{
	"temperature": 22.38,
	"timestamp": "2025-05-26T07:43:57.000578+00:00"
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/94f79afa-9f61-4364-b2b6-33a1cc5b48e2.png)