今回は CData Sync V26.1 で新たに追加された「パイプライン」機能を紹介します。パイプライン機能を使うと、複数のジョブを依存関係付きで連結し、一連のデータ処理を自動化できます。本記事では、基幹システム（SQL Server）とSaaS（kintone）のデータをSnowflakeに取り込み、その後データマートを自動生成するという一連の流れをパイプラインで実現する方法を解説します。本記事では検証にCData Sync 26.1.9516.0 を使用しています。
パイプライン機能とは？
CData Sync V26.1 で新たに追加されたパイプライン機能は、複数のジョブを順番に実行したり、あるジョブが完了したら次のジョブを実行したりといった、ジョブ間の依存関係を定義できる機能です。
従来のCData Sync でも、個々のデータソースからDWHへのレプリケーションは簡単に設定できました。しかし、「SQL Server のデータ取り込みが終わったら、kintone のデータを取り込み、両方が完了したらデータマート生成のSQLを実行する」といった一連の処理を自動化するには、スケジュールの実行時間を調整したりPost-Job イベント機能を駆使する必要がありました。
パイプライン機能を使えば、こうした一連の処理をGUI ベースで簡単に設定することができるようになりました。
今回構築するパイプライン
今回は以下のようなデータフローを実現するパイプラインを構築します。
データの内容
SQL Server（基幹システム）側
orders：受注データ（order_id, customer_code, product_code, quantity, amount, order_date）
customers：顧客マスタ（customer_code, customer_name, address, phone）
kintone 側
顧客管理：営業が管理する顧客情報（会社名, 顧客ランク, 業種, 都道府県, 電話番号）
案件管理：営業案件の進捗（会社名, 案件名, 提案商品, 商談フェーズ, 確度, 売上, 受注予定日）
統合後のデータマート
顧客ごとに「kintone 上の見込み金額の合計」と「SQL Server 上の実際の受注金額の合計」を比較できるビューを作成します。
設定手順
Step 1：接続設定
まず、SQL Server、kintone、Snowflake への接続を設定します。画面左側の「接続」タブ -> 「接続を追加」よりそれぞれのコネクタを選択し、接続情報を入力の上、接続を作成します。設定の詳細については各コネクタのヘルプページにある「接続の確立」セクションを参照ください。
Step 2：パイプライン作成 & SQL Server からのジョブを作成
画面左側の「Pipelines」タブを選択し、「Add Pipeline」からパイプラインを作成していきます。まずは任意のパイプライン名とデータ同期先(Snowflake) の接続を選択します。
パイプライン上で実行する連携ジョブを追加するため、「Add New Replication Job」を選択します。
最初にSQL Server から Snowflake へデータをレプリケーションするジョブを作成します。任意のジョブ名を入力し、Source に連携元の接続(SQL Server) を選択します。
次の画面で連携したいテーブル(customers, orders) を選択します。
連携先のスキーマを選択しSQL Server -> Snowflake のジョブの作成が完了です。
Step 3：kintoneからのジョブを作成
次にkintone -> Snowflake のジョブを作成していきます。SQL Server のジョブの次にkintone のデータを連携するようにします。現在作成されているパイプラインの画面から、線の上にマウスを置くと+アイコンが出てくるのでそれをクリックします。
SQL Serverの際と同様に、「Add New Replication Job」を選択してkintone をソースとして、ジョブ名・連携するテーブル(アプリ)・同期先スキーマを入力していきます。
Step 4：データマート作成の変換を作成
次にSnowflake 側に連携されたデータをもとに、顧客ごとの「kintone 上の見込み金額の合計」と「SQL Server 上の実際の受注金額の合計」を比較できるビューを作成していきます。この処理はSQL Server とkintone のデータ連携処理が完了したあとに実行します。kintone のジョブを作成した際と同様に、パイプラインの画面で線の上にマウスを置くと+アイコンをクリックて追加します。今回は追加時に「Add New Transformation」を選択して変換処理を作成していきます。
任意の変換名を入力し、Type はSQL を選択します。
ビュー作成のためのクエリをここで入力して、「Add」ボタンで追加します。
これで一連の処理が実行できるパイプラインの作成が完了しました。
上図一連のフローの中で処理がどこかでエラーになった場合は後続の処理が実行されません。もしエラーの場合でも後続の処理を進める場合は各処理ごとにある「Stop flow if error」を無効にすることで処理を進めることできます。また画面右上の通知アイコンをクリックすることで、エラー発生時やパイプライン終了時にメールで実行結果を通知することも可能です。
動作確認
パイプライン画面上部に「Run」ボタンをクリックし一連の処理を実行します。「Run」ボタンの脇にある「Add Schedule」からスケジュール実行の設定をすることも可能です。実行後、画面右上の「Run History」から各処理の実行結果などを確認することができます。
連携先のSnowflake を参照するとデータマートのテーブルが作成されていることを確認できます。このデータをTableau やLooker などのBI ツールに接続すれば、営業パイプラインと売上実績のダッシュボードを作成できます。
まとめ
今回は、CData Sync V26.1 で新たに追加されたパイプライン機能を使って、基幹システム（SQL Server）とSaaS（kintone）のデータをSnowflakeに統合し、データマートを自動生成する方法を紹介しました。パイプライン機能は、複数のデータソースを統合してデータ分析基盤を構築したい場合に非常に有効です。CData Sync V26.1 以降をお使いの方は、ぜひお試しください。CData Sync の30日間無償トライアルはこちらからダウンロードいただけます。
また、使っていて気になった点やご不明な点があれば、お気軽にテクニカルサポートまでお問い合わせください。
CData Sync V26.1 新機能「パイプライン」で基幹システムとSaaS のデータをDWH に統合する
