Databricks のデータフレームにて Microsoft Fabric における EventStream の Source のカスタムエンドポイント から SAS 認証によりデータを書き込む方法


## 概要

Databricks のデータフレームにて Microsoft Fabric (Fabric) における EventStream の Source のカスタムエンドポイント から SAS 認証によりデータを書き込む方法を共有します。

## 前提条件

下記を構築します。

- Databricks
- Microsoft Fabric

## 事前準備

### 1. Fabric のワークスペースにて Source のカスタムエンドポイントを持つEventStream を作成し公開

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7870b135-22d9-42ac-bb3f-9df8dbc6a6ac.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/73814f91-ba0a-4dde-8aa6-206aae31ce84.png)


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/24e83543-a3e1-4a9e-94db-c2e37f8f5e21.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/09d8b4be-f520-4d43-bb60-8f431b960153.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5497f413-3c49-4ad3-8278-d4b35124998a.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/afee81e3-e8aa-4444-93ad-819a9a4a31ef.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1eb9a16d-0c77-479a-9c71-598d6e048a63.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/26b7c667-fb8b-498b-932e-891f7f330d34.png)

## 手順

### 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. Fabric にて SAS を取得

作成したソース -> `Kafka` -> `SASキー認証`を選択し 、下記の値を取得します。

- サーバー名（ブーストトラップサーバーの`xxx..servicebus.windows.net:9093`における`xxx`の値）
- トピック名
- 接続文字列 - 主キー

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/82562259-bfa6-4aea-9643-341e711e7f4d.png)

### 2. Databriks にて Azure Event Hubs からデータを取得

Eventstream カスタムエンドポイント に接続するための設定値をセットします。

```python
# Eventstream カスタムエンドポイント に接続するための設定値をセット
esc_namespace_name = "esehsgi702j38of92v5kv7"  ## Eventstream カスタムエンドポイント の名前をセット
topic_name = "es_26f0a637-fa91-45c6-8061-5f832de539bf"  ## Eventstream カスタムエンドポイントのトピックをセット
connection_string = "Endpoint=sb://esehsgi702j38of92v5kv7.servicebus.windows.net/;SharedAccessKeyName=key_70d0a645-5784-4d26-9851-0370c4ded6a6;SharedAccessKey=zMIkewSb2=;EntityPath=es_26f0a637-fa91-45c6-8061-5f832de539bf" ## 接続文字列 
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2301dd30-322e-43e5-af1d-dfc48ad96c24.png)

Eventstream カスタムエンドポイントに書き込みを実施します。

```python
# Databricks から Eventstream カスタムエンドポイント に設定する際の設定をセット
eh_sasl = (
    "kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule "
    f'required username="$ConnectionString" password="{connection_string}";'
)
bootstrap_servers = f"{esc_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,
}


# 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/c4acbf57-8fdd-4602-a7f4-54b13ee3cf5d.png)

Fabric の Eventstream にて書き込まれたことを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c3ea6075-f4a1-45cd-be41-a506544453e8.png)


:::note info

EventStream の Source のカスタムエンドポイントでは、外部からの読み込みをサポートしていないようです。

> java.util.concurrent.ExecutionException: kafkashaded.org.apache.kafka.common.errors.TimeoutException: Call(callName=listOffsets on broker 0, deadlineMs=1747366497859, tries=1, nextAllowedTryMs=1747366497960) timed out at 1747366497860 after 1 attempt(s)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c1578656-67f5-4385-8c45-2a8e4bab0efe.png)


:::