SQL Server および Azure SQL 用の Apache Spark コネクタを使用する

APACHE Spark Connector for SQL Server と Azure SQL は、ビッグ データ分析でトランザクション データを使用し、アドホック クエリまたはレポートの結果を保持できる高パフォーマンス コネクタです。 コネクタを使用すると、オンプレミスまたはクラウド内の任意の SQL データベースを、Spark ジョブの入力データ ソースまたは出力データ シンクとして使用できます。 コネクタでは、SQL Server の一括書き込み API が使用されます。 一括書き込みパラメーターは、ユーザーが省略可能なパラメーターとして渡すことができます。また、コネクタによって基になる API に as-is 渡されます。 一括書き込み操作の詳細については、「 JDBC ドライバーでの一括コピーの使用」を参照してください。

コネクタは、SQL Server ビッグ データ クラスターに既定で含まれています。

オープン ソース リポジトリでコネクタの詳細を確認します。 例については、 サンプルを参照してください。

新しい SQL テーブルへの書き込み

server_name = "jdbc:sqlserver://{SERVER_ADDR}"
database_name = "database_name"
url = server_name + ";" + "databaseName=" + database_name + ";"

table_name = "table_name"
username = "username"
password = "password123!#" # Please specify password here

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("overwrite") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

SQL テーブルへの追加

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("append") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

分離レベルを指定する

このコネクタは、既定では、データベースへの一括挿入を実行するときにREAD_COMMITTED分離レベルを使用します。 これを別の分離レベルにオーバーライドする場合は、次に示すように mssqlIsolationLevel オプションを使用してください。

    .option("mssqlIsolationLevel", "READ_UNCOMMITTED") \

SQL テーブルからの読み取り

jdbcDF = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("user", username) \
        .option("password", password).load()

Active Directory 以外のモード

Active Directory モード以外のセキュリティでは、各ユーザーは、読み取りまたは書き込みを実行するために、コネクタのインスタンス化中にパラメーターとして指定する必要があるユーザー名とパスワードを持っています。

Active Directory 以外のモードのコネクタのインスタンス化の例を次に示します。 スクリプトを実行する前に、 ? をアカウントの値に置き換えます。

# Note: '?' is a placeholder for a necessary user-specified value
connector_type = "com.microsoft.sqlserver.jdbc.spark" 

url = "jdbc:sqlserver://master-p-svc;databaseName=?;"
writer = df.write \ 
   .format(connector_type)\ 
   .mode("overwrite") 
   .option("url", url) \ 
   .option("user", ?) \ 
   .option("password",?) 
writer.save() 

Active Directory モード

Active Directory モードのセキュリティでは、ユーザーがキー タブ ファイルを生成した後、コネクタのインスタンス化中にパラメーターとして principal と keytab を指定する必要があります。

このモードでは、ドライバーは、それぞれの Executor コンテナーに keytab ファイルを読み込みます。 次に、Executor はプリンシパル名と keytab を使用して、読み取り/書き込み用の JDBC コネクタを作成するために使用されるトークンを生成します。

Active Directory モードのコネクタのインスタンス化の例を次に示します。 スクリプトを実行する前に、 ? をアカウントの値に置き換えます。

# Note: '?' is a placeholder for a necessary user-specified value
connector_type = "com.microsoft.sqlserver.jdbc.spark"

url = "jdbc:sqlserver://master-p-svc;databaseName=?;integratedSecurity=true;authenticationScheme=JavaKerberos;" 
writer = df.write \ 
   .format(connector_type)\ 
   .mode("overwrite") 
   .option("url", url) \ 
   .option("principal", ?) \ 
   .option("keytab", ?)   

writer.save() 

次のステップ

ビッグ データ クラスターの詳細については、Kubernetes に SQL Server ビッグ データ クラスター を展開する方法に関するページを参照してください

SQL Server ビッグ データ クラスターに関するフィードバックや機能に関する推奨事項をお持ちですか? SQL Server ビッグ データ クラスターのフィードバックに関するメモをお寄せください。