Databricks DLT にて Azure Event Hubs へ OAuth 認証によりデータを書き込む取得する方法

## 概要

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

## 前提条件

下記を構築します。

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

## 手順

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

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

```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)

### 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 に接続するための設定値をセットし、 DLT パイプラインを実行します。

```python
from dlt import *
from pyspark.sql.functions import *

# Azure Event Hubs に接続するための設定値をセット
eh_namespace_name = "kafka-test-01"  ## Azure Event Hubs の名前空間をセット
ehi_name = topic_name = "dbx-to-eh-12"  ## 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 をセット

# 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 へデータ書き込み
dbx_to_eh_12_options = kafka_options.copy()
dbx_to_eh_12_options["topic"] = topic_name
dlt.create_sink(name="eh_sink_01", format="kafka", options=dbx_to_eh_12_options)


@dlt.append_flow(name="kafka_sink_flow", target="eh_sink_01")
def kafka_sink_flow():
    df = spark.readStream.table("eh_test_01.table_01")
    df = df.selectExpr("to_json(struct(id, string_col, ingest_ts)) AS value")

    return df
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8124c925-07d3-42ed-870d-aa90820ab9c2.png)


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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2b090071-d2b1-4263-abd0-f6a9c2511dd6.png)