Informaticaのクラウドを使ってSQL ServerからSnowflakeにリアルタイムで増分ロードしてみた

# はじめに
はじめまして。 [NTTデータ ソリューション事業本部 デジタルサクセスソリューション事業部](https://www.nttdata.com/jp/ja/services/data-and-intelligence/) の @nttd-nagano です。

[Informatica（インフォマティカ）](https://www.informatica.com/jp/) （※1）のクラウドデータマネジメントプラットフォームとして、「[Intelligent Data Management Cloud](https://www.informatica.com/jp/platform.html)」（※2。以下IDMCと記載）というものがあります。

**今回は、そのIDMCのデータ取り込みサービス「 [Cloud Data Ingestion and Replication](https://www.informatica.com/jp/products/cloud-integration/ingestion-at-scale.html) 」（略称「CDIR」、旧称「Mass Ingestion」）（※3）の機能「SuperPipe」を使って、SQL Serverからクラウドデータウェアハウス（※4）[Snowflake](https://www.snowflake.com/ja/) にリアルタイムで増分ロード（※6）してみた** ので、ご報告します。

:::note note
なお、記事執筆後に、当該記事で使用しているデータ取り込みサービスの名称が変更されました。本文は修正しましたが、スクリーンショットは旧名称でのものとなっております。ご了承ください。
* 旧名称：「Mass Ingestion」（「一括取り込み」）
* 新名称：「Data Ingestion and Replication」（略称CDIR。「データ取り込みおよびレプリケーション」）
:::

※. 余談ですが、 [弊社はSnowflakeも取り扱っており、「Snowflake Data Superheroes」（※7）が在籍しているほか、多数の受賞歴がございます。](https://www.nttdata.com/jp/ja/lineup/snowflake/#internal-link_news)

![idmc_summary.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/8195cf75-c1a2-67fa-790d-4de4d489f151.png)

<details><summary>※1. Informatica（インフォマティカ）</summary>

Informaticaは1993年に米国カリフォルニア州で設立され、2004年に日本法人を設立したデータマネジメント市場のパイオニア。Fortune100のうち86社を含む、100ヶ国以上の5000を超える企業がInformaticaのソリューションを採用しており、同社のクラウドは月間86兆件ものトランザクションを処理している。
</details>

<details><summary>※2. Intelligent Data Management Cloud</summary>

略称はIDMC。旧称はIICS。クラウドデータマネジメントプラットフォーム。以下IDMCと記載する。
</details>

<details><summary>※3. Cloud Data Ingestion and Replication（略称「CDIR」、旧称「Mass Ingestion」）</summary>

略称はCDIR。旧略称はCMI。データベースやSnowflakeなどのクラウドデータウェアハウス（※4）などへのデータ取り込みサービス。取り込み元は、データベースはもちろん、SAPやSalesforceなどのアプリケーション、ファイル、ストリーミングソースにも対応している。以下ではCDIRと記載する。
</details>

<details><summary>※4. クラウドデータウェアハウス</summary>

Snowflakeは当初同社のサービスを「クラウドデータウェアハウス」と位置づけていたが、その後、機能追加や適用範囲の拡大を踏まえて、現在は「データクラウド」と位置づけている。ここでは分かりやすさの観点から、当初の「クラウドデータウェアハウス」という表現を使用している。「クラウドデータウェアハウス」とは、クラウド版の「データウェアハウス」（※4）を指す。
</details>

<details><summary>※5. データウェアハウス</summary>

Data Warehouse、DWHとも表記される。「ウェアハウス」とは倉庫を表す英単語。「データウェアハウス」は、データ分析に必要な集計、結合、大量スキャン等の処理に特化したデータベースを指す。
</details>

<details><summary>※6. 増分ロード</summary>

ここでは「増分ロード」という言葉を、「差分ロード」という言葉と使い分けている。差分ロードの場合、毎回初回の全量連携から変更のあったレコードをロードする処理になる。一方、増分連携の場合、前回の連携から変更のあったレコードをロードする処理になる。
</details>

<details><summary>※7. Snowflake Data Superheroes</summary>

Snowflake Data Superheroは、Snowflakeが認定するSnowflakeのエキスパート集団。Snowflakeコミュニティーにおいて積極的に発信、情報交換を行い、Snowflake人材育成に貢献した功績が認められた人に与えられる。
</details>

# ビジネスのアジリティを高めるためにはリアルタイム性のあるデータ活用基盤が重要になってくる

昨今、データドリブン経営を目指し、 **日次あるいは数時間ごとに業務システムからデータ分析・活用基盤へデータを取り込んでいる例は多い** と思います。

こうした中で、 **ビジネスのアジリティをさらに高めていくためには、リアルタイム性のあるデータ活用基盤を実現することが重要になってくるのではないでしょうか。**

このようなデータ活用基盤を持つことで、様々な変化に迅速に対応し、顧客のニーズを先読みすることなどで、競争力を強化していくことができます。また、鮮度の落ちたデータによる誤った判断を防ぐこともできます。

* 例
    * 小売業におけるリアルタイムのマーケティングキャンペーン
    * 製造業における機器トラブルのリアルタイムの予兆検知
    * 金融業におけるリアルタイムの不正検出

# リアルタイム性のあるデータ活用基盤をSnowflakeを中心として実現する場合の要素技術：継続的ロードとCDC

さて、こうしたリアルタイム性のあるデータ活用基盤を、最近注目されている **クラウドデータウェアハウス [Snowflake](https://www.snowflake.com/ja/) を中心として実現する場合を考えてみましょう。**

一般的に、Snowflakeへデータをロードしたい場合、次のいずれかを使うことになります。

* **[COPYコマンドを使用したバルクロード](https://docs.snowflake.com/ja/user-guide/data-load-overview#bulk-loading-using-the-copy-command)**
* **[Snowpipeを使用した継続的ロード](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe)**
    * 以前からある仕組み。
    * ステージングファイルが追加されたことを検知してロードする。
    * クラウドストレージからの読み込みステップによりネットワーク的な遅延が数分加わる（高レイテンシ）。
    * ステージング領域によりストレージコストがかかる。
* **[Snowpipe Streamingを使用した継続的ロード](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe-streaming)**
    * 2023年7月に一般提供を開始した仕組み。
    * ステージングファイルを必要とせずに、データ行をテーブルに直接書き込む。
    * Snowpipeにはあったクラウドストレージからの読み込みステップがないため、遅延が短縮される（低レイテンシ）。
    * 高スループット（GB/秒）。
    * ストレージコストが不要で低コスト（取り込みの大小によらない）。

このうち「COPYコマンドを使用したバルクロード」は文字通りバルクでのロードを目的としていますので、リアルタイム性は期待できません。そのため、それ以外の「継続的ロード」系を検討することになります。

**「継続的ロード」を使う場合、取り込み元がデータベースである場合は、「Change Data Capture」（以下「CDC」と記載します）というソフトウェアデザインパターンと組み合わせると、その利点を活かせると思います。**

このデザインパターンの実現方法は様々ですが、たとえば、ほとんどのデータベース管理システムはすべての変更を記録する「トランザクションログ」や類似の仕組みを持っています。「CDC」の実現方法として、この **トランザクションログなどをスキャンし、データベースに加えられた変更を捉えるという方法があります。これが、ログベースのCDCです。** その他にも様々な手法があります。

* CDCの手法
    |手法|詳細|
    |---|---|
    |ログベース|データベースのすべての変更を記録するログ（トランザクションログなど）を読み取ることによりデータ変更をニアリアルタイムでキャプチャする。後述のInformaticaの実装では、Oracleの場合 [REDOログ](https://docs.oracle.com/cd/F82042_01/admin/managing-the-redo-log.html#GUID-79189E7E-7431-49CA-B342-3B971186503A) を読み込む、SQL Serverの場合は [トランザクションログ](https://learn.microsoft.com/ja-jp/sql/relational-databases/logs/the-transaction-log-sql-server?view=sql-server-ver16) と [CDCテーブル](https://learn.microsoft.com/ja-jp/sql/relational-databases/system-stored-procedures/sys-sp-cdc-enable-db-transact-sql?view=sql-server-ver16) を読み込む。|
    |CDCテーブル|[CDCテーブル](https://learn.microsoft.com/ja-jp/sql/relational-databases/system-stored-procedures/sys-sp-cdc-enable-db-transact-sql?view=sql-server-ver16) から直接データ変更をキャプチャする。|
    |クエリベース|CDCの対象カラムを指すWHERE句を使用してキャプチャする。|
    |トリガーベース|テーブルにトリガー（データベースでイベントが発生したときに自動的に実行される特殊なタイプのストアドプロシージャ）を仕掛け、データが変更されたときにトリガーを実行してキャプチャする。|

![idmc_cmi_superpipe_01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/49817c8d-dbb6-d0d5-b3d6-0c690d560cfa.png)

# リアルタイム性のあるデータ活用基盤をSnowflakeを中心として実現する方法を検討してみる

では、以上の要素技術（「CDC」および「継続的ロード」）に基づいたリアルタイム性のあるデータ活用基盤の実現方法を考えてみます。次の3点はどうでしょうか。

## AWS DMS + SnowpipeでSnowflakeにロードする

まず考えられるのは、CDCを利用できる [レプリケーションサービス AWS Database Migration Service（AWS DMS）](https://aws.amazon.com/jp/dms/) と [Snowpipe](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe) を使う方法です。

この方法を使うと、Snowflakeへの継続的な取り込みを実現できますが、継続的ロードの手段がSnowpipeであり、[Snowpipe Streaming](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe-streaming) ではないため、Snowpipe Streamingの場合より遅延（およびコスト）は大きくなります。

なお、本筋から逸れますが、取り込み元はデータベースに限られます。 [SAP](https://www.sap.com/japan/index.html) や [Salesforce](https://www.salesforce.com/jp/) などのアプリケーションのデータを取り込みたい場合は別の手段になります。

![idmc_cmi_superpipe_02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/7e833b5d-5bad-7e41-be72-bb82ae0a1b47.png)

## Debezium + Apache Kafka + Snowflake Sink Connector + SnowpipeでSnowflakeにロードする

次に考えられるのは、 [CDCのためのオープンソースプラットフォーム Debezium](https://debezium.io/) と [分散イベントストリーミングプラットフォーム Apache Kafka](https://kafka.apache.org/) と [Snowflake Sink Connector](https://docs.confluent.io/cloud/current/connectors/cc-snowflake-sink/cc-snowflake-sink.html) を通じて [Snowpipe](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe) を使う方法です。

こちらの方法も、Snowflakeへの継続的な取り込みを実現できますが、同様に継続的ロードの手段がSnowpipeであり、 [Snowpipe Streaming](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe-streaming) ではないため、Snowpipe Streamingの場合より遅延（およびコスト）は大きくなります。

また、Debeziumの環境とKafkaの環境を構築するための手間が必要になります（これは環境を自分たちの手でコントロールできることと表裏一体ですが）。

なお、再び本筋から逸れますが、こちらも取り込み元はデータベースに限られます。 [SAP](https://www.sap.com/japan/index.html) や [Salesforce](https://www.salesforce.com/jp/) などのアプリケーションのデータを取り込みたい場合は別の手段になります。

![idmc_cmi_superpipe_03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/b409bba4-3898-c1ab-88a9-5a09476da4be.png)

## Debezium + Apache Kafka + Snowflake Connector for Kafka + Snowpipe StreamingでSnowflakeにロードする

次に考えられるのは、 [Debezium](https://debezium.io/) と [Apache Kafka](https://kafka.apache.org/) と [Snowflake Connector for Kafka](https://docs.snowflake.com/ja/user-guide/data-load-overview#loading-data-from-apache-kafka-topics) を通じて [Snowpipe Streaming](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe-streaming) を使う方法です。

この方法は前述の2例とは異なり、継続的ロードの手段がSnowpipe Streamingであるため、低遅延（および低コスト）でロードできます。

ただし、同様にDebeziumの環境とKafkaの環境を構築するための手間が必要になります（これは環境を自分たちの手でコントロールできることと表裏一体ですが）。

なお、再び本筋から逸れますが、こちらも取り込み元はデータベースに限られます。 [SAP](https://www.sap.com/japan/index.html) や [Salesforce](https://www.salesforce.com/jp/) などのアプリケーションのデータを取り込みたい場合は別の手段になります。

![idmc_cmi_superpipe_04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/6c3b8c5a-75dc-2459-b41f-b6850672d9a6.png)

## ここまでのまとめ

ここまでの検討結果を整理すると、次のようになります。

 * [Snowpipe](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe) 系を使う場合、 [Snowpipe Streaming](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe-streaming) を使う場合より遅延が大きくなる。  
 リアルタイム性を重視するなら、できればSnowpipe Streamingを使いたい。また、そうすればコスト削減も期待できる。
 * [Debezium](https://debezium.io/) + [Apache Kafka](https://kafka.apache.org/) 系の場合、Snowpipeだけでなく、Snowpipe Streamingを使う環境を構築することもできる。  
 ただし、環境を構築するための手間が必要になる（これは環境を自分たちの手でコントロールできることと表裏一体）。
 * （本題ではないが）上記はすべて取り込み元がデータベースに限られる。  
 そのため、 [SAP](https://www.sap.com/japan/index.html) や [Salesforce](https://www.salesforce.com/jp/) などのアプリケーションのデータを取り込みたい場合は別の手段を用意する必要があり、データベースからの取り込みと統合的に運用できない。

これらの課題に対して、 **この記事では、Informaticaのクラウドの「 [CDIR](https://www.informatica.com/jp/products/cloud-integration/ingestion-at-scale.html) 」（※3）の「SuperPipe」をご紹介します。**

後述する通り、 **「SuperPipe」を使えば、GUIをポチポチ操作するだけで、CDCとSnowpipe Streamingを使ったリアルタイム性のある継続的ロードを実現できます。**

また、 **SAPやSalesforceなどのアプリケーションのデータの取り込みにも利用でき、さらにアプリケーションのデータの取り込みでも [Snowpipe Streamingでの継続的ロードに対応しています。](https://docs.informatica.com/ja_jp/integration-cloud/data-ingestion-and-replication/current-version/_-application-ingestion-and-replication_cloud-mass-ingestion_current-version_ditamap/GUID-FAA5CE48-0417-4355-8DB9-A049324D6349/GUID-7B0446E7-EB86-4B37-AE2E-86BDE1076978/GUID-2A27C797-2C99-41CC-A402-1672401AC47C/GUID-69DC5DD6-32DE-4B48-B249-DFC6BBC10A83.html)**

なお、言うまでもないことですが、この「SuperPipe」を含め、どんな条件・状況にも効くようなソリューションというのはありません。実際上は個別の条件・状況に合わせたソリューションを選択することになります。

# データ取り込みサービス「CDIR」とは

さて、Informaticaのクラウドサービス「IDMC」には、データ統合サービスとして「 [Cloud Data Integration](https://www.informatica.com/jp/products/cloud-integration/cloud-data-integration.html) 」（※8）がありますが、それとは別に、 **大量のデータを取り込むことに特化したデータ取り込みサービス「CDIR」（旧称「Mass Ingestion」）（※3）もあります。**

<details><summary>※8. Cloud Data Integration</summary>

略称はCDI。データ統合サービス。ETL処理（※9）やELT処理（※10）を担う。以下CDIと記載。
</details>

<details><summary>※9. ETL処理</summary>

データベースなどに蓄積されたデータから必要なものを抽出（Extract）し、目的に応じて変換（Transform）し、データを必要とするシステムに格納（Load）すること。
</details>

<details><summary>※10. ELT処理</summary>

ETL処理（※9）と対比して使われることが多い言葉。データ統合処理の順序を従来型のE→T→Lの順ではなく、E→L→Tの順でおこなう。近年ではDBMSの性能が爆発的に向上したことから、その性能を有効活用するために使われる手法｡
</details>

**この「CDIR」を使えば、大量のデータを各種データベースやSnowflakeなどに簡単に取り込むことができます。**

![idmc_cmi_superpipe_05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/fce4fde3-b222-a2a9-65a4-9324cabb2b1f.png)

CDIRの取り込み元は次の4種です（※11）。

<details><summary>※11. 取り込み元について</summary>
CDCに対応している取り込み元と、対応していない取り込み元があります。
</details>

* **データベース**
    * 機能名はDatabase Ingestion and Replication（旧称Mass Ingestion Databases）。
    * **[一部の取り込み元はCDCに対応](https://docs.informatica.com/ja_jp/integration-cloud/data-ingestion-and-replication/current-version/_database-ingestion-and-replication_cloud-mass-ingestion_current-version_ditamap/GUID-818FB11D-1A03-4EC9-9D22-E26A2679A744/GUID-7683E3AF-4969-4BEA-A7D7-812FDC15D6C7/GUID-824B4D3B-332A-4A6F-8670-AF79720797E5.html) 。**
    * 具体的な取り込み元（抜粋）
        * Oracle
        * PostgreSQL（Amazon RDS for PostgreSQL、Microsoft Azure Database for PostgreSQLを含む）
        * Microsoft SQL Server
        * Microsoft Azure SQL Database
* **SaaSアプリケーションおよびオンプレミスアプリケーション**
    * 機能名はApplication Ingestion and Replication（旧称Mass Ingestion Applications）。
    * 具体的な取り込み元（抜粋）
        * [Salesforce](https://www.salesforce.com/jp/products/what-is-salesforce/) 
※. 余談ですが [弊社はSalesforceも取り扱っており、多数の受賞歴がございます。](https://www.nttdata.com/jp/ja/lineup/salesforce/)
        * [Salesforce Marketing Cloud](https://www.salesforce.com/jp/products/marketing/)
        * [SAP S4/HANA](https://www.sap.com/japan/products/erp.html)
        * [ServiceNow](https://www.servicenow.com/jp/)
* **ファイル**
    * 機能名はFile Ingestion and Replication（旧称Mass Ingestion Files）。
    * 具体的な取り込み元（抜粋）
        * FTP/SFTP
        * Amazon S3
        * Microsoft Azure Blob Storage
* **ストリーミングソース**
    * 機能名はStreaming Ingestion and Replication（旧称Mass Ingestion Streaming）。
    * 具体的な取り込み元（抜粋）
        * Apache Kafka
        * Amazon Kinesis Data Streams
        * AMQP
        * MQTT

CDIRの取り込み先は次の通りです（※12）。

<details><summary>※12. 取り込み先について</summary>
後述のSuperPipeに対応している取り込み先はもちろんSnowflakeのみです。
</details>

* 共通
    * 具体的な取り込み先（抜粋）
        * Oracle（Database Ingestion and Replication、Application Ingestion and Replicationのみ対応）
        * Microsoft SQL Server（Database Ingestion and Replication、Application Ingestion and Replicationのみ対応）
        * [Snowflake](https://www.snowflake.com/ja/)  
※. 余談ですが [弊社はSnowflakeも取り扱っており多数の受賞歴がございます。](https://www.nttdata.com/jp/ja/lineup/snowflake/#internal-link_news)
        * [Databricks](https://www.databricks.com/jp)  
※. 余談ですが [弊社はDatabricksも取り扱っており多数の受賞歴がございます。](https://www.nttdata.com/jp/ja/lineup/databricks/)
        * Amazon Redshift
        * Microsoft Azure Synapse Analytics
        * Google BigQuery
        * Amazon S3
        * Microsoft Azure Data Lake Storage
        * Kafka（Streaming Ingestion and Replicationのみ対応）
        * Amazon Kinesis Data Firehose（Streaming Ingestion and Replicationのみ対応）
        * Microsoft Azure Event Hubs（Streaming Ingestion and Replicationのみ対応）

# 「CDIR」の「SuperPipe」とは

さて、昨年2023年12月、このうち、 **「Database Ingestion and Replication」と「Application Ingestion and Replication」が前述の [Snowpipe Streaming](https://docs.snowflake.com/ja/user-guide/data-load-overview#continuous-loading-using-snowpipe-streaming) に対応しました。**

Informaticaはこれに **「SuperPipe」という名称をつけてリリースしています。**

* [Informatica Announces General Availability of Superpipe for Snowflake with Up to 3.5x Faster Data Integration and Replication | Informatica](https://www.informatica.com/about-us/news/news-releases/2023/12/20231205-informatica-announces-general-availability-superpipe-for-snowflake-with-up-35x-faster-data-integration-replication.html)

**この「SuperPipe」を使えば、GUIをポチポチ操作するだけで、Snowpipe Streamingでのリアルタイム性のある継続的ロードを実現できます。**

![idmc_cmi_superpipe_06.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/944adf87-a698-9f2c-e53d-d531ae35930d.png)

# 参考：「SuperPipe」のための準備方法・実行方法のドキュメント

「SuperPipe」のための準備方法と実行方法は、次の公式ドキュメントに記載されています。

* 今回の取り込み元であるSQL Server接続（※13）の設定値の詳細は、 [公式ドキュメント「Microsoft SQL Serverコネクタ」](https://docs.informatica.com/ja_jp/integration-cloud/data-integration-connectors/current-version/_microsoft-sql-server-connector_cloud-data-integration-connectors_current-version_ditamap/GUID-6343E622-87E3-4E1A-ADF4-8D0D225A026B/GUID-753154E9-A486-4E04-BF48-79F4AB0BB0CF/GUID-06FAFA6B-64CB-4CD9-991F-A269A70FC283.html) をご覧ください。その他の取り込み元の設定方法も隣接部にあります。
* 取り込み先であるSnowflake接続（※13）の設定値の詳細は、 [公式ドキュメント「Snowflake Data Cloud Connector」](https://docs.informatica.com/ja_jp/integration-cloud/data-integration-connectors/current-version/_snowflake-data-cloud-connector_cloud-data-integration-connectors_current-version_ditamap/GUID-E8EDFBCD-9AF0-44AD-B773-18E7E9B2FFB2/GUID-0C745147-7A14-4E8A-B50B-8138E4549499/GUID-A6DF72AF-1E73-4EB9-A95B-DE04537A5C7E/GUID-169E5C40-0E86-4489-92E7-4392C07FD15B.html) をご覧ください。  
なお、SuperPipeを使う場合は「Authentication」欄の設定値を「Standard」ではなく、「KeyPair」にする必要があります。秘密鍵と公開鍵の生成方法は、 [公式ドキュメント「Snowflake Data Cloud Connector」](https://docs.informatica.com/ja_jp/integration-cloud/data-integration-connectors/current-version/_snowflake-data-cloud-connector_cloud-data-integration-connectors_current-version_ditamap/GUID-E8EDFBCD-9AF0-44AD-B773-18E7E9B2FFB2/GUID-0C745147-7A14-4E8A-B50B-8138E4549499/GUID-75F8ADB7-CDFB-4B14-BAB3-26919732A280/GUID-1AD88EFA-461C-43E8-A000-A79FFF0254DC.html) をご覧ください。
* データベース取り込みタスクの作成方法の詳細は、 [公式ドキュメント「Database Ingestion and Replication」](https://docs.informatica.com/ja_jp/integration-cloud/data-ingestion-and-replication/current-version/_database-ingestion-and-replication_cloud-mass-ingestion_current-version_ditamap/GUID-818FB11D-1A03-4EC9-9D22-E26A2679A744/GUID-AA3F9DF9-200C-4C0C-A7A2-2EFA72668F32/GUID-1F48EFA4-9032-4023-AD2F-AB1CA0C03C0D/GUID-38AD83A2-8164-4D54-BE11-DC1679597BFF.html) をご覧ください。

<details><summary>※13. 接続</summary>

認証情報などを管理するIDMC上の定義体。
</details>

# 実際に「SuperPipe」を使ってみる

では、実際に「SuperPipe」を使ってみます。今回の取り込み元はSQL Serverとします。

なお、ここではAmazon RDS for SQL Serverを使っていますが、オンプレミスのSQL Serverの場合も同一の手順で実現できます。

手順は次の通りです。

1. CDCが有効になっているかを確認する。
    ![idmc_cmi_superpipe_07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/1c124cce-5a76-3287-c32a-22cf6395542b.png)
1. SQL Serverにテーブルを作成する。
1. 作成したテーブルにデータを投入する。今回は下図のようなダミーデータを10000件投入しました。
    ![idmc_cmi_superpipe_07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/eeae8882-b206-4518-0565-883b145046d1.png)
1. IDMCにログインする。
1. 「マイサービス」ダイアログにて「管理者」（Administrator）をクリックする。
    ![idmc_cmi_superpipe_08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/f1674335-287f-f213-fb9c-a1cd652bd6d6.png)
1. サイドバーの「接続」をクリックする。
1. 「新しい接続」ボタンをクリックする。
1. 取り込み元として下図のようなSQL Server接続を作成する。  
    ![idmc_cmi_superpipe_09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/2dcbaefd-00a4-9760-0279-d66986264eb7.png)
1. 「新しい接続」ボタンをクリックする。
1. 取り込み先として下図のようなSnowflake接続を作成する。  
    ![idmc_cmi_superpipe_10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/c90590f3-40b2-0de9-0841-fa28cb6911d8.png)
1. 左上の「＞」部をクリックする。
1. 「マイサービス」ダイアログにて「データ取り込みおよびレプリケーション」（Cloud Data Ingestion and Replication）をクリックする。
    ![idmc_cmi_superpipe_11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/7fb2aab2-6030-e43f-9b95-c84b2b6e8bcf.png)
1. 「データベース取り込みタスク」をクリックする。
    ![idmc_cmi_superpipe_12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/e706e238-2562-9e0e-7034-48cfd2f55cdf.png)
1. 「名前」欄に任意の名称を入力する。（例：「dbmi_sqlserver_snowflake_cdc_streaming」）
1. 「ランタイム環境」欄にてSecure Agentをインストール済みの環境を選択する。
1. 「ロードタイプ」欄にて「初期ロードと増分ロード」を選択する。
    ![idmc_cmi_superpipe_13.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/97a43481-a619-156e-7c5f-6a6b8dd483cb.png)
1. 「次へ」ボタンをクリックする。
1. 「ソース」パートの「接続」欄にて、取り込み元であるSQL Server接続を選択する。
1. 「ソース」パートの「スキーマ」欄にて取り込み元とするスキーマを選択する。
1. 「データキャプチャメソッドの変更」パートの「CDCメソッド」欄にて「ログベース」を選択する。
1. 「テーブルの選択」パートの「ルール」パートに「テーブルルール」欄の設定値が「含む」で条件欄の設定値が「*」の行があることを確認する（必要に応じて変更する）。
1. 「テーブルの選択」パートの更新アイコンをクリックして、「影響を受けるテーブル」欄に対象テーブル数を表示させる。
    ![idmc_cmi_superpipe_14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/8f92ba27-7913-c552-21d7-4563dcea0839.png)
1. 「テーブルの選択」パートの「テーブルビュー」パートに対象テーブルが表示されていることを確認する。
1. 「テーブルの選択」パートの「テーブルビュー」パートの「CDCスクリプト」欄にて「すべてのカラムのCDCを有効化」が選択されていることを確認する（必要に応じて変更する）。
1. 「テーブルの選択」パートの「テーブルビュー」パートの「CDCスクリプト」欄の右の「実行」ボタンをクリックする。
    ![idmc_cmi_superpipe_15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/362aa5c9-945a-02c4-a249-ac2a1d8d9194.png)
1. 「CDCスクリプトは正常に完了しました。」と表示されることを確認する。
1. 「次へ」ボタンをクリックする。
1. 「ターゲット」パートの「接続」欄にて、取り込み先であるSnowflake接続を選択する。
1. 「ターゲット」パートの「スキーマ」欄にて取り込み先とするスキーマを選択する。
1. 「ターゲット」パートの「適用モード」欄にて「標準」を選択する（1回の適用サイクルの間の変更を累積し、それをターゲットに適用する前により少ないSQL文になるようにそれらをインテリジェントにマージするという設定値です）。
1. 「詳細」パートの「SuperPipe」欄を有効にする。
1. 「詳細」パートの「マージ頻度」欄に「300」と入力する。これは300秒＝5分を示す（Superpipeの場合、変更データ行がマージされターゲットテーブルに適用される頻度を秒単位で設定できます。有効な値は300～604800）。
    ![idmc_cmi_superpipe_16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/88e48f43-d2c1-fbc7-85dc-dc37f08edbd1.png)
1. 「次へ」ボタンをクリックする。
    ![idmc_cmi_superpipe_17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/127b2f0b-ecfd-2bcf-ee9b-fa73503a076f.png)
1. 「保存」ボタンをクリックする。
1. 「デプロイ」ボタンをクリックする。
1. サイドバーの「マイジョブ」をクリックする。
    ![idmc_cmi_superpipe_18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/86355f5b-2df3-2a77-2aa8-baf9bab22f7b.png)
1. デプロイしたデータ取り込みタスクを探し、右端の「…」をクリックし、ドロップダウンリストにて「実行」をクリックする。
    ![idmc_cmi_superpipe_19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/6cdb79f4-a08c-b468-2a9c-a6d95103dd4a.png)
1. デプロイしたデータ取り込みタスクをクリックする。
1. 「概要」パートにて、データ取り込みタスクの最初の実行の結果（読み取られたレコード数、書き込まれたレコード数などの情報）を確認する。
    ![idmc_cmi_superpipe_20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/708e17b7-73d8-69c0-7c3a-389b61e242e7.png)
1. 動作確認のため、取り込み元テーブルのレコードを更新してみる。
    ![idmc_cmi_superpipe_21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/c4da1d14-d4a0-f5eb-e325-8a6f56f6bdfa.png)
1. 「概要」パートにて、タスクの実行の結果を確認する（更新レコード数などの情報）。
    ![idmc_cmi_superpipe_22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/0c1b809e-f9cb-4531-54a1-e191b5d9152b.png)
1. 再度、動作確認のため、取り込み元テーブルのレコードを更新してみる。
    ![idmc_cmi_superpipe_23.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/79d8874f-44f6-dfcd-c13a-a830456cb8b0.png)
    ![idmc_cmi_superpipe_24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/6ce43db1-39fd-db29-c938-2298ca1229a2.png)
1. 「概要」パートにて、タスクの実行の結果を確認する（更新レコード数などの情報）。
    ![idmc_cmi_superpipe_25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/5d853987-5b73-9477-2e1a-6e054d683200.png)
1. 最後に、Snowpipeではなく、Snowpipe Streamingを使っていることを確認するために、Snowflakeにログインし、 `ACCOUNT_USAGE` スキーマの`SNOWPIPE_STREAMING_CLIENT_HISTORY` ビューをSELECTしてみる。
    ![idmc_cmi_superpipe_26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/d532bb41-d785-03c4-1301-5afbf87caf9a.png)

以上で実際に「SuperPipe」を使ってみることができました。

そして確かにSnowpipe Streamingを使ってロードされたようですね！

# 参考：CDIRのソース/ターゲットの対応バージョン・組み合わせ、CDCの可否などの詳細

* ソース/ターゲットの対応バージョン・組み合わせなど
    * [FAQ: What are the supported sources and targets for Cloud Data Ingestion and Replication (formerly Mass Ingestion) service?](https://knowledge.informatica.com/s/article/618516?language=en_US)
* 各データベースごとのCDCの可否、可能な場合そのメカニズムなど
    * 公式ドキュメント [データベース取り込みとレプリケーションソース - 準備と使用](https://docs.informatica.com/ja_jp/integration-cloud/data-ingestion-and-replication/current-version/_database-ingestion-and-replication_cloud-mass-ingestion_current-version_ditamap/GUID-818FB11D-1A03-4EC9-9D22-E26A2679A744/GUID-7683E3AF-4969-4BEA-A7D7-812FDC15D6C7.html)

# おわりに

以上、「Informaticaのクラウドを使ってSQL ServerからSnowflakeにリアルタイムで増分ロードしてみた」でした。

Snowflakeへのリアルタイム性のあるロードが、ほぼGUIだけで実現できるのは手軽ですね。
必要なテーブルのみを対象にできるのも利便性が高いと思います。

IDMCをご契約中の方はぜひ試していただければと思います。
未契約の方は弊社までお声掛けください。

**[IDMCのデータ統合サービス「CDI」は30日間の無料体験ができる](https://www.informatica.com/jp/trials/informatica-cloud.html)** ので、この機会に試してみてはいかがでしょうか。

## 仲間募集

NTTデータ ソリューション事業本部 では、以下の職種を募集しています。

<details><summary>1. クラウド技術を活用したデータ分析プラットフォームの開発・構築(ITアーキテクト/クラウドエンジニア)</summary>

クラウド／プラットフォーム技術の知見に基づき、DWH、BI、ETL領域におけるソリューション開発を推進します。
https://enterprise-aiiot.nttdata.com/recruitment/career_sp/cloud_engineer

</details>

<details><summary>2. データサイエンス領域（データサイエンティスト／データアナリスト）</summary>

データ活用／情報処理／AI／BI／統計学などの情報科学を活用し、よりデータサイエンスの観点から、データ分析プロジェクトのリーダーとしてお客様のDX／デジタルサクセスを推進します。
https://enterprise-aiiot.nttdata.com/recruitment/career_sp/datascientist

</details>

<details><summary>3.お客様のAI活用の成功を推進するAIサクセスマネージャー</summary>

DataRobotをはじめとしたAIソリューションやサービスを使って、
お客様のAIプロジェクトを成功させ、ビジネス価値を創出するための活動を実施し、
お客様内でのAI活用を拡大、NTTデータが提供するAIソリューションの利用継続を推進していただく人材を募集しています。
https://nttdata-career.jposting.net/u/job.phtml?job_code=804

</details>

<details><summary>4.DX／デジタルサクセスを推進するデータサイエンティスト《管理職/管理職候補》</summary>
データ分析プロジェクトのリーダとして、正確な課題の把握、適切な評価指標の設定、分析計画策定や適切な分析手法や技術の評価・選定といったデータ活用の具現化、高度化を行い分析結果の見える化・お客様の納得感醸成を行うことで、ビジネス成果・価値を出すアクションへとつなげることができるデータサイエンティスト人材を募集しています。

https://nttdata-career.jposting.net/u/job.phtml?job_code=898

</details>

## ソリューション紹介

<details><summary> Trusted Data Foundationについて</summary><div>

～データ資産を分析活用するための環境をオールインワンで提供するソリューション～
https://www.nttdata.com/jp/ja/lineup/tdf/
最新のクラウド技術を採用して弊社が独自に設計したリファレンスアーキテクチャ（Datalake+DWH+AI/BI）を顧客要件に合わせてカスタマイズして提供します。
可視化、機械学習、DeepLearningなどデータ資産を分析活用するための環境がオールインワンで用意されており、これまでとは別次元の量と質のデータを用いてアジリティ高くDX推進を実現できます。

</div></details>


<details><summary> TDFⓇ-AM（Trusted Data Foundation - Analytics Managed Service）について</summary><div>

～データ活用基盤の段階的な拡張支援（Quick Start) と保守運用のマネジメント（Analytics Managed）をご提供することでお客様のDXを成功に導く、データ活用プラットフォームサービス～
https://www.nttdata.com/jp/ja/lineup/tdf_am/
TDFⓇ-AMは、データ活用をQuickに始めることができ、データ活用の成熟度に応じて段階的に環境を拡張します。プラットフォームの保守運用はNTTデータが一括で実施し、お客様は成果創出に専念することが可能です。また、日々最新のテクノロジーをキャッチアップし、常に活用しやすい環境を提供します。なお、ご要望に応じて上流のコンサルティングフェーズからAI/BIなどのデータ活用支援に至るまで、End to Endで課題解決に向けて伴走することも可能です。

</div></details>

<details><summary> NTTデータとInformaticaについて</summary><div>

データ連携や処理方式を専門領域として10年以上取り組んできたプロ集団であるNTTデータは、データマネジメント領域でグローバルでの高い評価を得ているInformatica社とパートナーシップを結び、サービス強化を推進しています。
https://www.nttdata.com/jp/ja/lineup/informatica/

</div></details>

<details><summary>NTTデータとTableauについて </summary><div>

ビジュアル分析プラットフォームのTableauと2014年にパートナー契約を締結し、自社の経営ダッシュボード基盤への採用や独自のコンピテンシーセンターの設置などの取り組みを進めてきました。さらに2019年度にはSalesforceとワンストップでのサービスを提供開始するなど、積極的にビジネスを展開しています。

これまでPartner of the Year, Japanを4年連続で受賞しており、2021年にはアジア太平洋地域で最もビジネスに貢献したパートナーとして表彰されました。
また、2020年度からは、Tableauを活用したデータ活用促進のコンサルティングや導入サービスの他、AI活用やデータマネジメント整備など、お客さまの企業全体のデータ活用民主化を成功させるためのノウハウ・方法論を体系化した「デジタルサクセス」プログラムを提供開始しています。
https://www.nttdata.com/jp/ja/lineup/tableau/

</div></details>

<details><summary>NTTデータとAlteryxについて </summary><div>
Alteryxは、業務ユーザーからIT部門まで誰でも使えるセルフサービス分析プラットフォームです。

Alteryx導入の豊富な実績を持つNTTデータは、最高位にあたるAlteryx Premiumパートナーとしてお客さまをご支援します。

導入時のプロフェッショナル支援など独自メニューを整備し、特定の業種によらない多くのお客さまに、Alteryxを活用したサービスの強化・拡充を提供します。

https://www.nttdata.com/jp/ja/lineup/alteryx/

</div></details>

<details><summary>NTTデータとDataRobotについて </summary><div>
DataRobotは、包括的なAIライフサイクルプラットフォームです。

NTTデータはDataRobot社と戦略的資本業務提携を行い、経験豊富なデータサイエンティストがAI・データ活用を起点にお客様のビジネスにおける価値創出をご支援します。

https://www.nttdata.com/jp/ja/lineup/datarobot/

</div></details>

<details><summary>NTTデータとDatabricksについて </summary><div>
NTTデータでは、Databricks Inc.とソリューションパートナー契約を締結し、クラウド・データプラットフォーム「Databricks」の導入・構築、および活用支援を開始しています。

NTTデータではこれまでも、独自ノウハウに基づき、ビッグデータ・AIなど領域に係る市場競争力のあるさまざまなソリューションパートナーとともにエコシステムを形成し、お客さまのビジネス変革を導いてきました。
Databricksは、これら先端テクノロジーとのエコシステムの形成に強みがあり、NTTデータはこれらを組み合わせることでお客さまに最適なインテグレーションをご提供いたします。

https://www.nttdata.com/jp/ja/lineup/databricks/

</div></details>