「Cloud Dataflow＆Cloud Scheduler＆CData JDBC Driver」でSaaS-BigQueryの連携ジョブをスケジューリングしてみた

CData Software Advent Calendar 2020 2日目の記事です。

## はじめに
本内容はCloud Dataflow の「JDBC to BigQuery テンプレート」上でCData JDBC Driver を利用して、SaaS データをBigQueryに連携し、そのジョブをCloud Scheduler に登録して定期実行していく内容となります。  
Dataflow のテンプレートを使用しますので特にコードを書かずに（SQLはあります）実現することができます。  
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/d83d49f6-10e3-2e32-eeb8-381e458d54c6.png)

SaaS データの部分は今回はMarketoのLeadデータを連携していきますが、Marketo 以外に接続したい場合は以下より接続先を確認できます。  
https://www.cdata.com/jp/jdbc/

## ジョブ作成手順
やることは以下の内容です。

- CData Marketo JDBC Driver のインストール
- CData Marketo JDBC Driver をGCSにアップロード
- BigQuery に連携用テーブルを作成
- Cloud Dataflow でジョブを作成
- Cloud Scheduler でジョブをスケジューリング

### CData Marketo JDBC Driver のインストール
ますは以下のリンクよりJDBC Driver をダウンロードします。（30日間の無償評価版があります）  
https://www.cdata.com/jp/drivers/marketo/jdbc/
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/33c6f7cb-ea4e-410b-ea32-1b1ad9cc85f4.png)  
  
ダウンロードしたら、jarファイルを実行してインストールします。インストールが完了すると、Windows の場合は以下のパスにjarファイルが配置されます。  
C:\Program Files\CData\CData JDBC Driver for Marketo 2020J\lib 
  
### MarketoデータをDriverで事前確認
必須ではないですが後程BigQueryに連携用テーブルを作成しますので、事前にMarketoにJDBCで接続して取得できるテーブル（Marketoのオブジェクト）情報を確認しておくほうが良いかと思います。
一応ヘルプドキュメントからもスキーマ情報は取得できます。
http://cdn.cdata.com/help/DKF/jp/jdbc/pg_restalltables.htm

JDBC Driver の利用方法はJavaアプリケーションで直接使用していただくほかに、DBVisualizer などのDB 接続ツールで確認することもできますので今回はDBVisualizer というツールで確認していきます。

DBVisualizer は以下リンクよりダウンロードすることができます。
www.dbvis.com

インストール後、CData Marketo JDBC Driver のヘルプドキュメントにも例はありますが、接続時に使用する接続文字列は以下のような形式で設定します。

```
jdbc:marketo:Schema=REST;RESTEndpoint=https://XXX-XXX-XXX.mktorest.com/rest;OAuthClientID=XXXXXX;OAuthClientSecret=XXXXXX;InitiateOAuth=GETANDREFRESH;
```
実際に接続してみると、このようにMarketo のデータがRDBライクに参照することができます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/c411caea-752c-026e-8a23-57b160487db2.png)

ここでDataflow で実行したいSQL（取得項目、条件条件など）を確認しておきます。

今回は全項目ではなく、先頭に実行日時カラム＆項目をいくつか絞ってやっていきます。

```sql
select cast(CURRENT_TIMESTAMP() as varchar) as dataflow_execute,Id,company,Website,NumberOfEmployees,Industry,MktoCompanyNotes,AnnualRevenue,
FirstName,MiddleName,LastName,Email,Phone,MobilePhone,Fax,Address,
cast(CreatedAt as varchar) as CreatedAt,cast(UpdatedAt as varchar) as UpdatedAt 
from Leads  
```
ちなみに日時型の項目のままDataflowで実行すると、内部でUnixタイムに変換されてBigQuery にインサートする際に失敗してしまっていたので、文字列にキャストしておきます。

### JDBC DriverをGCSにアップロード
Dataflow ではGoogle Cloud Storage に配置したJDBC Driverを参照していきますので、先ほどインストールしたJDBC Driver をGCSにアップロードします。  

cdatadriver というバケットを作成し、その配下にJDBC Driver をアップロードしました。ついでにDataflow が使用するフォルダも２つ作成しました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/8fe7f578-6c92-0df5-2ee8-175da3d46b64.png)

### BigQueryに連携用テーブルを作成
データの型まで合わせた状態で、BigQuery に連携用テーブルを作成します。
データ型の確認は先ほどのDBVisualizer などのDB 接続ツールで確認するか、ヘルプを参照ください。
（カスタムオブジェクトの場合はヘルプに記載はないので、DB 接続ツールなどで確認する必要があります）
http://cdn.cdata.com/help/DKF/jp/jdbc/pg_resttable-leads.htm

BigQuery にテーブルを作成しましたら次に進みます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/67e7db5d-424f-1cb2-f31a-12d1c7e7fb58.png)


### クラウドサービス上でCData Driverを利用する際の注意点
クラウドサービスなどにJDBC ドライバーをアップロードして使用する場合には、RTK というライセンスを使用する必要があります。
ただ、RTK についてはCData サポートチームから直接ご提供させていただくことになりますので、以下のサポートフォームより「Cloud Dataflow で使用したい」とお問合せください。
https://www.cdata.com/jp/support/


### Cloud Dataflowでジョブ作成
それではDataflow でジョブを作成していきます。Dataflow の画面より「テンプレートからジョブを作成」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/02eaca32-05e2-1687-a2de-21dd6e067ab9.png)
ジョブ名、リージョンの設定し、Dataflow テンプレートでは「Jdbc to BigQuery」 を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/8f70d99d-495a-ab42-93f9-9a5301256e8d.png)
必須パラメータにそれぞれ以下の内容を設定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/c8f2da04-5d6f-1798-5305-783d0a7cdba1.png)

**入力例**

|項目&nbsp;|設定値|メモ&nbsp;|
|:--- |:--|:--|
|Connection URL String|jdbc:marketo:Schema=REST;RESTEndpoint=https://XXX-XXX-XXX.mktorest.com/rest;OAuthClientID=XXXXX;OAuthClientSecret=XXXXX;InitiateOAuth=GETANDREFRESH;RTK=XXXXXXX|サポートチームより受領したRTKをセット|
|Driver Class|cdata.jdbc.marketo.MarketoDriver|ヘルプにも記載してます|
|Query|select ~ ※長いので省略|「Marketo ドライバーを利用して事前にデータを確認」で使ったクエリ|
|BigQuery Output Table|dataflow-cdata:demo.marketo_leads|データセット名．テーブル名|
|GCS paths for Jdbc drivers|gs://cdatadriver/cdata.jdbc.marketo.jar|CData JDBC Driver を配置したパス|
|Temporary Path|gs://cdatadriver/temp|JDBC Driver 配置した際に作成したフォルダ|
|一時的なロケーション|gs://cdatadriver/temp2|JDBC Driver 配置した際に作成したフォルダ|


設定はこれだけですので、画面下部より実行ボタンをおして実際に連携をしてみます。


#### Cloud Dataflowでジョブ実行
正常に連携できるかジョブを実行して確認してみます。実行ボタンクリック後、少し待つとこのようにジョブステータスに完了しましたのメッセージが表示されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/25759ac1-870b-0d2c-a6b4-5255c06c56d6.png)

BigQuery のコンソールを見てみると、Marketo のLeadデータがBigQuery に連携されているのが確認できました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/527a5520-2eee-896c-ebe0-c8dbaca99083.png)

### Cloud Schedulerでジョブを定期実行
Dataflow 単体ではジョブを定期的に実行することができないので、Cloud Scheduler からDataflow のAPIをキックするようにして、定期実行していきます。

では、実行済みのDataflowから生成されたAPIを確認します。
クローン → RESTをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/388b0a66-f252-fc22-484a-d018c51dc821.png)

そうすると表示されたRESTリクエストが表示されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/2220adca-008a-5511-bca7-51174dda6e60.png)

残すはCloud Schedulerの設定です。赤枠部分には先ほど確認した内容を設定します。URL部分には確認した内容の先頭行"/v1b3~"以降をコピーして、"https\://dataflow.googleapis.com"のあとに設定します。
本文には残りの部分をそのまま貼り付けます。
あとは頻度のところで1時間おきに実行するよう指定しました。
※タイムゾーンのところは文字化けされてますが、とりあえずJSTで設定。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/7a3507d0-e8fc-ed56-61d9-b0dc3b97970a.png)

設定が完了したらあとは画面下部の作成ボタン（更新ボタン）をクリックします。 

最後にジョブの実行を行います。右にある「今すぐ実行」ボタンをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/4a6f6db8-8203-70a6-858e-45c4c0a4a44e.png)
設定内容に問題がなければ、このようにステータスが成功になります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/085f227f-c8df-2891-a989-63d53136ef45.png)

Dataflowを参照すると・・・・
ちゃんと実行されています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/e0d06562-1fb9-90a6-c870-e33fa928aae0.png)

これでCloud Schedulerからの実行が確認できました。あとは定期的に実行できているか確認します。今回は1時間ごとに実行するよう設定しています。

しばらく経ってから再度確認すると、前回の実行時間が更新されていました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/12550d6c-69c9-133b-784a-7b923eff2660.png)

Dataflowもみると、1時間に1回実行されています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/378300/4d1dbdee-767a-f8d1-8321-e16ece3f43a3.png)

こんな感じで、DataflowのJDBC to BigQuery テンプレートとCloud Scheduler、CData JDBC DriverでさくっとSaaSデータの連携処理が作成できます。