こんにちは、CDataの宮本です。
前回の記事では、Cloud Data Fusion × CData Connect でSaaS 連携をご紹介しました。
今回は同じGCPサービスのCloud Dataflow のJDBC テンプレートを使って、ノーコードでCData Connect 経由でSaaS データを連携する内容でやってみます。
Google Cloud Dataflow とは？
バッチデータストリーミングとリアルタイムデータストリーミングの両方に対応するフルマネージド型のETL サービスです。Web分析やビッグデータ分析アプリケーションに見られるような大規模なデータセットを統合、準備、分析するための処理パイプラインを設定できます。
Cloud Dataflow ではJava やPython でApache Beam SDK を使用してデータパイプラインを定義・実行するプラットフォームになりますが、最初からいくつかのパターンでテンプレートが用意されています。
その中にサードパーティのJDBCドライバーを利用してBigQuery に連携するテンプレートが存在しているのでこれを利用して、SaaS のデータをBigQuery に連携してみます。
本記事のシナリオ
やることは前回のDataFusion の記事と同じく、Dynamics 365 （CRM）のデータをCData Connect 経由でMySQL インタフェースとして公開し、そこにDataflow からアクセスしてデータを取得 → BigQuery への連携を行っていきます。
※Dynamics 365（CRM）を選択してますが、以下のCData ドライバー一覧にあるものが利用できますので、以降の内容を違うデータソースに読み替えていただいても大丈夫です。
必要なもの
手順
CData Connect とDynamics 365（CRM）への接続や設定、MySQL のJDBC ドライバーのダウンロードは、前回の記事でご紹介した内容と全く同じなので、手順は「CData Connect でDynamics 365（CRM）への接続設定とエンドポイントの作成」と「MySQLドライバーのダウンロード」をご参照ください。
Google Cloud Data Fusion のSaaS 連携をCData Connect で拡張：Dynamics 365（CRM）編 - CData Software Blog
Cloud Storage の設定＆MySQL JDBC ドライバーの配置
Dataflow がデータパイプラインを実行中、一時ファイルなどをCloud Storage に書き込む必要があるので、Dataflow 用のディレクトリを用意しておきます。
また、MySQL の公式サイトからダウンロードしたMySQL JDBC ドライバーを、このタイミングでアップロードします。
今回はcdataconnect1 というバケットの中にMySQL JDBC ドライバーを格納し、
同じ階層にtemp とtemp2 というフォルダを作成しました。
BigQuery に連携先用テーブルを作成
BigQuery のコンソールを開き、今回連携したいデータのテーブルを新規作成していきます。
テーブル定義はMySQL WorkBench などでCData Connect へ接続いただき、それをもとにカラム名とデータ型を合わせて作成します。
MySQL WorkBench からCDataConnect への接続方法はこちらの記事を参考にしてください。
Dataflow の設定
それではさっそくDataflow でCData Connect へ接続していきましょう。
まずは、サイドメニューよりDataflow を選択し、「テンプレートからジョブを作成」をクリックします。
ジョブ名、リージョンの設定し、Dataflow テンプレートでは「Jdbc to BigQuery」 を選択します。
必須パラメータにそれぞれCData Connect への接続や、使用するGCS のURL を入力します。
入力例
接続文字列：
jdbc:mysql://<CDataConnectで作成されたMySQL エンドポイント（ポート部分は不要）>/?user=CDataConnect で作成したユーザー&password=利用ユーザーのパスワード クエリ：
select Id, Address1_Country, Address1_StateOrProvince, Address1_City, Address1_Line1, Address1_PostalCode, EMailAddress1, Name, NumberOfEmployees, PrimaryContactId_Name, Telephone1, WebSiteURL, CAST(ModifiedOn AS CHAR) as ModifiedOn from DynamicsCRM1.Account;
（日時型の項目のままにしてしまうと、内部でUnixタイムに変換されてBigQuery にインサートする際に失敗していまいますので、文字列にキャストしておきます）BigQuery の連携先テーブル：
dataflow:test.cdatatestMySQL JDBC ドライバー のパス：
gs://cdataconnect1/mysql-connector-java-8.0.21.jarGCS のディレクトリ：
gs://cdataconnect1/temp
gs://cdataconnect1/temp2
Dataflow の実行
入力が終わりましたら、実行ボタンをクリックしDataflow を実行します。
ジョブステータスが緑色で「完了しました」とメッセージが出たら正常終了となります。
それではBigQuery を確認してみます。
先ほど作成したテーブルの中にDynamics 365（CRM）のレコードが連携されていることが確認できました。
あとは定期実行するような場合は、Cloud Scheduler でスケジューリングなども行えるようですのでお試しください。
Dataflow 実行タイミングをBigQuery で保持
上記方法では、Dataflow を実行するごとにBigQueryにデータが蓄積されていきますが、完全に重複してしまうデータが存在してしまったりと、今後管理していくうえで悩ましいところにもなると思います。
そういった場合の一つの手段として、Dataflow に埋め込んだクエリにnow() を加えて、ジョブ実行時の日時をBigQuery に一緒に渡してあげる方法もあります。
select Id, Address1_Country, Address1_StateOrProvince, Address1_City, Address1_Line1, Address1_PostalCode, EMailAddress1, Name, NumberOfEmployees, PrimaryContactId_Name, Telephone1, WebSiteURL, CAST(ModifiedOn AS CHAR) as ModifiedOn, CAST(now() AS CHAR) as now from DynamicsCRM1.Account;
あとはBigQuery のテーブルにもtimestamp 型で項目を用意してあげれば、実行タイミングでレコードをグルーピングすることが可能になります。
あとは、必要に応じてBigQuery にスケジュールで動くSQL をセットしておけば、実行タイミングごとにテーブルを分割することも可能かと思います。
おわりに
いかがでしたでしょうか。プログラミングをしないとCloud Dataflow から外部のデータソースにアクセスできないイメージがあったのですが、JDBC テンプレートと公式のMySQL JDBC ドライバー、それとCData Connect でノーコードでBigQuery へ連携できるテンプレートが作成・実行できることが今回確認することができました。
Dynamics 365（CRM）以外のデータソースでも同じことができますので、是非CDataConnect をお試しいただければと思います。
