Databricks のデータフレームにて Azure Event Hubs へ OAuth 認証によりデータを書き込む方法

## 概要

Databricks のデータフレームにて Azure Event Hubs へ OAuth 認証によりデータを書き込む方法を紹介します。

## 前提条件

下記を構築します。

- Databricks
- Azure Event Hubs 名前空間（本手順では`kafka-test-01`という名称）
- Azure Event Hubs インスタンス （本手順では`dbx-to-eh-10`という名称）

## 手順

### 1. Databricks ソースのテーブルと Volumes を作成

カレントカタログを設定します。

```sql
%sql
USE CATALOG manabian_test_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/46c6531c-41d9-4d7d-8d9f-2a17f615ebb6.png)

スキーマを作成します。

```sql
%sql
CREATE SCHEMA IF NOT EXISTS eh_test_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cbde6cc2-6bb2-4cc8-b0fd-da146ae174ca.png)


ソースのテーブルを作成します。

```sql
%sql
CREATE OR REPLACE TABLE eh_test_01.table_01
AS
SELECT
  1 AS id,
  'John' AS string_col,
  CAST('2025-01-01' AS TIMESTAMP) AS ingest_ts
  ;
SELECT * FROM eh_test_01.table_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c738eb65-de53-4a08-b5ce-b545db513c36.png)

Volume を作成し、初期化します。

```sql
%sql
CREATE VOLUME IF NOT EXISTS eh_test_01.volume_01;
```

```python
volume_path = "/Volumes/manabian_test_01/eh_test_01/volume_01"
dbutils.fs.rm(volume_path, True)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dfaa7a69-deb5-41c0-91f9-00bde81a5b58.png)

### 2. Azure Entra ID にてアプリを作成

Azure Poral にて Azure Entra ID を表示し、`アプリの登録` -> `+ 新規登録`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/adeccf1c-1837-4205-b46b-f8ea4336e4d1.png)

名前欄を入力し、`登録`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/da5de324-3b0d-4581-8ea0-60572e615c52.png)

作成したアプリにおける`アプリケーション (クライアント) ID`と`ディレクトリ (テナント) ID`をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/33a01f09-1934-4a7c-bbe3-63b20324d0ba.png)

`証明書とシークレット` -> `新しいクライアント シークレット`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f9f4aa73-69dd-4879-9bbd-11e19edcb066.png)

`追加`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/81903f00-d518-458c-87ee-27b3d164bb76.png)

表示されているシークレットの値をコピーします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c492305b-8dd5-469e-9c6f-2f03554a7f4e.png)

### 3. Azure Event Hubs にてアプリに対して`Azure Event Hubs のデータ送信者`権限を付与

Azure Event Hubs 名前空間（あるいは、 Azure Event Hubs インスタンス）にて、`アクセス制御 (IAM)` -> `+ 追加` -> `ロールの割り当ての追加`を選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3d7d86c6-a9b1-45a7-9151-0f0ca6bb8468.png)

`Azure Event Hubs のデータ送信者`権限を作成したアプリに付与します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/36bb7a67-7343-491e-9586-195ab9ad5a4f.png)


### 4. Databriks にて Azure Event Hubs へデータの書き込み

Azure Event Hubs に接続するための設定値をセットします。

```python
# Azure Event Hubs に接続するための設定値をセット
eh_namespace_name = "kafka-test-01"  ## Azure Event Hubs の名前空間をセット
ehi_name = topic_name = "dbx-to-eh-10"  ## Azure Event Hubs のEvent Hubs のインスタンスをセット

tenant_id = "095ca098-XXXX-XXX-XXX-XXX"  # Azure Entra ID の Tenant ID をセット
client_id = "214cee56-7fd6-418d-bd21-aded919113b4"  # Azure Entra ID の Application (クライアント) ID をセット
client_secret = "pIj8Q~fPMueU3YHXzyYmXXXXXX"  # Azure Entra ID の Application (クライアント) Secret をセット
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/10bd475a-7fba-4cb6-833a-b823c01606f0.png)

Azure Event Hubs へデータを書きこみます。

```python
# Databricks から Azure Event Hubs に設定する際の設定をセット
bootstrap_servers = f"{eh_namespace_name}.servicebus.windows.net"
sasl_config = ""
sasl_config = (
    "kafkashaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required "
    f'clientId="{client_id}" '
    f'clientSecret="{client_secret}" '
    f'scope="https://{bootstrap_servers}/.default" ssl.protocol="SSL";'
)

kafka_options = {
    "kafka.bootstrap.servers": f"{bootstrap_servers}:9093",
    "kafka.sasl.jaas.config": sasl_config,
    "kafka.sasl.oauthbearer.token.endpoint.url": f"https://login.microsoft.com/{tenant_id}/oauth2/v2.0/token",
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.mechanism": "OAUTHBEARER",
    "kafka.sasl.login.callback.handler.class": "kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler",
}

# Azure Event Hubs からデータを取得
from pyspark.sql.functions import col, expr, from_json

df = spark.readStream.table("eh_test_01.table_01")
df = df.selectExpr("to_json(struct(id, string_col, ingest_ts)) AS value")


_ = (
    df.writeStream.format("kafka")
    .options(**kafka_options)
    .option("topic", topic_name)
    .option("checkpointLocation", volume_path+"/table_01")
    .trigger(availableNow=True)
    .start()
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/96964b19-238d-474a-88f4-6151f55fa5cd.png)


Azure Event Hubs の Data Explorere にてデータが書き込まれていることを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dc56093f-55a5-4c6e-8677-83b4242c1a11.png)