Databricks のデータフレームにて Microsoft Fabric における EventStream の Destination のカスタムエンドポイント から SAS 認証によりデータを取得する方法


## 概要

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

## 前提条件

下記を構築します。

- Databricks
- Microsoft Fabric

## 事前準備

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/931441e4-cc9d-44c8-9bf4-8c0e99d0551d.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/649bc26b-9d5c-4e84-90c0-c548ce137b9e.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3ab23405-1b7f-4120-a2ba-6028a69657a5.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d6f1d014-1d36-4153-9434-ad7f65259be6.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4a0bbc7c-c340-4de8-9a22-7364fc2ff8e2.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ebe7b3dc-e0d9-4d7c-97b2-73969bc2e16f.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fcea258a-17f6-475f-9712-e62d50f79b12.png)

## 手順

### 1. Fabric にて SAS を取得

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

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/894cd422-3d71-487b-80c5-0296c81e642f.png)


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

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

```python
# Eventstream カスタムエンドポイント に接続するための設定値をセット
esc_namespace_name = "esehsgi702j38of92v5kv7"  ## Eventstream カスタムエンドポイント の名前をセット
topic_name = "es_b5502d41-18b0-469c-87b1-a32b38e3a700"  ## Eventstream カスタムエンドポイントのトピックをセット
connection_string = "Endpoint=sb://esehsgi702j38of92v5kv7.servicebus.windows.net/;SharedAccessKeyName=key_70d0a645-5784-4d26-9851-0370c4ded6a6;SharedAccessKey=zMIkewSb2=;EntityPath=es_913f810e-0a03-4cba-98f2-e385d1bc9e21" ## 接続文字列 
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3b9987bb-2a3d-49ac-b4ec-d0a5c83f6518.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,
}

# Eventstream カスタムエンドポイント からデータを取得
from pyspark.sql.functions import col, expr, from_json

df = (
    spark.readStream.format("kafka")
    .options(**kafka_options)
    .option("subscribe", topic_name)
    .option("startingOffsets", "earliest")  # 検証目的であるため earliest に設定
    .load()
)
df = df.withColumn(
    "decoded_value",
    col("value").cast("string"),
)

df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8747e632-b78f-4d68-9050-cdc6fbf313a8.png)


:::note info

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4a3a7875-3e8b-4e8b-97c8-096ce038e0be.png)


> org.apache.spark.SparkException: Job aborted due to stage failure: kafkashaded.org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [es_913f810e-0a03-4cba-98f2-e385d1bc9e21]

:::