DatabricksとAzure Synapse Analyticsの連携

[Azure Synapse Analytics \| Databricks on AWS](https://docs.databricks.com/data/data-sources/azure/synapse-analytics.html) [2021/3/18]の翻訳です。

[Azure Synapse Analytics](https://azure.microsoft.com/services/synapse-analytics/)(以前のSQL Data Warehouse)は、ペタバイトのデータに対する複雑なクエリーを高速するために、大量並列処理(MPP)を活用するクラウドベースのエンタープライズデータウェアハウスです。Azureでのビッグデータにおけるキーコンポーネントとなっています。シンプルな[PolyBase](https://docs.microsoft.com/azure/synapse-analytics/sql-data-warehouse/load-data-wideworldimportersdw)のT-SQLあるいは[COPY](https://docs.microsoft.com/sql/t-sql/statements/copy-into-transact-sql)文を用いてビッグデータをAzureにインポートし、MPPのパワーを活用して高性能の分析を実行します。統合、分析を進めていくことで、このデータウェアハウスが、ビジネス洞察を得る際に信頼できる単一の真実(the single version of truth)になることでしょう。

Azure Synapseコネクターを用いることで、DatabricksからAzure Synapseにアクセスすることができます。Azure Synapseコネクターは、[Azure Blob storage](https://azure.microsoft.com/services/storage/blobs/)、PolyBase、Azure Synapseにおける`COPY`文を使用するApache Sparkに対するデータソース実装であり、Azure SynapseインスタンスとDatabricksクラスター間で効率的に大量データを転送することができます。

二つのシステム間でデータをやり取りするために、DatabricksクラスターとAzure Synapseインスタンスは、共通のBlobストレージコンテナーにアクセスします。Databricksでは、Blobストレージコンテナーにデータを読み書きするために、Azure SynapseコネクターによってApache Sparkジョブが起動されます。Azure Synapse側では、JDBCを通じてAzure Synapseコネクターによって、PolyBaseによるデータのローディング、アンローディングの操作が起動されます。Databricksランタイム7.0以降では、JDBC経由でのAzure SynapseコネクターによるAzure Synapseへのデータロードには、デフォルトで`COPY`が使用されます。

> **注意**
[より高い性能](https://docs.microsoft.com/azure/sql-data-warehouse/upgrade-to-latest-generation)を提供するAzure Synapse Gen2インスタンスでのみ、`COPY`を利用できます。お持ちのデータベースがGen1インスタンスを使用しているのであれば、Gen2に移行することをお勧めします。

Azure Synapseコネクターは、インタラクティブなクエリーよりも、クエリー実行において大量のデータをBlobストレージから抽出するETLに適しています。同じAzure Syapseテーブルに繰り返しクエリーを実行する際には、抽出したデータをParquetなどのフォーマットに保存することをお勧めします。

# 要件

Azure Synapseに対する[database master key](https://docs.microsoft.com/sql/relational-databases/security/encryption/create-a-database-master-key)が必要となります。

# 承認

Azure Synapseコネクターは3種類のネットワークコネクションを使用します。

- SparkドライバーからAzure Synapseへの接続
- SparkドライバーとエグゼキューターからAzureストレージアカウントへの接続
- Azure SynapseからAzureストレージアカウントへの接続

```
                                 ┌─────────┐
      ┌─────────────────────────>│ STORAGE │<────────────────────────┐
      │   Storage acc key /      │ ACCOUNT │  Storage acc key /      │
      │   Managed Service ID /   └─────────┘  OAuth 2.0 /            │
      │                               │                              │
      │                               │                              │
      │                               │ Storage acc key /            │
      │                               │ OAuth 2.0 /                  │
      │                               │                              │
      v                               v                       ┌──────v────┐
┌──────────┐                      ┌──────────┐                │┌──────────┴┐
│ Synapse  │                      │  Spark   │                ││ Spark     │
│ Analytics│<────────────────────>│  Driver  │<───────────────>│ Executors │
└──────────┘  JDBC with           └──────────┘    Configured   └───────────┘
              username & password /                in Spark
```

以下の章では、それぞれのコネクションの認証設定方法を説明します。

## SparkドライバーからAzure Synapseへの接続

Sparkドライバーは、ユーザー名とパスワードを伴うJDBC接続、あるいは認証のためのサービスプリンシパルを伴うOAuth 2.0で、Azure Synapseに接続することができます。

### ユーザー名とパスワード

どちらの認証方式においても、Azureポータルで提供される接続用文字列を使用することをお勧めします。これによって、SparkドライバーとAzure Synapseインスタンスとの間でJDBC経由でやり取りされる全てのデータがSecure Sockets Layer (SSL)で暗号化されます。SSL暗号化が有効化されているかどうかは、接続用文字列に`encrypt=true`があることで確認できます。

SparkドライバーがAzure Synapseに到達できるようにするためには、AzureポータルからアクセスできるAzure Synapseサーバーのfirewallペインで、**Allow access to Azure services**を**ON**に設定します。この設定により、全てのAzure IPアドレスとAzureサブネットからの通信を許可し、SparkドライバーがAzure Synapseインスタンスに到達できるようになります。

### サービスプリンシパルとOAuth 2.0

> **プレビュー**
この機能は[パブリックプレビュー](https://docs.databricks.com/release-notes/release-types.html)です。

サービスプリンシパルを用いることで、Azure Synapse Analyticsが背後にあるストレージアカウントへのアクセスを許可することができます。Azureストレージアカウントにアクセスするサービスプリンシパルの認証情報の使い方に関しては、[Access Azure Data Lake Storage Gen2 using OAuth 2\.0 with an Azure service principal](https://docs.databricks.com/data/data-sources/azure/adls-gen2/azure-datalake-gen2-sp-access.html)を参照ください。コネクターに対するサービスプリンシパルによる承認を有効化するには、`enableServicePrincipalAuth`オプションを`true`に設定する必要があります。

Azure Synapse Analytics接続に対して、異なるサービスプリンシパルを用いることができます。Synapseにおける、ストレージアカウントに対するサービスプリンシパルの例を以下に示します。

**ini**

```ini
; Defining the Service Principal credentials for the Azure storage account
fs.azure.account.auth.type OAuth
fs.azure.account.oauth.provider.type org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider
fs.azure.account.oauth2.client.id <application-id>
fs.azure.account.oauth2.client.secret <service-credential>
fs.azure.account.oauth2.client.endpoint https://login.microsoftonline.com/<directory-id>/oauth2/token

; Defining a separate set of service principal credentials for Azure Synapse Analytics (If not defined, the connector will use the Azure storage account credentials)
spark.databricks.sqldw.jdbc.service.principal.client.id <application-id>
spark.databricks.sqldw.jdbc.service.principal.client.secret <service-credential>
```

**Scala**

```Scala
// Defining the Service Principal credentials for the Azure storage account
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", "https://login.microsoftonline.com/<directory-id>/oauth2/token")

// Defining a separate set of service principal credentials for Azure Synapse Analytics (If not defined, the connector will use the Azure storage account credentials)
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>")
```

**Python**

```python
# Defining the service principal credentials for the Azure storage account
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", "https://login.microsoftonline.com/<directory-id>/oauth2/token")

# Defining a separate set of service principal credentials for Azure Synapse Analytics (If not defined, the connector will use the Azure storage account credentials)
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>")
```

**R**

```R
# Load SparkR
library(SparkR)
conf <- sparkR.callJMethod(sparkR.session(), "conf")

# Defining the service principal credentials for the Azure storage account
sparkR.callJMethod(conf, "set", "fs.azure.account.auth.type", "OAuth")
sparkR.callJMethod(conf, "set", "fs.azure.account.oauth.provider.type",  "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
sparkR.callJMethod(conf, "set", "fs.azure.account.oauth2.client.id", "<application-id>")
sparkR.callJMethod(conf, "set", "fs.azure.account.oauth2.client.secret", "<service-credential>")
sparkR.callJMethod(conf, "set", "fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<directory-id>/oauth2/token")

# Defining a separate set of service principal credentials for Azure Synapse Analytics (If not defined, the connector will use the Azure storage account credentials)
sparkR.callJMethod(conf, "set", "spark.databricks.sqldw.jdbc.service.principal.client.id", "<application-id>")
sparkR.callJMethod(conf, "set", "spark.databricks.sqldw.jdbc.service.principal.client.secret", "<service-credential>")
```

## SparkドライバーとエグゼキューターからAzureストレージアカウントへの接続

Azureストレージコンテナーは、Azure Synapseに対してバルクデータを読み書きする際に、中間層として動作します。Sparkは、ビルトインコネクター：[Azure Blob storage](https://docs.databricks.com/data/data-sources/azure/azure-storage.html#azure-storage)、[Azure Data Lake Storage \(ADLS\) Gen2](https://docs.databricks.com/data/data-sources/azure/adls-gen2/index.html)の一つを用いてストレージコンテナーに接続します。[Azure Data Lake Storage Gen1](https://docs.databricks.com/data/data-sources/azure/azure-datalake.html#adls-gen1)はサポートされておらず、SSLで暗号化されたHTTPアクセスのみが許可されます。このため、サポートされているURIスキーマは`wasbs`と`abfss`です。

以下の認証方法を利用できます：

- Azure Blob storage(`wasbs`)
    - ストレージアカウントのアクセスキーとシークレット
- Azure Data Lake Storage Gen2(`abfss`)
    - ストレージアカウントのアクセスキーとシークレット
    - OAuth 2.0認証。OAuth 2.0認証とサービスプリンシパル、[Access Azure Data Lake Storage Gen2 using OAuth 2\.0 with an Azure service principal](https://docs.databricks.com/data/data-sources/azure/adls-gen2/azure-datalake-gen2-sp-access.html)を参照ください。

以下の例では、ストレージアカウントキーのアプローチにおける二つの方法を説明します。これはOAuth 2.0においても適用できます。

### ノートブックセッションの設定(推奨)

このアプローチでは、コマンドを実行するノートブックに関連づけられたセッション設定にアカウントのアクセスキーが設定されます。この設定は、同じクラスターにアタッチされている他のノートブックには影響を与えません。`spark`がノートブックで提供される`SparkSession`オブジェクトとなります。

**Python**

```python
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net",
  "<your-storage-account-access-key>")
```

### グローバルHadoop設定

このアプローチでは、全てのノートブックで共有される`SparkContext`に関連づけられるHadoop設定を更新します。

**Scala**

```Scala
sc.hadoopConfiguration.set(
  "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net",
  "<your-storage-account-access-key>")
```

**Python**

`hadoopConfiguration`は全てのバージョンのPySparkでは外部に公開されていません。以下のコマンドはある意味Sparkの内部機構に依存しています。このため、これは全てのPySparkで動作するはずですが、動作が将来的に保証されている訳ではありません。

```python
sc._jsc.hadoopConfiguration().set(
  "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net",
  "<your-storage-account-access-key>")
```

## Azure SynapseからAzureストレージアカウントへの接続

Azure Synapseは一時データのロード、アンロードの際にもストレージアカウントに接続します。

ストレージアカウントに対するアカウントキーとシークレットを設定した場合には、`forwardSparkAzureStorageCredentials`を`true`に設定することができます。これにより、Azure Synapseはノートブックのセッション設定あるいはグローバルHadoop設定にあるアカウントアクセスキーとシークレットを探し出し、ストレージアカウントのアクセスキーを転送し、Azureの[database scoped credential](https://docs.microsoft.com/sql/t-sql/statements/create-database-scoped-credential-transact-sql)を作成することで、Azure Synapseインスタンスに接続できるようになります。

ADLS Gen2 + OAuth 2.0認証を使用している場合、Azure SynapseインスタンスがManaged Service Identity(多くの場合、[VNet \+ Service Endpoints setup](https://azure.microsoft.com/blog/general-availability-of-vnet-service-endpoints-for-azure-sql-data-warehouse/)の組み合わせ)を持つように設定されている場合には、`useAzureMSI `を`true`に設定する必要があります。この場合、コネクターは、データベーススコープの認証情報に`IDENTITY = 'Managed Service Identity'`を設定し、`SECRET`は設定しません。

# ストリーミングのサポート

Azure Synapseインスタンス、Databricksクラスター間で大量データをやり取りするためにPolyBaseや`COPY`を用いて一貫性のあるユーザーエクスペリエンスを提供するAzure Synpseに対して、Azure Synapseコネクターは、効率的かつスケーラブルな構造化ストリーミングによる書き込みもサポートしています。バッチによる書き込みと同様、ストリーミングは主にETL向けに設定されており遅延が大きくなるケースがあり、リアルタイムデータ処理には適さないケースがあります。

## フォールトトレランス

デフォルトでは、Azure Synapseストリーミングは、エンドツーエンドで*exactly-once*の書き込みを保証します。これには、DBFSに格納されるチェックポイントとAzure Synapseテーブルに格納されるチェックポイント、ロック機構を組み合わせて、あらゆる種類の失敗、リトライ、クエリーの再起動をハンドリングできるようにしています。オプションで、`spark.databricks.sqldw.streaming.exactlyOnce.enabled`を`false`に設定することで、Azure Synapseストリーミングのat-least-onceの制限を緩和することができます。この場合、Azure Synapseに対する切断、予期せぬクエリーの停止などで、データの重複が起こり得ます。

# 利用方法(バッチ)

Scala、Python、SQL、RのノートブックからデータソースAPIを通じてコネクターを利用することができます。

**Scala**

```Scala
// Otherwise, set up the Blob storage account access key in the notebook session conf.
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net",
  "<your-storage-account-access-key>")

// Get some data from an Azure Synapse table.
val df: DataFrame = spark.read
  .format("com.databricks.spark.sqldw")
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>")
  .option("tempDir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .option("dbTable", "<your-table-name>")
  .load()

// Load data from an Azure Synapse query.
val df: DataFrame = spark.read
  .format("com.databricks.spark.sqldw")
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>")
  .option("tempDir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .option("query", "select x, count(*) as cnt from table group by x")
  .load()

// Apply some transformations to the data, then use the
// Data Source API to write the data back to another table in Azure Synapse.

df.write
  .format("com.databricks.spark.sqldw")
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .option("dbTable", "<your-table-name>")
  .option("tempDir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>")
  .save()
```

**Python**

```python

# Otherwise, set up the Blob storage account access key in the notebook session conf.
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net",
  "<your-storage-account-access-key>")

# Get some data from an Azure Synapse table.
df = spark.read \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
  .option("tempDir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "<your-table-name>") \
  .load()

# Load data from an Azure Synapse query.
df = spark.read \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
  .option("tempDir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("query", "select x, count(*) as cnt from table group by x") \
  .load()

# Apply some transformations to the data, then use the
# Data Source API to write the data back to another table in Azure Synapse.

df.write \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "<your-table-name>") \
  .option("tempDir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>") \
  .save()
```

**SQL**

```sql
-- Otherwise, set up the Blob storage account access key in the notebook session conf.
SET fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net=<your-storage-account-access-key>;

-- Read data using SQL.
CREATE TABLE example_table_in_spark_read
USING com.databricks.spark.sqldw
OPTIONS (
  url 'jdbc:sqlserver://<the-rest-of-the-connection-string>',
  forwardSparkAzureStorageCredentials 'true',
  dbTable '<your-table-name>',
  tempDir 'wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>'
);

-- Write data using SQL.
-- Create a new table, throwing an error if a table with the same name already exists:

CREATE TABLE example_table_in_spark_write
USING com.databricks.spark.sqldw
OPTIONS (
  url 'jdbc:sqlserver://<the-rest-of-the-connection-string>',
  forwardSparkAzureStorageCredentials 'true',
  dbTable '<your-table-name>',
  tempDir 'wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>'
)
AS SELECT * FROM table_to_save_in_spark;
```

**R**

```R
# Load SparkR
library(SparkR)

# Otherwise, set up the Blob storage account access key in the notebook session conf.
conf <- sparkR.callJMethod(sparkR.session(), "conf")
sparkR.callJMethod(conf, "set", "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net", "<your-storage-account-access-key>")

# Get some data from an Azure Synapse table.
df <- read.df(
   source = "com.databricks.spark.sqldw",
   url = "jdbc:sqlserver://<the-rest-of-the-connection-string>",
   forward_spark_azure_storage_credentials = "true",
   dbTable = "<your-table-name>",
   tempDir = "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>")

# Load data from an Azure Synapse query.
df <- read.df(
   source = "com.databricks.spark.sqldw",
   url = "jdbc:sqlserver://<the-rest-of-the-connection-string>",
   forward_spark_azure_storage_credentials = "true",
   query = "select x, count(*) as cnt from table group by x",
   tempDir = "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>")

# Apply some transformations to the data, then use the
# Data Source API to write the data back to another table in Azure Synapse.

write.df(
  df,
  source = "com.databricks.spark.sqldw",
  url = "jdbc:sqlserver://<the-rest-of-the-connection-string>",
  forward_spark_azure_storage_credentials = "true",
  dbTable = "<your-table-name>",
  tempDir = "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>")
```

# 利用方法(ストリーミング)

Scala、Pythonノートブックから構造化ストリーミングを用いてデータを書き込むことができます。

**Scala**

```Scala
// Set up the Blob storage account access key in the notebook session conf.
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net",
  "<your-storage-account-access-key>")

// Prepare streaming source; this could be Kafka or a simple rate stream.
val df: DataFrame = spark.readStream
  .format("rate")
  .option("rowsPerSecond", "100000")
  .option("numPartitions", "16")
  .load()

// Apply some transformations to the data then use
// Structured Streaming API to continuously write the data to a table in Azure Synapse.

df.writeStream
  .format("com.databricks.spark.sqldw")
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>")
  .option("tempDir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>")
  .option("forwardSparkAzureStorageCredentials", "true")
  .option("dbTable", "<your-table-name>")
  .option("checkpointLocation", "/tmp_checkpoint_location")
  .start()
```

**Python**

```python
# Set up the Blob storage account access key in the notebook session conf.
spark.conf.set(
  "fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net",
  "<your-storage-account-access-key>")

# Prepare streaming source; this could be Kafka or a simple rate stream.
df = spark.readStream \
  .format("rate") \
  .option("rowsPerSecond", "100000") \
  .option("numPartitions", "16") \
  .load()

# Apply some transformations to the data then use
# Structured Streaming API to continuously write the data to a table in Azure Synapse.

df.writeStream \
  .format("com.databricks.spark.sqldw") \
  .option("url", "jdbc:sqlserver://<the-rest-of-the-connection-string>") \
  .option("tempDir", "wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>") \
  .option("forwardSparkAzureStorageCredentials", "true") \
  .option("dbTable", "<your-table-name>") \
  .option("checkpointLocation", "/tmp_checkpoint_location") \
  .start()
```

# 設定

本章では、コネクターに対する書き込み設定、要求されるアクセス権、その他の設定パラメーターをご説明します。

- [バッチ書き込みにおけるセーブモード](#バッチ書き込みにおけるセーブモード)
- [ストリーミング書き込みにおけるアウトプットモード](#ストリーミング書き込みにおけるアウトプットモード)
- [書き込みのセマンティクス](#書き込みのセマンティクス)
- [PolyBaseに必要なAzure Synapseの権限](#polybaseに必要なazure-synapseの権限)
- [`COPY`に必要なAzure Synapseの権限](#copyに必要なazure-synapseの権限)
- [パラメーター](#パラメーター)
- [Azure Synapseへのクエリープッシュダウン](#azure-synapseへのクエリープッシュダウン)
- [テンポラリーデータの管理](#テンポラリーデータの管理)
- [テンポラリーオブジェクトの管理](#テンポラリーデータの管理)
- [ストリーミングチェックポイントテーブルの管理](#ストリーミングチェックポイントテーブルの管理)

## バッチ書き込みにおけるセーブモード

Azure Synapseコネクターは、`ErrorIfExists`、`Ignore`、`Append`、`Overwrite`セーブモードをサポートしており、デフォルトは`ErrorIfExists`です。詳細は[Spark SQL documentation on Save Modes](https://spark.apache.org/docs/latest/sql-data-sources-load-save-functions.html#save-modes)を参照ください。

## ストリーミング書き込みにおけるアウトプットモード

Azure Synapseコネクターは、レコード追加と集計のための`Append`、`Complete`のアウトプットモードをサポートしています。アウトプットモードと互換性に関する詳細は、[Structured Streaming guide](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#output-modes)を参照ください。

## 書き込みのセマンティクス

PolyBaseに加えて、Azure Synapseコネクターは`COPY`文をサポートしています。`COPY`文は、外部テーブルを作成することなしにデータロードが可能で、データロードに必要な権限が少なくてすみ、Azure Synapseに対して高速なデータ投入を可能とする便利な方法を提供します。

デフォルトでは、コネクターは自動で適切な書き込みセマンティクスを検出します。しかし、以下の設定で書き込みセマンティクスを強制することができます。

**Scala**

```Scala
// Configure the write semantics for Azure Synapse connector in the notebook session conf.
spark.conf.set("spark.databricks.sqldw.writeSemantics", "<write-semantics>")
```

**Python**

```python
# Configure the write semantics for Azure Synapse connector in the notebook session conf.
spark.conf.set("spark.databricks.sqldw.writeSemantics", "<write-semantics>")
```

**SQL**

```sql
-- Configure the write semantics for Azure Synapse connector in the notebook session conf.
SET spark.databricks.sqldw.writeSemantics=<write-semantics>;
```

**R**

```R
# Load SparkR
library(SparkR)

# Configure the write semantics for Azure Synapse connector in the notebook session conf.
conf <- sparkR.callJMethod(sparkR.session(), "conf")
sparkR.callJMethod(conf, "set", "spark.databricks.sqldw.writeSemantics", "<write-semantics>")
```

ここで、`<write-semantics>`は、

- `polybase` (Azure SynapseへのデータロードにPolyBaseを使用)
- `copy` (Databricksランタイム7.0以降で、データロードに`COPY`を使用)
- 指定なし (デフォルト設定に従う：Databricksランタイム7.0以降でADLS Gen2であれば`COPY`、そうでない場合は`polybase`)

## PolyBaseに必要なAzure Synapseの権限

PolyBaseを使用する際、Azure Synapseコネクターは、接続先のAzure Synapseインスタンスで以下のコマンドを実行できるユーザーによるJDBC接続を必要とします。

- [CREATE DATABASE SCOPED CREDENTIAL](https://docs.microsoft.com/sql/t-sql/statements/create-database-scoped-credential-transact-sql)
- [CREATE EXTERNAL DATA SOURCE](https://docs.microsoft.com/sql/t-sql/statements/create-external-data-source-transact-sql)
- [CREATE EXTERNAL FILE FORMAT](https://docs.microsoft.com/sql/t-sql/statements/create-external-file-format-transact-sql)
- [CREATE EXTERNAL TABLE](https://docs.microsoft.com/sql/t-sql/statements/create-external-table-transact-sql)

最初のコマンドを実行する前に、既にデータベースマスターキーが特定のAzure Synapseインスタンスに存在している必要があります。そうでない場合には、[CREATE MASTER KEY](https://docs.microsoft.com/sql/t-sql/statements/create-master-key-transact-sql)コマンドでキーを作成する必要があります。

加えて、`dbTable`を通じて設定された、あるいは、`query`で参照されるAzure Synapseテーブルを読み込むには、JDBCユーザーは、当該のAzure Synapseテーブルに対する権限を有している必要があります。`dbTable`を通じてAzure Synapseテーブルにデータを書き戻す場合には、JDBCユーザーはAzure Synapseテーブルに対する書き込み権限を有している必要があります。

以下の表にPolyBaseの全ての操作に関わる権限をまとめています。

| オペレーション | 権限 |
|:--|:--|
|バッチ書き込み   |CONTROL   |
|ストリーミング書き込み   |CONTROL   |
|読み取り   |CONTROL   |

## `COPY`に必要なAzure Synapseの権限

> **注意**
本機能はDatabricksランタイム7.0以上で利用できます。

`COPY`文を使用する際には、Azure Synapseコネクターは、接続先のAzure Synapseインスタンスで以下のコマンドを実行できるユーザーによるJDBC接続を必要とします。

- [COPY INTO](https://docs.microsoft.com/sql/t-sql/statements/copy-into-transact-sql)

Azure Synapseにターゲットのテーブルが存在しない場合、上のコマンドの実行権限に加えて、以下のテーブルに記載されている権限が必要となります。

- [CREATE TABLE](https://docs.microsoft.com/sql/t-sql/statements/create-table-azure-sql-data-warehouse)

以下のテーブルに、`COPY`で行うバッチ処理、ストリーミング処理に必要な権限をまとめています。

| オペレーション | 権限(存在するテーブルへのインサート) | 権限(新規テーブルへのインサート) |
|:--|:--|:--|
|バッチ書き込み   |ADMINISTER DATABASE BULK OPERATIONS<br><br>INSERT   |ADMINISTER DATABASE BULKOPERATIONS<br><br>INSERT<br><br>CREATE TABLE<br><br>ALTER ON SCHEMA :: dbo   |
|ストリーミング書き込み   |ADMINISTER DATABASE BULK OPERATIONS<br><br>INSERT   |ADMINISTER DATABASE BULKOPERATIONS<br><br>INSERT<br><br>CREATE TABLE<br><br>ALTER ON SCHEMA :: dbo   |

## パラメーター

パラメーター、Spark SQLにおける`OPTION`で設定可能な内容についてはこちらを[参照](https://docs.databricks.com/data/data-sources/azure/synapse-analytics.html#parameters)ください。

## Azure Synapseへのクエリープッシュダウン

Azure Synapseコネクターは、以下のオペレーターをAzure Synapseにプッシュダウンする最適化ルールを実装しています。

- `Filter`
- `Project`
- `Limit`

`Project`と`Filter`オペレーターは以下の表現をサポートしています。

- 多くのブールロジックオペレーター
- 比較
- 基礎的な計算オペレーション
- 数値、文字列のキャスト

`Limit`オペレーターに対しては、`ORDER`が指定されない場合にのみプッシュダウンがサポートされます。例えば、

`SELECT TOP(10) * FROM table`ではプッシュダウンが行われますが、`SELECT TOP(10) * FROM table ORDER BY col`ではプッシュダウンが行われません。

> **注意**
Azure Synapseコネクターは、文字列、日付、タイムスタンプに対する条件はプッシュダウンしません。

Azure Synapseコネクターで実装されるクエリープッシュダウンはデフォルトで有効化されています。`spark.databricks.sqldw.pushdown`を`false`に設定することで無効化することができます。

## テンポラリーデータの管理

Azure Synapseコネクターは、Blobストレージコンテナーに作成されるテンポラリーファイルを削除しません。このため、ユーザーが指定した`tempDir`配下にある一時ファイルを定期的に削除することをお勧めします。

データのクリーンアップを容易にするために、Azure Synapseコネクターは`tempDir`直下にデータファイルを格納しません。代わりに`<tempDir>/<yyyy-MM-dd>/<HH-mm-ss-SSS>/<randomUUID>/`の形式でサブディレクトリを作成します。定期的ジョブ(Databricksの[ジョブ](https://docs.databricks.com/jobs.html)機能などを活用)をセットアップして、特定の期間(例えば2日)を経過した古いサブディレクトリを削除することができます。

よりシンプルな代替案としては、定期的にコンテナー全体を削除して、同じ名前のコンテナーを再作成するというものです。これには、Azure Synapseコネクターが生成する一時データ専用のコンテナーと、コネクターが関わるクエリーが実行されていないことを保証できる時間帯が必要となります。

## テンポラリーオブジェクトの管理

Azure Synapseコネクターは、Azure SynapseインスタンスとDatabricksクラスター間のデータ転送を自動化します。Azure Synapseテーブルからデータを読み込んだり、Azures Synapseテーブルに書き込んだりする際には、Azure Synapseコネクターは、`DATABASE SCOPED CREDENTIAL`、`EXTERNAL DATA SOURCE`、`EXTERNAL FILE FORMAT`、`EXTERNAL TABLE`を含むテンポラリーオブジェクトを生成します。これらのオブジェクトは、対応するSparkジョブが存在している間のみ生存し、その後は自動的に削除されます。

Azure Synapseコネクターを用いたクエリーがクラスターで実行される際、Sparkドライバーがクラッシュ、あるいは、強制的に再起動された場合、または、クラスターが強制的に停止、再起動された場合、テンポラリーオブジェクトは削除されずに残ってしまう場合があります。この場合に、手動での削除を容易にするために、Azure Synapseコネクターは、Azure Synapseインスタンスで作成された全ての中間一時オブジェクトに対して、`tmp_<yyyy_MM_dd_HH_mm_ss_SSS>_<randomUUID>_`の形式で名前を付けます。

定期的に以下のようなクエリーを実行することで、削除されていないオブジェクトがないか確認することをお勧めします。

- `SELECT * FROM sys.database_scoped_credentials WHERE name LIKE 'tmp_databricks_%'`
- `SELECT * FROM sys.external_data_sources WHERE name LIKE 'tmp_databricks_%'`
- `SELECT * FROM sys.external_file_formats WHERE name LIKE 'tmp_databricks_%'`
- `SELECT * FROM sys.external_tables WHERE name LIKE 'tmp_databricks_%'`

## ストリーミングチェックポイントテーブルの管理

Azure Synapseコネクターは、新たなストリーミングクエリーが開始した際に作成されるストリーミングチェックポイントテーブルを削除しません。これはDBFSにおける`checkpointLocation`の挙動と同じです。このため、定期的に将来的に実行する予定が無いクエリーのDBFS上のチェックポイントを削除するのと同じタイミングでチェックポイントテーブルを削除することをお勧めします。

デフォルトでは、チェックポイントテーブルの名称は`<prefix>_<query_id>`となっています。`<prefix>`は設定することができ、デフォルトは`databricks_streaming_checkpoint`となっています。`query_id`はストリーミングクエリーIDから`_`を除外したものです。古い、あるいは削除されたストリーミングクエリーに対するチェックポイントテーブルを検索するには、以下のクエリーを実行します。

**SQL**

```sql
SELECT * FROM sys.tables WHERE name LIKE 'databricks_streaming_checkpoint%'
```

Spark SQLの設定オプション`spark.databricks.sqldw.streaming.exactlyOnce.checkpointTableNamePrefix`でプレフィクスを変更することができます。

# FAQ

**Azure Synapseコネクターを使っている時にエラーが発生しました。どうすればエラーがAzure Synapseによるものなのか、Databricksによるものなのかを識別できますか？**

エラーのデバッグを容易にするために、Azure Synapseコネクター特有のコードからスローされた全ての例外は、`SqlDWException`トレイトを拡張した例外でラップされています。また、以下のように例外を識別することが可能です。

- `SqlDWConnectorException` Azure Synapseコネクターによってスローされた例外
- `SqlDWSideException` 接続先のAzure Synapseインスタンスによってスローされた例外

**“No access key found in the session conf or the global Hadoop conf”というエラーが発生してクエリーに失敗した時にはどうすればいいですか？**

このエラーは、`tempDir`で指定されたストレージアカウントに対するアクセキーが、ノートブックのセッション設定やHadoop設定に見つからなかったことを意味します。どのようにストレージアカウントへのアクセスを設定するのかに関しては、[利用方法(バッチ)](#利用方法バッチ)を参照ください。Sparkテーブルが、Azure Synapseコネクターで作成された場合には、Sparkテーブルに読み書きするためにストレージアカウントの認証情報を提供する必要があります。

**`tempDir`に指定されたBlobストレージコンテナーにアクセスするために、Shared Access Signature (SAS)を使用することはできますか？**

Azure Synapseは[SASを用いたBlobストレージへのアクセス](https://docs.microsoft.com/sql/t-sql/statements/create-database-scoped-credential-transact-sql#b-creating-a-database-scoped-credential-for-a-shared-access-signature)をサポートしていません。このため、Azure Synapseコネクターは`tempDir`で指定されたBlobストレージコンテナーにアクセスするために[SAS](https://docs.microsoft.com/azure/storage/common/storage-dotnet-shared-access-signature-part-1)を使用することはできません。

**`dbTable`オプションを指定してAzure SynapseコネクターでSparkテーブルを作成し、Sparkテーブルに書き込みを行い、Sparkテーブルを削除しました。この場合、Azure Synapse側のテーブルも削除されますか？**

いいえ。Azure Synapseは外部データソースと看做されます。`dbTable`を通じて設定されたAzure Synapseテーブルは、Sparkテーブルが削除されても削除されません。

**Azure Synapseにデータフレームを書き込む際に、`.saveAsTable(tableName)`ではなく`.option("dbTable", tableName).save()`を指定しなくてはならないのですか？**

以下の区別を明確にするためです：

- `.option("dbTable", tableName)`はデータベース(Azure Synapse)のテーブルを参照します。
- `.saveAsTable(tableName)`はSparkテーブルを参照します。

実際のところは、`df.write. ... .option("dbTable", tableNameDW).saveAsTable(tableNameSpark)`のように二つを組み合わせることができます。これによって、Azure Synapseには`tableNameDW`を作成し、Sparkの外部テーブルには、Azure Synapseにバックアップがある`tableNameSpark `を作成します。

> **警告!**
以下の`.save()`と`.saveAsTable()`の違いに注意してください。

- `df.write. ... .option("dbTable", tableNameDW).mode(writeMode).save()`においては、`writeMode `は予想通りAzure Synapseテーブルに対して作用します。
- `df.write. ... .option("dbTable", tableNameDW).mode(writeMode).saveAsTable(tableNameSpark)`においては、`writeMode`はSparkテーブルに作用します。`tableNameDW`がAzure Synapseに存在する際には、サイレントで上書きされます。

この振る舞いは、異なるデータソースに対して書き込みを行う際にも同様です。これは、Sparkの[DataFrameWriter API](https://github.com/apache/spark/blob/v2.3.0/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala)を使用する際のちょっとしたTipsとなります。

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)