Databricks にて Azure  Synapse Analytics Dedicated SQL (専用 SQL プール) から Spark コネクター経由でデータを取得する方法の検証

## 概要

Databricks にて Azure  Synapse Analytics Dedicated SQL (専用 SQL プール) から Spark コネクター経由でデータを取得する方法の検証結果を共有します。

専用 SQL プールに対して Spark コネクター経由でデータ交換を行う場合には、下記図のようになります。Databricks(Spark) <-> Storage <-> 専用 SQL プール というようにデータのやり取りを実施するため、それぞれの認証設定が必要となります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9a2f11eb-4119-6280-036e-26e7fe022046.png)

引用元：[Azure Synapse Analytics - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/external-data/synapse-analytics#--authentication)

3 つのサービス間での認証方法としては、下記表の方法があります。

| #    | 認証経路                     | 認証方法                                                     |
| ---- | ---------------------------- | ------------------------------------------------------------ |
| 1    | databricks --> 専用 SQL プール       | 1. Service Principal<br/>2. SQL 認証              |
| 2    | 専用 SQL プール --> Azure Storage    | 1. Managed Service ID<br/>2. Storage Access Key<br />3. Databricks で保持している Storage Access Key *1 |
| 3    | databricks --> Azure Storage | 1. Service Principal<br/>2. Storage Access Key *2<br /> |


*1 Databirkcs にてデータフレーム操作時に `forwardSparkAzureStorageCredentials` を `true` を設定することで実行できる方法

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/588699db-9252-ce61-4379-7a9034fa433e.png)

引用元：[Azure Synapse Analytics - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/external-data/synapse-analytics#parameters)

*2 SAS での認証はサポートされていないようです。

> Azure Synapse では、SAS を使用した BLOB ストレージへのアクセスはサポートされていません。 

引用元：[Azure Synapse Analytics - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/external-data/synapse-analytics#frequently-asked-questions-faq)

`synapse --> Azure Storage` のサービスでは 基本的には `Managed Service ID` による認証を選択することになるため、Databricksにおける認証にてサービスプリンシパルを利用できるかが認証方法の検討の要因となり、次のいずれかを選択することが多くなりそうです。

**1. Databricks にてサービスプリンシパルをベースに認証する方法**

| #    | 認証経路                     | 認証方法                                                     |
| ---- | ---------------------------- | ------------------------------------------------------------ |
| 1    | databricks --> 専用 SQL プール       | 2. Service Principal              |
| 2    | 専用 SQL プール --> Azure Storage    | 1. Managed Service ID |
| 3    | databricks --> Azure Storage | 1. Service Principal |

**2. Databricks にてそれぞれのサービスにおける認証機能をベースに認証する方法**

| #    | 認証経路                     | 認証方法                                                     |
| ---- | ---------------------------- | ------------------------------------------------------------ |
| 1    | databricks --> 専用 SQL プール       | 1. SQL 認証              |
| 2    | 専用 SQL プール --> Azure Storage    | 1. Managed Service ID |
| 3    | databricks --> Azure Storage | 2. Storage Access Key |


Azure Storage 上に一時的なデータが残ってしまうため、定期的に削除することが必要となります。[ライフサイクル管理ポリシー](https://learn.microsoft.com/ja-jp/azure/storage/blobs/lifecycle-management-overview)を利用するなどの削除方法を検討してください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/95a34143-8757-01ab-7026-71c395f69257.png)

引用元：[Azure Synapse Analytics - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/external-data/synapse-analytics#--temporary-data-management)

**参考リンク**

-   [チュートリアル:Azure portal および SSMS を使用してデータを読み込む - Azure Synapse Analytics | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/synapse-analytics/sql-data-warehouse/load-data-wideworldimportersdw)
-   [Azure Databricks で Azure Data Lake Storage Gen2 と Blob Storage にアクセスする - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/external-data/azure-storage)
-   [Azure SQL Database にサービス プリンシパル ユーザーを作成する - Azure SQL Database | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/azure-sql/database/authentication-aad-service-principal-tutorial?view=azuresql#create-the-service-principal-user-in-azure-sql-database)


## 動作検証

### 0. 事前準備

#### 0-1. Azure リソースを作成

- Azure Synapse Analytics 
- 専用 SQL プール（Azure Synapse Analytics 内に作成）
- データAzure Storage （階層型名前空間が有効）
- Azure Databricks

#### 0-2. Azure Synapse Analytics のマネージド ID に Azure Storageへの `ストレージ BLOB データ共同作成者` ロールを付与

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/202ac042-8426-2382-56db-4c15d461ed27.png)

#### 0-3. 専用 SQL プールにてサンプル用テーブルを作成

```sql
IF OBJECT_ID(N'dbo.unload_test') IS NOT NULL
DROP TABLE dbo.unload_test
;

CREATE TABLE dbo.unload_test (
  int_col int
)
;

INSERT INTO dbo.unload_test
SELECT 1 AS col_int
UNION ALL
SELECT 2 AS col_int
UNION ALL
SELECT 3 AS col_int
UNION ALL
SELECT 4 AS col_int
UNION ALL
SELECT 5 AS col_int
;

SELECT COUNT(*) FROM dbo.unload_test;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1569b5f7-a200-358a-4787-b2d11aff01c6.png)

### 1.  Databricks にてサービスプリンシパルをベースに認証する方法**

#### 1-1. サービスプリンシパルの作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/aa5092db-8a32-902e-a3a5-6b0ff9d9448a.png)

#### 1-2. サービスプリンシパルに対して Azure Storageへの `ストレージ BLOB データ共同作成者` ロールを付与

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6e048d70-2d24-b72e-30e3-2481de5aebb3.png)

#### 1-3. サービスプリンシパルに対して 専用 SQL プールにおけるユーザーとして登録後、 `db_owner` ロールを付与

```sql
CREATE USER syanpse_data_unload FROM EXTERNAL PROVIDER
GO

EXEC sp_addrolemember 'db_owner', [syanpse_data_unload]
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f715741b-e6c3-9c46-4deb-0ca2ec42d458.png)


#### 1-4. サービスプリンシパルの`ディレクトリ (テナント) ID`、 `アプリケーション (クライアント) ` 、および、`シークレットの値` を取得

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9db8c31e-62a8-ef12-2923-53da20b8b6c9.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c0a11a22-1efe-a0fc-04eb-badca4cad924.png)

#### 1-5. Databricks にて認証情報を設定後、専用 SQL プールからデータを取得

```python
# サービスプリンシパルの情報をセット
directory_id = '095ca098-c723-478f-9bb6-c764031b159e'
application_id = '06774ac4-e863-409f-b30e-7fcd8a5b1121'
service_credential = 'jN68Q~SbQBNAogXhJLui0OccN'

# Azure storage への Databricks にて利用する認証情報を設定
spark.conf.set("fs.azure.account.auth.type", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type",  "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.id", application_id)
spark.conf.set("fs.azure.account.oauth2.client.secret", service_credential)
spark.conf.set("fs.azure.account.oauth2.client.endpoint", f"https://login.microsoftonline.com/{directory_id}/oauth2/token")

# 専用 SQL プールへの Databricks にて利用する認証情報を設定
spark.conf.set("spark.databricks.sqldw.jdbc.service.principal.client.id", application_id)
spark.conf.set("spark.databricks.sqldw.jdbc.service.principal.client.secret", service_credential)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/47b49806-1dbe-41a5-c2ba-393652230562.png)


```python
# Azure Storage の基本情報をセット
storage_account = 'synapseunload1234455323'
container_name = 'unload'
```

```python
# Synapse 専用 SQL の基本情報をセット
synapse_server_name = 'synapse-unload'
synapse_db_name = 'unload'

jdbc_url = ''
jdbc_url += f'jdbc:sqlserver://{synapse_server_name}.sql.azuresynapse.net:1433;'
jdbc_url += f'database={synapse_db_name};'
jdbc_url += f'encrypt=true;'
jdbc_url += f'trustServerCertificate=false;'
jdbc_url += f'hostNameInCertificate=*.sql.azuresynapse.net;loginTimeout=30;'
print(jdbc_url)
```

```python
# 処理で利用する情報をセット
src_tbl_name = 'dbo.unload_test'
tmp_dir = '/unload/dbo/unload_test'
temp_dir = f"abfss://{container_name}@{storage_account}.dfs.core.windows.net/{tmp_dir}"

print(temp_dir)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3bdd4c45-b763-0039-d357-15e9214c3da8.png)

```python
# 専用 SQL プールからデータを取得して表示
df = (
    spark.read.format("com.databricks.spark.sqldw")
    .option("url", jdbc_url)
    .option("tempDir", temp_dir)
    .option("enableServicePrincipalAuth", "true")
    .option("useAzureMSI", "true")
    .option("dbTable", src_tbl_name)
    .load()
)

df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c1a032ef-0b20-d235-1dc5-f0a7bdbd9145.png)

### 2. Databricks にてそれぞれのサービスにおける認証機能をベースに認証する方法

#### 2-1. Azure Storageのアクセスキーを取得

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/800d9955-c3fa-04e2-78b7-83e8096f66b7.png)

#### 2-2. Databricks にて認証情報を設定後、専用 SQL プールからデータを取得

```python
# Azure Storage の基本情報をセット
storage_account = 'synapseunload1234455323'
storage_account_access_key = 'Sb5LjlpdKcCmJ5UOrMdS8SR28uhgLg2lu+BxGXrcers8C8vgdNdZPnxVCwknx7+'

spark.conf.set(
  f"fs.azure.account.key.{storage_account}.dfs.core.windows.net",
  storage_account_access_key,
)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0f9dd3d2-457c-1977-f182-3b1caef699d9.png)


```python
# Azure Storage の基本情報をセット
storage_account = 'synapseunload1234455323'
container_name = 'unload'
```

```python
# Synapse 専用 SQL の基本情報をセット
synapse_server_name = 'synapse-unload'
synapse_db_name = 'unload'

jdbc_url = ''
jdbc_url += f'jdbc:sqlserver://{synapse_server_name}.sql.azuresynapse.net:1433;'
jdbc_url += f'database={synapse_db_name};'
jdbc_url += f'encrypt=true;'
jdbc_url += f'trustServerCertificate=false;'
jdbc_url += f'hostNameInCertificate=*.sql.azuresynapse.net;loginTimeout=30;'
print(jdbc_url)

user = 'sqladminuser@synapse-unload'
password = 'PH_$Vw9#F`'
```

```python
# 処理で利用する情報をセット
src_tbl_name = 'dbo.unload_test'
tmp_dir = '/unload/dbo/unload_test'
temp_dir = f"abfss://{container_name}@{storage_account}.dfs.core.windows.net/{tmp_dir}"

print(temp_dir)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d8926139-3b48-066b-8bd6-36f7ef61f003.png)

```python
# 専用 SQL プールからデータを取得して表示
df = (
    spark.read.format("com.databricks.spark.sqldw")
    .option("url", jdbc_url)
    .option("tempDir", temp_dir)
    .option("useAzureMSI", "true")
    .option("user", user)
    .option("password", password)
    .option("dbTable", src_tbl_name)
    .load()
)

df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5d7bb290-61d9-0b37-db9f-02824e114c47.png)