[IDMC]CDIR ApplicationでSalesforceからSnowflakeにデータをロードする(実装編)

前編(セッティング編)は以下のリンクから！

はじめに

様々な業務アプリケーションが存在する現在のデータ界隈で、ノーメンテナンスでリアルタイムなデータ取り込みを簡単に実現したい、という要望は多いと思います。
もちろん、Informaticaの提供するIDMC-CDIを用いれば、データの抽出・加工・格納（ETL）をノーコードで実装可能です。
しかし、より簡単にデータを取り込むためのサービスとして、InformaticaはIDMC-CDIR (Cloud Data Ingestion and Replication)を提供しています！
そこで今回は、IDMC-CDIRを用いたSalesforceからSnowflakeへのデータロードを実装していきます。

実装するのは以下の図の通り、SalesforceからSnowflakeに対して、直接データをIngestionする部分です。
(Snowflake内のデータ変換には、CDIのSQL ELTが最適です。)

Salesforceの項目追加を自動検知しSnowflakeに反映する機能(スキーマドリフト)についてもご紹介します！

CDIRとは

CDIR(Cloud Data Ingestion and Replication)は、InformaticaがIDMC(Intelligent Data Management Cloud)上で提供する、大量データロードのためのサービスです。
大規模かつ低レイテンシなデータの移動を、数ステップのウィザード形式で簡単に実現します。

CDIRについては以下の記事から。

今回は、CDIRの中でもアプリケーション取り込み(CDIR-Application)をご紹介します。

CDIR-Applicationとは

アプリケーション取り込みとレプリケーションは、Software-as-a-Service（SaaS）およびオンプレミスアプリケーションからクラウドベースのデータレイク、データウェアハウス、およびイベントストリーミングプラットフォームにデータを転送できます。
ビジネスまたは組織で使用されるSaaSとオンプレミスのアプリケーションは、ビジネスに不可欠な大量のデータを毎日保存しています。 アプリケーション取り込みとレプリケーションを使用すると、アプリケーションによって保存されたデータを、大量のデータを処理できるターゲットに転送できます。データをターゲットに転送した後、そのデータを統合して、高度なデータ分析やデータウェアハウジングなどのさまざまな目的に使用できます。

SaaSアプリケーションからのデータ転送に特化したサービスです。
Salesforce、ServiceNow、SAP、Micorsoft Dynamics365、Marketoなど、さまざまなソースをサポートしています。
特に、今回ご紹介したいのはスキーマドリフト機能です。

アプリケーション取り込みとレプリケーションは、スキーマドリフトとも呼ばれるソーススキーマの変更を自動的に検出し、ターゲットでこれらの変更を処理するように設定できます。

以下の図のように、ソースアプリケーションの項目変更(スキーマドリフト)を検知し、ターゲットテーブルのスキーマを自動で変更します。
これにより、データの取りこぼしを防いだり、変更対応にかかる工数を削減することが可能です！

実装

では実装していきましょう。
実装～実行までの流れは以下の図の通りです。

実装に必要なSalesforceの設定や接続定義の作成は、以下の記事で解説しています。

CDIRにアクセス

IDMCにログインし、「データ取り込みおよびレプリケーション」を開きます。

CDIRサービスはCDIに統合されているため、「データ統合」からも利用可能です。
(同じ画面に遷移します。)

取り込みタスクの作成

「取り込み」を選択します。

「アプリケーション取り込みおよびレプリケーション」を選択します。

ランタイム環境を指定します。

ロード先を指定します。
今回はSnowflakeにデータを取り込みます。

ソースとなるアプリケーションを指定します。
前の記事で作成したSalesforce接続定義を選択します。

タスクの詳細設定

ソースの詳細設定

• ソースのプロパティ
  • ロードタイプ：初期ロードと増分ロード
  • Salesforce API：Bulk API 2.0
• 詳細ソースプロパティ
  • CDC間隔：5分
• ソースオブジェクト
  • 取り込み対象のオブジェクトを選択します
    ※今回は「Opportunity(商談)」を取り込みます

設定内容は用途や検証内容に合わせてカスタムしてください。

ターゲットの詳細設定

• ターゲットプロパティ
  • ターゲット作成：ターゲットテーブルの作成
  • ステージ：Superpipeを利用しない場合に指定必須
• 詳細ターゲットプロパティ
  • 最終レプリケーション時刻を追加：オン
  • Superpipe：オフ
    ※当ORGではSuperpipeの指定ができなかったため、オフにしています。

設定内容は用途や検証内容に合わせてカスタムしてください。

トランスフォーメーションの設定

今回は割愛します。
簡単な文字列のトリムやフィルタの設定です。

開始の設定

タスク名や保存するプロジェクト/フォルダを指定します。

詳細オプション内では、先ほどご紹介したスキーマドリフトが設定できます。
今回はフィールドの追加・変更をターゲットテーブルに自動反映する設定をしています。

フィールドの追加や変更を『無視』に設定しておくと、ソースが予告なく変更された場合でも、下流ジョブへの影響を回避できます。

以上が、CDIRの設定の流れです。
簡単な設定のみでデータロードの実装ができることをご理解いただけたかと思います！

データ確認

タスクの実行前に、ソースとターゲットのデータを確認します。

データ準備

初期ロード用にSalesforceのOpportunity(商談)にデータを登録します。
2件の初期データを登録しました。

Snowflakeのスキーマ確認

Snowflakeの指定したスキーマにはテーブルが存在しません。

デプロイ

作成したCDIRタスクをデプロイします。
「モニタ」からデプロイされているジョブを確認することができます。

この時点で、Snowflakeのスキーマには、自動でロード先のテーブルが生成されています。

SnowflakeのQuery Historyからは、以下のCREATE文が実行されたことを確認できます。

CREATE  TABLE "*****Schema*****"."Opportunity_14"  ("Id" CHAR(72) NOT NULL , "IsDeleted" BOOLEAN  , "IsPrivate" BOOLEAN  , "Name" CHAR(480)  , "Description" CHAR(128000)  , "StageName" CHAR(1020)  , "Amount" NUMBER(18, 2)  , "Probability" INTEGER  , "ExpectedRevenue" NUMBER(18, 2)  , "TotalOpportunityQuantity" NUMBER(18, 2)  , "CloseDate" DATE  , "Type" CHAR(1020)  , "NextStep" CHAR(1020)  , "LeadSource" CHAR(1020)  , "IsClosed" BOOLEAN  , "IsWon" BOOLEAN  , "ForecastCategory" CHAR(160)  , "ForecastCategoryName" CHAR(1020)  , "HasOpportunityLineItem" BOOLEAN  , "OwnerId" CHAR(72)  , "CreatedDate" DATETIME(3)  , "CreatedById" CHAR(72)  , "LastModifiedDate" DATETIME(3)  , "LastModifiedById" CHAR(72)  , "SystemModstamp" DATETIME(3)  , "LastActivityDate" DATE  , "FiscalQuarter" NUMBER(10, 0)  , "FiscalYear" NUMBER(10, 0)  , "Fiscal" CHAR(24)  , "LastViewedDate" DATETIME(3)  , "LastReferencedDate" DATETIME(3)  , "HasOpenActivity" BOOLEAN  , "HasOverdueTask" BOOLEAN  , "LastAmountChangedHistoryId" CHAR(72)  , "LastCloseDateChangedHistoryId" CHAR(72)  , "DeliveryInstallationStatus__c" CHAR(1020)  , "TrackingNumber__c" CHAR(48)  , "OrderNumber__c" CHAR(32)  , "CurrentGenerators__c" CHAR(400)  , "MainCompetitors__c" CHAR(400)  , "INFA_LAST_REPLICATED" TIMESTAMP_NTZ(9)   DEFAULT CURRENT_TIMESTAMP, CONSTRAINT "PK_Opportunity_14_OpportunityPK" PRIMARY KEY ("Id"))

実行

では、タスクを実行します。
実行中のジョブからタスクを選択し「実行」します。

ステージの作成やデータのロード(COPY INTO)など、いくつかのSQLが実行されています。

create stage if not exists "*****Schema*****"."*****Stage*****" FILE_FORMAT=(type='csv' COMPRESSION=NONE)

モニタ上で、初期ロードの2件が追加されました。

無事、ターゲットテーブルにデータがロードされていました！

増分ロード＆スキーマドリフト

タスクを実行したまま、Salesforceにいくつかの変更を加えます。

商談データの追加と変更

TEST003を追加し、TEST002のフェーズを変更します。

データの追加と変更がテーブルに反映されました。

商談オブジェクトにカスタム項目を追加

オブジェクト設定から、「Custom001」を追加します。

「項目とリレーション」から「新規」を選択。

追加したカスタム項目。

スキーマ検出はソース上でのDML操作がトリガーとなるため、商談データ(TEST004)を追加します。

アプリケーション取り込みとレプリケーションがソースオブジェクトのスキーマ変更を検出するのは、変更されたソースオブジェクトでデータ操作言語（DML）操作が発生した後のみです。DML操作に干渉せずに複数のスキーマ変更が発生した場合、 アプリケーション取り込みとレプリケーションはDML操作が発生すると、スキーマのすべての変更をまとめて検出します。

カスタム項目がテーブルに自動追加されました！

おわりに

いかがでしたでしょうか？
簡単な実装やスキーマドリフトはCDIRならではの魅力なので、是非ご活用いただければ幸いです。

セッティング編から、長文記事にお付き合いいただきありがとうございました。
Informatica学習のための記事は今後も投稿していきますので、今後ともよろしくお願いします！

参考