Apache Spark でSalesforce のデータをSQL で操作する方法

杉本和也
リードエンジニア

CData JDBC ドライバーを使用して、Apache Spark でSalesforce にデータ連携。

Apache Spark は大規模データ処理のための高速エンジンです。CData JDBC Driver for Salesforce と組み合わせると、Spark はリアルタイムでSalesforce のデータに連携して処理ができます。本記事では、Spark シェルに接続してSalesforce をクエリする方法について解説します。

CData JDBC Driver は、最適化されたデータ処理がドライバーに組み込まれているため、リアルタイムSalesforce と対話するための高いパフォーマンスを提供します。Salesforce に複雑なSQL クエリを発行すると、ドライバーはフィルタや集計など、サポートされているSQL操作を直接Salesforce にプッシュし、組込みSQL エンジンを使用してサポートされていない操作（SQL 関数やJOIN 操作）をクライアント側で処理します。組み込みの動的メタデータクエリを使用すると、ネイティブデータ型を使用してSalesforce を操作して分析できます。

CData JDBC Driver for Salesforce をインストール

まずは、本記事右側のサイドバーからSalesforce JDBC Driver の無償トライアルをダウンロード・インストールしてください。30日間無償で、製品版の全機能が使用できます。

Spark Shell を起動してSalesforce のデータに接続

ターミナルを開き、Spark shell でCData JDBC Driver for Salesforce JAR file をjars パラメータに設定します:
```
$ spark-shell --jars /CData/CData JDBC Driver for Salesforce/lib/cdata.jdbc.salesforce.jar
```
Shell でJDBC URL を使ってSalesforce に接続し、SQL Context load() function でテーブルを読み込みます。
Salesforce 接続プロパティの設定方法

埋め込みOAuth（UI でのログイン）による接続設定

それでは、Salesforce への接続について説明していきましょう。最も簡単な方法として、Salesforce にログインする際と同様にUI 上からログインするだけで接続設定が完了します（埋め込みOAuth）。この方法をご利用になる場合は、「Salesforce への接続」をクリックしてください。

標準認証の設定

埋め込みOAuth 以外の方法を利用する場合、以下の3つの認証方式をご利用いただけます。標準的な認証方式では、以下の情報が必要となります。
- ユーザー名
- パスワード
- セキュリティトークン
セキュリティトークンの取得方法については、セキュリティトークン取得手順をご確認ください。

OAuth 認証の設定

ユーザー名とパスワードによる認証がご利用いただけない（避けたい）場合は、OAuth 認証をお使いいただけます。

SSO（シングルサインオン）の設定

最後に、IDプロバイダー経由でのシングルサインオンをご利用になる場合は、以下のプロパティを設定してください。
- SSOProperties
- SSOLoginUrl
- TokenUrl
より詳細な設定手順については、ヘルプドキュメントの「はじめに」セクションをご確認ください。

組み込みの接続文字列デザイナー

JDBC 接続文字列URL の作成には、Salesforce JDBC Driver にビルトインされたデザイナを使用できます。JAR ファイルをダブルクリックするか、コマンドラインでJAR ファイルを実行するとデザイナが開きます。
```
java -jar cdata.jdbc.salesforce.jar
```
接続プロパティを入力し、接続文字列をクリップボードにコピーします。
```
scala> val salesforce_df = spark.sqlContext.read.format("jdbc").option("url", "jdbc:salesforce:User=username;Password=password;SecurityToken=Your_Security_Token;").option("dbtable","Account").option("driver","cdata.jdbc.salesforce.SalesforceDriver").load()
```
接続が完了し、データがロードされたら、テーブルスキーマが表示されます。
Salesforce をテンポラリーテーブルとして登録します:
```
scala> salesforce_df.registerTable("account")
```
データに対して、次のようなカスタムSQL クエリを実行します。
```
scala> salesforce_df.sqlContext.sql("SELECT Industry, AnnualRevenue FROM Account WHERE Name = GenePoint").collect.foreach(println)
```
コンソールで、次のようなSalesforce のデータを取得できました！これでSalesforce との連携は完了です。