Spark（Google Dataproc）からMicrosoft SQL Server（Azure SQL Database）にアクセスする方法

Spark（Google Dataproc）から、Microsoft SQL Server（Azure Database）にアクセスする方法です。

手順

Spark設定

下記のSparkの設定をすると、SparkからSQL Serverのデータの読み書きができるようになります。

MS SQL ServerのJDBCのjarファイルをダウンロードします
- https://docs.microsoft.com/ja-jp/sql/connect/jdbc/download-microsoft-jdbc-driver-for-sql-server?view=sql-server-2017
SparkにJDBCのjarファイルを読み込みます
- 「spark-defaults.conf」の「spark.jars」にjarファイルのパスを設定します
SparkのClassPathにJDBCのjarファイルを設定します
- 「spark-defaults.conf」の「spark.driver.extraClassPath」にjarファイルのパスを設定します

Dataproc設定

Dataproceはマネージドサービスのため、Sparkを直接いじることができません。

代わりに、Dataproceの「クラスタープロパティ」と「初期化アクション」を使って、クラスター作成時にSparkの設定します。

事前に、MS SQL ServerのJDBCのjarファイルを、Google Cloud Storageにアップしておきます。

下記の初期化アクションを作って、Google Cloud Storageにアップします。

init_action.sh

#!/bin/bash

JAR_PATH=/usr/local/lib/jars
mkdir $JAR_PATH

# JDBC SQL Server
JDBC_MSSQL=mssql-jdbc-7.2.2.jre8.jar
gsutil cp gs://bnobq-olive-dataproc/init-action/jdbc/$JDBC_MSSQL $JAR_PATH/

初期化アクションでは、Sparkに追加するjarファイルの置き場所を作って、StorageにアップしたJDBCのjarファイルをダウンロードしています。

クラスター作成

Spark（Google Dataproc）からMicrosoft SQL Server（Azure Database）にアクセスする方法

初期化アクション
- 先程Google Cloud Storageにアップした初期化アクションファイルを指定します
- クラスター作成時に、このアクションが実行されます
プロジェクトアクセス
- チェックを入れる
- 初期化アクションで、StorageからJDBCのjarファイルをダウンロードしているため、クラスターにStorageへのアクセス権限を付与します
クラスタプロパティ
- [spark]：[spark.jars]：[/usr/local/lib/jars/*]
  - 「spark」でSparkの設定「spark-defaults.conf」を設定します
  - 「spark.jars」「/usr/local/lib/jars/*」で、SparkにJDBCのjarファイルを読み込ませます
- [spark]：[spark.driver.extraClassPath]：[/usr/local/lib/jars/*]
  - 「spark」でSparkの設定「spark-defaults.conf」を設定します
  - 「spark.driver.extraClassPath」「/usr/local/lib/jars/*」で、SparkにClassPass情報として、JDBCのjarファイルを読み込ませます
- [dataproc]：[dataproc.conscrypt.provider.enable]：[false]
  - 未設定の場合、ssl関連のエラーが出るため、「false」にセットします

利用例

options = {
    "url":"jdbc:sqlserver://<server>:1433;databaseName=<db_name>",
    "driver":"com.microsoft.sqlserver.jdbc.SQLServerDriver",
    "dbtable":"<table_name>",
    "user":"xxxx",
    "password":"xxxx"
}

df = spark.read.format("jdbc").options(**options).load()
df.show()