Databricks にて Azure IoT Hub に対して ジョブの作成と実行を実施する方法

## 概要

Databricks にて Azure IoT Hub に対して ジョブの作成と実行を実施する方法を共有します。下記の 2 種類のジョブの実行方法を記述しています。

1. scheduleDeviceMethod ジョブ
1. scheduleUpdateTwin ジョブ

## 事前準備

### 環境構築

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

### デバイスの作成と接続文字列の取得

Azure IoT Hub のリソース画面にて、`デバイス`タブを選択後、`+ デバイスの追加`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/32708e89-0264-4b64-b5f2-494be974a7d4.png)

`デバイス ID`に任意の値を入力し、`保存`を選択してデバイスを作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/37ac238e-cbd3-48d6-82f5-68cf9038354e.png)

作成したデバイスを表示して、`プライマリ接続文字列`の`コピー済み`を選択して値をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4f8d5549-6a72-4ff4-9e10-0e23214201f6.png)

### Azure IoT Hub の接続文字列を取得

Azure IoT Hub のリソース画面にて、`共有アクセス ポリシー`タブ -> `iothubowner`ポリシーを選択して、`プライマリ接続文字列`の値をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ac063b7d-54f5-4dbd-a27a-92cde4a1a04a.png)

## ジョブの実行

### 1. scheduleDeviceMethod ジョブの実行

#### 1-1. ジョブの作成

Datbricks にてノートブックを作成し、ライブラリをインストールします。

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

ジョブの作成を実施します。

```python
iot_conn_str = "HostName=iot-hub-test-001.azure-devices.net;SharedAccessKeyName=iothubowner;SharedAccessKey=FUum/tiOgEuhye/JqAb+XXXXX="
device_id = "qiita_test_01"
```

```python
import os, uuid, datetime, time
from datetime import timezone, timedelta
from azure.iot.hub import IoTHubJobManager
from azure.iot.hub.models import JobRequest, Twin, TwinProperties, CloudToDeviceMethod

jm = IoTHubJobManager.from_connection_string(iot_conn_str)

# --- ① ダイレクト メソッド呼び出しジョブ ----------------------
method_job_id = str(uuid.uuid4())
query_condition = f"deviceId IN ['{device_id}']"
method_req = JobRequest(
    job_id  = method_job_id,
    type    = "scheduleDeviceMethod",
    query_condition = query_condition,
    start_time = (datetime.datetime.now(timezone.utc) + timedelta(seconds=5)).isoformat(),   # 必ず UTC & 未来
    max_execution_time_in_seconds = 300,
    cloud_to_device_method = CloudToDeviceMethod(
        method_name  = "lockDoor",
        payload      = {"lockTime": "15m"},
        response_timeout_in_seconds = 30,   # ★ 必須
        connect_timeout_in_seconds  = 5     # ★ 必須
    ),
)
print("JobRequest: ")
print(method_req.as_dict())

jm.create_scheduled_job(method_job_id, method_req)


# --- 状態ポーリング（例） --------------------------------------
print("")
print("Job Status: ")
for _ in range(10):
    m = jm.get_scheduled_job(method_job_id).status
    print("method:", m)
    if m in ("completed", "failed"):
        break
    time.sleep(10)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0b3bbc29-df0e-4d77-954e-84fb7b4bc735.png)


#### 1-2. ジョブの実行

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_01;SharedAccessKey=r2fxXUdn/4w9iNJdhqnIJZ+h0qxxxxx="
```

```python
import os, time
from datetime import datetime, timezone
from azure.iot.device import IoTHubDeviceClient, MethodResponse

CONN = device_conn_str
client = IoTHubDeviceClient.create_from_connection_string(device_conn_str)


# --- ダイレクトメソッド "lockDoor" を受信 ---
def method_handler(request):
    if request.name == "lockDoor":
        print("🔐 Locking door for", request.payload.get("lockTime"))
        resp = MethodResponse.create_from_method_request(request, 200, {"result": "OK"})
        client.send_method_response(resp)

client.on_method_request_received = method_handler
client.connect()
print("Device connected – waiting for jobs …")

time.sleep(60)
client.shutdown()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bea68a6d-5ade-4c83-a898-dc3ad704e7d4.png)


ジョブの作成を実施したノートブックのセル結果をみると`completed`と表示されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dd749e2b-29a5-4c32-8eae-ed806ae5c159.png)

Azure IoT Hub のポータルにて、IoT Hub クエリ言語により結果を確認することもできます。

```sql
SELECT * FROM devices.jobs WHERE deviceId = 'qiita_test_01'
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/231097fe-aadb-4f18-83f6-1cfb34bbee6f.png)

### 2. scheduleUpdateTwin ジョブの実行


#### 2-1. ジョブの作成

Datbricks にてノートブックを作成し、ライブラリをインストールします。

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

ジョブの作成を実施します。

```python
iot_conn_str = "HostName=iot-hub-test-001.azure-devices.net;SharedAccessKeyName=iothubowner;SharedAccessKey=FUum/tiOgEuhye/JqAb+XXXXX="
device_id = "qiita_test_01"
```

```python
import os, uuid, datetime, time
from datetime import timezone, timedelta
from azure.iot.hub import IoTHubJobManager
from azure.iot.hub.models import JobRequest, Twin, TwinProperties, CloudToDeviceMethod

# --- ② Twin Desired 一括更新ジョブ -----------------------------
twin_job_id = str(uuid.uuid4())
twin_req = JobRequest(
    job_id  = twin_job_id,
    type    = "scheduleUpdateTwin",
    query_condition = f"deviceId IN ['{device_id}']",
    start_time = method_req.start_time,
    max_execution_time_in_seconds = 300,
    update_twin = Twin(
        etag="*",
        properties=TwinProperties(
            desired={"fwVersion": "2.1.0", "building": 43, "floor": 3}
        ),
    ),
)
print("JobRequest: ")
print(twin_req.as_dict())
jm.create_scheduled_job(twin_job_id, twin_req)

# --- 状態ポーリング（例） --------------------------------------
print("")
print("Job Status: ")
for _ in range(10):
    t = jm.get_scheduled_job(twin_job_id).status
    print("twin:", t)
    if t in ("completed", "failed"):
        break
    time.sleep(10)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b74c6e2e-80f5-4dfb-ad4c-cb11f02e6080.png)

#### 2-2. ジョブの実行

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_01;SharedAccessKey=r2fxXUdn/4w9iNJdhqnIJZ+h0qxxxxx="
```

```python
import os, time
from datetime import datetime, timezone
from azure.iot.device import IoTHubDeviceClient, MethodResponse

client = IoTHubDeviceClient.create_from_connection_string(device_conn_str)


# --- Twin Desired パッチを受信 ---
def twin_patch_handler(patch):
    print("📦 Desired patch:", patch)
    client.patch_twin_reported_properties(
        {
            "lastAppliedUtc": datetime.now(timezone.utc).isoformat(),
        }
    )


client.on_twin_desired_properties_patch_received = twin_patch_handler
client.connect()
print("Device connected – waiting for jobs …")

time.sleep(60)
client.shutdown()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1782274a-1516-442d-b965-42eba3c75590.png)


Azure IoT Hub のポータルにて、IoT Hub クエリ言語により結果を確認できます。

```sql
SELECT * FROM devices.jobs WHERE deviceId = 'qiita_test_01'
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d61cfbcd-b665-4e63-a443-4a0f9313109d.png)