はじめに
業務システムやSaaSのデータをデータ基盤へ継続的に反映する場合、CDC（Change Data Capture）を利用することで、ソースで発生した変更を検知し、ターゲットへ連携できます。
しかし、実際のデータパイプラインは、データを転送するだけでは完結しません。取り込んだデータに対して、クレンジング、変換、集計などの後続処理が必要になることも多くあります。
このとき、CDC処理と後続処理が別々のツールで管理されていると、ジョブの依存関係や実行タイミングが複雑になります。
InformaticaのIDMC（Intelligent Data Management Cloud）では、CDIR（Cloud Data Ingestion and Replication）による変更データの取り込みを起点として、CDI（Cloud Data Integration）のTaskflowを自動実行できるため、リアルタイム性の高いデータパイプラインを簡単に実現できます。
本記事では、CDIRのCDCサイクル完了後にTaskflowを起動し、そのサイクルで取り込まれた差分データだけを処理する方法をご紹介します。
概要
CDIRとは
CDIRは、InformaticaがIDMC上で提供する、大量データロードのためのサービスです。
大規模かつ低レイテンシなデータの移動を、数ステップのウィザード形式で簡単に実現します。
以前の記事では、SalesforceからSnowflakeへのデータロードや、Snowflake管理のIcebergテーブルへのデータロードを実装しました。
本記事では、CDIRによるデータ取り込み完了後の後続処理に焦点を当てます。
CDIRをトリガーにTaskflowを起動する
CDCによって変更データを継続的に取り込む場合、取り込み後にクレンジングや変換、集計などの後続処理が必要になることがあります。
このとき課題となるのが、後続処理を開始するタイミングの制御と、そのサイクルで取り込まれたデータの識別です。
IDMCでは、CDIRによるデータの取り込み完了をトリガーとしてTaskflowを起動できます。その際、CDIRから渡されたCycle IDをMapping Taskのフィルタ条件として使用することで、対象のCDCサイクルで取り込まれた差分データだけを処理できます。
CDIRを単なるデータ取り込み機能ではなく、後続処理を開始する起点として活用できることが大きな特徴です。
実装
以下のような構成を実装します。
CDIRでSalesforceからSnowflakeのBronzeレイヤにデータを取り込んだ後、CDIのTaskflowが自動起動し、Silverレイヤへデータを連携します。
CDIR-Applicationの作成
まず、CDIR-Applicationを作成します。
今回は、Taskflow起動に必要な設定に絞ってご紹介します。
CDIRの全般的な設定については、以下の記事をご参照ください。
「サイクルIDの追加」
今回の構成で重要になるのが、Cycle IDです。
Cycle IDは、CDIRの各取り込みサイクルを識別するための値です。
後続処理では、このCycle IDを条件に使用することで、対象の取り込みサイクルで連携されたデータだけを処理できます。
「タスクの詳細 - ターゲットの詳細」ステップで「サイクルIDの追加」を設定します。
「タスクフローで実行」
「開始」ステップにて「タスクフローで実行」オプションを有効にします。
この設定を有効にすると、CDIRの取り込みサイクル完了時にイベントが発行され、指定したTaskflowを起動できるようになります。
デプロイ
設定を保存し、「デプロイ」します。
タスクをデプロイすると、設定内容に基づいてSnowflake上にターゲットテーブルが作成されます。
Mappingの作成
CDIRで取り込んだデータを処理するMappingを作成します。
入出力パラメータの作成
CDIRから渡されるCycle IDを受け取るため、Mappingに入出力パラメータを作成します。
フィルタトランスフォーメーションの作成
ソーステーブルに格納されたCycle IDと、Taskflow経由で受け取ったCycle IDが一致する行だけを抽出するように、フィルタトランスフォーメーションを設定します。
Mappingの保存
ターゲットとしてSilverレイヤのテーブルを指定し、Mappingを保存します。
※企業名と電話番号を整形する簡単なクレンジングを追加しています。
Mapping Taskを作成します。
Taskflowの作成
次に、Taskflowを作成します。
CDIR Eventの設定
Taskflowの開始プロパティで、CDIRタスクを指定します。
データタスク内でパラメータを受け渡す
データタスクを追加し、先ほど作成したMapping Taskを指定します。
CDIRイベントから受け取ったCycle IDを、Mapping Taskの入力フィールドへ割り当てます。
保存し、パブリッシュします。
DEMO
では、実際に動かしてみましょう。
事前にSalesforceには3件の初期データを登録しています。
先ほどデプロイしたCDIRタスクを実行します。
初期データの取り込み
CDIRタスクが起動し、SalesforceからSnowflakeへの初期ロードが開始されます。
Bronzeレイヤにデータが初期ロードされます。
初期ロードのサイクルが完了すると、CDIRイベントを受け取ったTaskflowが起動し、Mapping Taskが実行されます。
BronzeレイヤからSilverレイヤにデータが連携されました。
変更データの取り込み
Salesforceに新たなAccount（取引先）を追加します。
CDIRによってデータがロードされ、その後Taskflowが起動します。
Bronzeレイヤにデータが増分ロードされます。
増分ロードで追加されたデータには、初期ロード時とは異なるCycle IDが付与されています。
CDIRをトリガーにTaskflowが起動し、BronzeレイヤからSilverレイヤにデータが連携されました。
Taskflowには新しいCycle IDが渡されるため、今回追加されたデータだけが処理されます。
以降も、CDCサイクルごとに新しいCycle IDが発行されます。後続処理では、イベントから受け取ったCycle IDを使って、各サイクルの対象データを識別できます。
おわりに
本記事では、CDIRのCDCサイクル完了後にTaskflowを起動し、そのサイクルで取り込まれた差分データだけを処理する方法をご紹介しました。
CDIRを単なるデータ転送機能ではなく、後続処理を含むデータパイプライン全体の起点として活用できる点が、データマネジメントのためのオールインワンプラットフォームであるIDMCの大きな強みです。
データの取り込みから加工までを自動化し、シンプルで運用しやすいデータ基盤の構築を検討されている方は、ぜひCDIRを起点とした構成を試してみてください。
参考