Informaticaのデータ統合サービスCloud Data IntegrationでSnowflakeにELTしてみた

# はじめに
はじめまして。 [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 Integration」（※3。以下CDIと記載）の機能「プッシュダウン最適化」機能（Pushdown Optimization）にて、クラウドデータウェアハウス（※5） [Snowflake](https://www.snowflake.com/ja/) （※7）にELT（※8）してみた** ので、ご報告します。

以前Qiitaに投稿した記事（「 [Informaticaのデータ統合サービスCloud Data IntegrationでRedshiftにELTしてみた](https://qiita.com/nttd-nagano/items/7a935219d3be1bea67e6) 」）のSnowflake版です。

![idmc_summary.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/51687b25-6068-c6ee-8fe9-841924650391.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 Integration</summary>
略称はCDI。データ統合サービス。ETL処理（※3）やELT処理（※6）を担う。以下CDIと記載。
</details>

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

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

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

<details><summary>※7. Snowflake</summary>
Snowflakeはクラウドデータウェアハウス（※5）。Snowflakeは、クラウドネイティブなアーキテクチャに基づき高い柔軟性・弾力性・拡張性を持ちながらも、標準SQLにも準拠しているため非常に使いやすく、広範なワークロードを容易かつセキュアに実行できる。これらの機能によって、1か所にまとめられたあらゆるデータに、いつでも、どこでも、何人でもセキュアかつ容易にアクセスできるようにすることができるデータ分析環境を容易に立ち上げられる。
</details>

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

データ統合処理は、データの変換を伴うことが多いですが、 この **「プッシュダウン最適化」機能では、ELT処理によりDBMSの処理能力を最大限に活用し、高いパフォーマンスを実現します。**

具体的には、 **データ変換処理を各DBMS固有の命令を含む最適なSQLクエリに変換して、DBMS側で実行します。**

今回は、 [Snowflake](https://www.snowflake.com/ja/) を対象に、この「プッシュダウン最適化」機能の使い方をご紹介します。

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

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

# S3バケットを作って入力側となるファイルを格納する

CDIは [様々なシチュエーションでデータ統合処理をSnowflakeにプッシュダウンすることができます](https://docs.informatica.com/integration-cloud/data-integration-connectors/current-version/snowflake-data-cloud-connector/part-3--sql-elt-with-snowflake-data-cloud-connector.html) が、今回、Snowflakeにロードするデータは、S3にCSVファイルとして格納しておくことにしました。

まず、AWSマネジメントコンソールの「S3」にて、S3バケットを作りました。

次に、入力ファイルとして、次のような3種類のダミーデータを用意し、格納しました。

* order_data.csv （注文情報）
* customer_data.csv （顧客マスタ）
* product_data.csv　（商品マスタ）

![informatica_cdi_pushdown_snowflake_001.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/6fd7f33c-1be3-3b9a-785d-deffd52c346f.png)
![informatica_cdi_pushdown_snowflake_002.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/7b1c8b43-bd23-8e4a-120c-bd05ab417674.png)
![informatica_cdi_pushdown_snowflake_003.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/1e432f3a-add9-0a3b-f036-141ae4ef4100.png)

# Snowflakeに出力側となるテーブルを作る

今回、Snowflake側にロードする際には、前述の入力を変換しながら結合していくこととします。（詳細は後述）

SnowflakeのWebインターフェース「 [Snowsight](https://docs.snowflake.com/ja/user-guide/ui-snowsight) 」にて、次のようなテーブルを作りました。

![informatica_cdi_pushdown_snowflake_004.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/dc77b35b-c40b-0c09-179d-ee9c51e78f81.png)

# IDMCにログインし、アドオンコネクタとしてS3とSnowflakeを有効にする

CDIでは接続先のリソース毎に、「コネクタ」を使用します。標準のコネクタのほかに、 [様々なアドオンコネクタがあります。](https://docs.informatica.com/integration-cloud/data-integration-connectors.html)

[CDIの30日間の無料体験](https://www.informatica.com/jp/trials/informatica-cloud.html) を使用している場合などは、アドオンコネクタを使えるようにする作業が必要です。下記の操作を実施してください。すでに使えるようになっている場合はこの作業は不要です。

1. IDMCにログインする。
1. アプリケーションピッカー（マイサービス）で「管理者」をクリックする。
1. 左ペインで「アドオンコネクタ」をクリックする。
1. 右上の検索欄に「s3」と入力する。
1. 「Amazon S3 Connector」の「無料トライアルを開始する」ボタンをクリックする。
1. ダイアログで「OK」ボタンをクリックする。
1. 右上の検索欄に「snowflake」と入力する。
1. 「Informatica for Snowflake Data Cloud」の「無料トライアルを開始する」ボタンをクリックする。
1. ダイアログで「OK」ボタンをクリックする。

# 接続オブジェクトを作る

CDIでは接続先の設定を「[接続](https://docs.informatica.com/cloud-common-services/administrator/current-version/connections.html)」（あるいは「接続オブジェクト」）という定義体で管理しています。

今回はS3とSnowflakeの接続オブジェクトを作ります。

なお、CDIにて、プッシュダウンをしない場合（ETL処理をする場合）は、下表の「Access Key」「Secret Key」を空にしておき、IAMロール（S3バケットからCSVファイルを読み込む権限を付与したIAMロール）をSecure AgentをインストールしたEC2インスタンスに割り当てておくことも可能ですが、 **プッシュダウンをする場合（ELT処理をする場合）は、SnowflakeがS3に直接接続しにいく都合上、「Access Key」「Secret Key」が必須となります。**

1. 左ペインで「接続」をクリックする。
1. 「新しい接続」ボタンをクリックする。
    ![informatica_cdi_pushdown_snowflake_005.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/23b2c8cd-9d64-f652-b839-f857e9252cec.png)
1. [S3用の接続オブジェクト](https://docs.informatica.com/integration-cloud/data-integration-connectors/current-version/amazon-s3-v2-connector.html) を作るために、下記のように入力する。
    |設定項目|設定値|
    |---|---|
    |接続名|【任意。以降では「conn_amazons3」と記載）】|
    |説明||
    |タイプ|Amazon S3 v2|
    |ランタイム環境|【自分のランタイム環境（Secure Agentをインストールしたマシン）】|
    |Access Key|【あらかじめAWSマネジメントコンソールの「IAM」にて払い出しておく】|
    |Secret Key|【あらかじめAWSマネジメントコンソールの「IAM」にて払い出しておく】|
    |IAM Role ARN||
    |External Id||
    |Use EC2 Role to Assume Role|
    |Folder Path|【先ほど作成したS3バケット名】|
    |Master Symmetric Key||
    |Customer Master Key ID||
    |S3 Account Type|Amazon S3 Storage|
    |Region Name|【S3バケットが存在するリージョン名】|
    |Federated SSO IdP|NONE|
    |Other Authentication Type |NONE|
    |S3 VPC Endpoint Type|NONE|
    |STS VPC Endpoint Type|NONE|
    |KMS VPC Endpoint Type|NONE|
1. 「保存」ボタンをクリックする。
1. 「テスト接続」ボタンをクリックする。
1. 「この接続のテストに成功しました。」と表示されることを確認する。
1. 「新しい接続」ボタンをクリックする。
1. [Snowflake用の接続オブジェクト](https://docs.informatica.com/integration-cloud/data-integration-connectors/current-version/snowflake-data-cloud-connector.html) を作るために、下記のように入力する。
    |設定項目|設定値|
    |---|---|
    |接続名|【任意。以降では「conn_snowflake」と記載）】|
    |説明||
    |タイプ|Snowflake Data Cloud|
    |ランタイム環境|【自分のランタイム環境（Secure Agentをインストールしたマシン）】|
    |Authentication|Standard|
    |Username|【ユーザー名】|
    |Password|【パスワード】|
    |Account|【 [アカウント識別子](https://docs.snowflake.com/ja/user-guide/admin-account-identifier) （左記リンク先の「形式2（レガシー）：リージョン内のアカウントロケーター」を使う必要がある） 】|
    |Warehouse|【 [仮想ウェアハウス](https://docs.snowflake.com/ja/user-guide/warehouses) 】|
    |Role||
    |Additional JDBC URL Parameters||
1. 「保存」ボタンをクリックする。
1. 「テスト接続」ボタンをクリックする。
1. 「この接続のテストに成功しました。」と表示されることを確認する。

# マッピングを作る

CDIでは、データ統合処理を「[マッピング](https://docs.informatica.com/integration-cloud/data-integration/current-version/mappings.html)」という定義体で管理しています。

まず、空のマッピングを作ります。

1. アプリケーションピッカー（マイサービス）で「データ統合」をクリックする。
1. 左ペインにて「新規」をクリックする。
1. ダイアログで、「マッピング」＞「マッピング」をクリックする。
    ![informatica_cdi_pushdown_snowflake_006.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/3e3125cb-1581-f171-ddad-01147cf80a98.png)
1. 「作成」ボタンをクリックする。

さて、マッピングでのデータ統合処理の内容ですが、今回は、「S3バケット内の複数のCSVファイルのデータを抽出し、一部を変換して、結合し、さらに変換した後に、Snowflakeのテーブルにロードする」というものにしてみます。

詳細は割愛しますが、おおむね次のような手順で作成しました。

データ変換はあくまで一例なので、その内容よりも、プッシュダウン先であるターゲットオブジェクトの設定に注目していただければと思います。

1. [ソースオブジェクト](https://docs.informatica.com/integration-cloud/data-integration/current-version/transformations/source-transformation.html) が3個になるように配置する。
1. 各ソースオブジェクトを選択し、「ソース」パートに以下のように設定する。
    |設定項目|設定値|
    |---|---|
    |接続|【先ほど作成したS3用接続オブジェクト】|
    |ソースタイプ|単一オブジェクト|
    |オブジェクト|【各CSVファイル】|
    |形式|Flat|

    ![informatica_cdi_pushdown_snowflake_007.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/8c4709b0-f500-425a-f678-048dd51a334e.png)
    ![informatica_cdi_pushdown_snowflake_008.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/ae3a698e-5f53-d35d-3f66-07fcade1495c.png)
    ![informatica_cdi_pushdown_snowflake_009.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/cb9cbece-063c-ad77-bef6-27e9537bb57e.png)
1. [フィルタトランスフォーメーション](https://docs.informatica.com/integration-cloud/data-integration/current-version/transformations/filter-transformation.html) を配置し、customerと接続する。18歳以上を通過させるように設定する。
    ![informatica_cdi_pushdown_snowflake_010.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/4f2ec737-d429-0b66-263b-11cd64debeeb.png)
1. [ジョイナトランスフォーメーション](https://docs.informatica.com/integration-cloud/data-integration/current-version/transformations/joiner-transformation.html) を配置し、orderとcustomerを接続する。
    ![informatica_cdi_pushdown_snowflake_011.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/0969109f-973a-0b13-4469-f6ced9f33aa7.png)
1. [ジョイナトランスフォーメーション](https://docs.informatica.com/integration-cloud/data-integration/current-version/transformations/joiner-transformation.html) を配置し、前項のジョイナトランスフォーメーションとproductを接続する。
    ![informatica_cdi_pushdown_snowflake_012.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/5443bc88-cf45-3996-95c2-cce3136f4048.png)
1. [式トランスフォーメーション](https://docs.informatica.com/integration-cloud/data-integration/current-version/transformations/expression-transformation.html) を配置し、前項のジョイナトランスフォーメーションを接続する。
1. 式トランスフォーメーションにて、「LASTNAME」と「FIRSTNAME」を [文字列結合](https://docs.informatica.com/integration-cloud/data-integration/current-version/function-reference/functions/concat.html) するように設定する。
    ![informatica_cdi_pushdown_snowflake_013.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/187b8fe6-abbc-4e45-ade5-c4cd0874ad24.png)
1. [ターゲットオブジェクト](https://docs.informatica.com/integration-cloud/data-integration/current-version/transformations/target-transformation.html) を選択し、「ターゲット」パートに以下のように設定する。
    |設定項目|設定値|
    |---|---|
    |接続|【先ほど作成したSnowflake用接続オブジェクト】|
    |ターゲットの種類|単一オブジェクト|
    |オブジェクト|【先ほど作成したテーブル】|
    |操作|Insert|
    |UpdateMode|【デフォルト値のまま】|
    |Database||
    |Schema||
    |Warehouse||
    |Role||
    |Pre SQL||
    |Post Sql||
    |Batch Row Size||
    |Number of local staging files||
    |Truncate Target Table|【デフォルト値のまま】|
    |Additional Write Runtime Parameters||
    |Table Name||
    |Rejected File Path||
    |Update Override||
    |拒否された行の転送|【デフォルト値のまま】|

    ![informatica_cdi_pushdown_snowflake_014.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/bd4dc7d9-0575-9aab-0c7a-2749d6925b87.png)
1. ターゲットオブジェクトに前述の式トランスフォーメーションを接続する。
1. ターゲットオブジェクトの「フィールドマッピング」パートにて、受信フィールドとターゲットフィールドをマッピングする。


# プッシュダウン最適化結果をプレビューする

[プッシュダウン最適化機能によってマッピングがどのようにSQLクエリに変換されるかをプレビューしてみましょう。](https://docs.informatica.com/integration-cloud/data-integration/current-version/mappings/mappings/sql-elt-query-preview.html)

次のように操作します。

1. 画面右上の「プッシュダウン最適化」アイコンをクリックする。
1. 「プッシュダウンのプレビュー」ボタンをクリックする。
    ![informatica_cdi_pushdown_snowflake_015.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/64d8e56c-0e62-4cdc-3d8d-aed92335cfde.png)
1. 「次へ」ボタンをクリックする。
1. 「プッシュダウンプレビュー」ボタンをクリックする。
1. 生成されたSQLクエリが表示されている箇所の右上の「⇔」アイコンをクリックする。
    ![informatica_cdi_pushdown_snowflake_016.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/2be4d933-3e59-2aac-4e9c-88ac28ea7a49.png)
1. プレビューを見終わったら、「閉じる」ボタンをクリックする。
    ![informatica_cdi_pushdown_snowflake_017.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/4836d2da-0443-351e-2733-4a006f3c80a8.png)

下記がプッシュダウン最適化機能によって生成されたSQLクエリです。

<details>
<summary>
生成されたSQLクエリ1
</summary>
<pre>
/** SQLクエリ1 **/
CREATE
OR REPLACE TEMPORARY STAGE "DEMO_DB"."DEMO"."STAGE_ADULT_ORDER_1695197685139_7f5dca55-e292-4a16-a0da-5da8ab230266" URL = * * * * * * * * * * CREDENTIALS = * * * * * * * * * *;
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ2
</summary>
<pre>
/** SQLクエリ2 **/
CREATE TEMP TABLE "DEMO_DB"."DEMO"."ADULT_ORDER_1695197685139_7f5dca55-e292-4a16-a0da-5da8ab230266"(
  "PRODUCT_ID" VARCHAR(256),
  "PRODUCT_NAME" VARCHAR(256)
);
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ3
</summary>
<pre>
/** SQLクエリ3 **/
COPY INTO "DEMO_DB"."DEMO"."ADULT_ORDER_1695197685139_7f5dca55-e292-4a16-a0da-5da8ab230266" ("PRODUCT_ID", "PRODUCT_NAME")
FROM
  (
    SELECT
      $ 1,
      $ 2
    FROM
      @"DEMO_DB"."DEMO"."STAGE_ADULT_ORDER_1695197685139_7f5dca55-e292-4a16-a0da-5da8ab230266"
  ) FILE_FORMAT = (
    FIELD_DELIMITER = ',' FIELD_OPTIONALLY_ENCLOSED_BY = '0x0022' ESCAPE = '0x005c' ENCODING = 'UTF-8' SKIP_HEADER = 1
  );
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ4
</summary>
<pre>
/** SQLクエリ4 **/
CREATE
OR REPLACE TEMPORARY STAGE "DEMO_DB"."DEMO"."STAGE_ADULT_ORDER_1695197685111_2ed7ce33-20c3-45a6-b523-7778065aede0" URL = * * * * * * * * * * CREDENTIALS = * * * * * * * * * *;
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ5
</summary>
<pre>
/** SQLクエリ5 **/
CREATE TEMP TABLE "DEMO_DB"."DEMO"."ADULT_ORDER_1695197685111_2ed7ce33-20c3-45a6-b523-7778065aede0"(
  "CUSTOMER_ID" VARCHAR(256),
  "LASTNAME" VARCHAR(256),
  "FIRSTNAME" VARCHAR(256),
  "LASTNAME_KANA" VARCHAR(256),
  "FIRSTNAME_KANA" VARCHAR(256),
  "AGE" VARCHAR(256)
);
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ6
</summary>
<pre>
/** SQLクエリ6 **/
COPY INTO "DEMO_DB"."DEMO"."ADULT_ORDER_1695197685111_2ed7ce33-20c3-45a6-b523-7778065aede0" (
  "CUSTOMER_ID",
  "LASTNAME",
  "FIRSTNAME",
  "LASTNAME_KANA",
  "FIRSTNAME_KANA",
  "AGE"
)
OM
(
  SELECT
    $ 1,
    $ 2,
    $ 3,
    $ 4,
    $ 5,
    $ 6
  FROM
    @"DEMO_DB"."DEMO"."STAGE_ADULT_ORDER_1695197685111_2ed7ce33-20c3-45a6-b523-7778065aede0"
) FILE_FORMAT = (
  FIELD_DELIMITER = ',' FIELD_OPTIONALLY_ENCLOSED_BY = '0x0022' ESCAPE = '0x005c' ENCODING = 'UTF-8' SKIP_HEADER = 1
);
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ7
</summary>
<pre>
/** SQLクエリ7 **/
CREATE
OR REPLACE TEMPORARY STAGE "DEMO_DB"."DEMO"."STAGE_ADULT_ORDER_1695197684899_0d5fbdc0-5084-442c-a561-bbed1572ca25" URL = * * * * * * * * * * CREDENTIALS = * * * * * * * * * *;
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ8
</summary>
<pre>
/** SQLクエリ8 **/
CREATE TEMP TABLE "DEMO_DB"."DEMO"."ADULT_ORDER_1695197684899_0d5fbdc0-5084-442c-a561-bbed1572ca25"(
  "ORDER_ID" VARCHAR(256),
  "CUSTOMER_ID" VARCHAR(256),
  "PRODUCT_ID" VARCHAR(256)
);
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ9
</summary>
<pre>
/** SQLクエリ9 **/
COPY INTO "DEMO_DB"."DEMO"."ADULT_ORDER_1695197684899_0d5fbdc0-5084-442c-a561-bbed1572ca25" ("ORDER_ID", "CUSTOMER_ID", "PRODUCT_ID")
FROM
  (
    SELECT
      $ 1,
      $ 3,
      $ 4
    FROM
      @"DEMO_DB"."DEMO"."STAGE_ADULT_ORDER_1695197684899_0d5fbdc0-5084-442c-a561-bbed1572ca25"
  ) FILE_FORMAT = (
    FIELD_DELIMITER = ',' FIELD_OPTIONALLY_ENCLOSED_BY = '0x0022' ESCAPE = '0x005c' ENCODING = 'UTF-8' SKIP_HEADER = 1
  );
</pre>
</details>
<details>
<summary>
生成されたSQLクエリ10
</summary>
<pre>
/** SQLクエリ10 **/
INSERT INTO
  "DEMO_DB"."DEMO"."ADULT_ORDER"(
    "ORDER_ID",
    "CUSTOMER_FULLNAME",
    "CUSTOMER_AGE",
    "PRODUCT_NAME"
  )
SELECT
  t5.ORDER_ID :: VARCHAR(32),
  ((CONCAT(t5.LASTNAME, t5.FIRSTNAME)) :: VARCHAR(40)) :: VARCHAR(255),
  t5.AGE :: DOUBLE,
  ADU0."PRODUCT_NAME" :: VARCHAR(32)
FROM
  "DEMO_DB"."DEMO"."ADULT_ORDER_1695197685139_7f5dca55-e292-4a16-a0da-5da8ab230266" AS ADU0
  Join (
    SELECT
      ADU2.LASTNAME,
      ADU2.FIRSTNAME,
      ADU2.LASTNAME_KANA,
      ADU2.FIRSTNAME_KANA,
      ADU2.AGE,
      ADU3."ORDER_ID",
      ADU3."PRODUCT_ID"
    FROM
      (
        SELECT
          ADU1."CUSTOMER_ID",
          ADU1."LASTNAME",
          ADU1."FIRSTNAME",
          ADU1."LASTNAME_KANA",
          ADU1."FIRSTNAME_KANA",
          ADU1."AGE"
        FROM
          "DEMO_DB"."DEMO"."ADULT_ORDER_1695197685111_2ed7ce33-20c3-45a6-b523-7778065aede0" AS ADU1
        WHERE
          '18' <= ADU1."AGE"
      ) AS ADU2(
        CUSTOMER_ID,
        LASTNAME,
        FIRSTNAME,
        LASTNAME_KANA,
        FIRSTNAME_KANA,
        AGE
      )
      Join "DEMO_DB"."DEMO"."ADULT_ORDER_1695197684899_0d5fbdc0-5084-442c-a561-bbed1572ca25" AS ADU3 ON ADU3."CUSTOMER_ID" = ADU2.CUSTOMER_ID
  ) AS t5(
    LASTNAME,
    FIRSTNAME,
    LASTNAME_KANA,
    FIRSTNAME_KANA,
    AGE,
    ORDER_ID,
    PRODUCT_ID
  ) ON t5.PRODUCT_ID = ADU0."PRODUCT_ID"
</pre>
</details>

# マッピングタスクを作って実行する

さてこれで、どんなSQLクエリが生成されるかは分かったので、実際に実行してみます。

実行するには、まず、「[マッピングタスク](https://docs.informatica.com/integration-cloud/data-integration/current-version/tasks/mapping-tasks.html)」という定義体を作成し、その後、そのマッピングタスクを実行します。

1. 左ペインにて「新規」をクリックする。
1. ダイアログで、「タスク」＞「マッピングタスク」をクリックする。
    ![informatica_cdi_pushdown_snowflake_018.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/31898e6d-28b2-9d72-5658-7f5dd6ac5e4f.png)
1. 「作成」ボタンをクリックする。
1. 次のように設定する。
    |設定項目|設定値|
    |---|---|
    |タスク名|【任意】
    |プロジェクト|【デフォルト値のまま】|
    |説明||
    |ランタイム環境|【自分のランタイム環境（Secure Agentをインストールしたマシン）】|
    |マッピング|【先ほど作成したマッピング】|
1. 「次へ」ボタンをクリックする。
1. 「スケジュールと詳細設定」の [「プッシュダウン最適化」パート](https://docs.informatica.com/integration-cloud/data-integration/current-version/tasks/mapping-tasks/mapping-task-configuration/configuring-runtime-options.html) にて、次のように設定する。
    |設定項目|設定値|
    |---|---|
    |プッシュダウンの最適化タイプ|【任意。「全体」など。プッシュダウン先を片寄せしたい場合は他の値を選ぶ。】|
    |最適化コンテキストタイプ|【デフォルト値のまま】|
    |プッシュダウンモードが可能でない場合、タスクをキャンセルしてください|【デフォルト値のまま】|
    |一時的なビューの作成|【デフォルト値のまま】|
    |一時的なシーケンスの作成|【デフォルト値のまま】|

    ![informatica_cdi_pushdown_snowflake_019.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/53437b3c-38fa-3182-52f9-7263a2705174.png)
1. 「完了」ボタンをクリックする。
1. 「実行」ボタンをクリックする。
    ![informatica_cdi_pushdown_snowflake_020.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/d16f6627-7797-55a7-6029-e0e61e7b1e79.png)

# マッピングタスクの実行結果を見る

マッピングタスクを実行できたので、結果を見てみましょう。

次のように操作します。

1. 左ペインにて「[マイジョブ](https://docs.informatica.com/cloud-common-services/monitor/current-version/monitor/monitoring-data-integration-jobs/monitoring-mappings-and-mapping-tasks.html)」をクリックする。
1. 最上部（最新）のジョブインスタンスをクリックする。
1. 「ステータス」に「成功」と表示されていることを確認する。
    ![informatica_cdi_pushdown_snowflake_021.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/39cfe7b4-7370-30aa-fd88-145b168796e3.png)
1. 下部の「プッシュダウン最適化」パートにて、プッシュダウンされたSQLクエリを確認する。

ここで、本当にプッシュダウンされているか、この画面からセッションログをダウンロードして見てみます。

![informatica_cdi_pushdown_snowflake_022.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/df903218-f1d8-2a52-efbe-00a3296e0c1f.png)
![informatica_cdi_pushdown_snowflake_023.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/1fa6a50a-4f5a-34f5-f57a-a8d5a36468da.png)
![informatica_cdi_pushdown_snowflake_024.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/c6585313-3dd0-fcdb-dde5-715700322e4a.png)

実際にプッシュダウンされており、先ほどプレビューで見たSQLクエリが、Snowflakeに発行されていることを確認できました。

おまけとして、Snowsightの [Query History](https://docs.snowflake.com/ja/user-guide/ui-snowsight-activity#query-history) でも見てみましょう。
![informatica_cdi_pushdown_snowflake_025.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3031738/918b9b1b-3266-3a8a-70b1-08089b988ea7.png)

先のSQLクエリが、Snowflakeに発行されていることを確認できました。

# 結局、プッシュダウンでパフォーマンスは上がったの？

さて、ここまで読んできた皆さんが気になるのは、「結局、プッシュダウンの有無によって、データ統合処理のパフォーマンスは向上したの？」という点かと思いますが、それは、次に挙げるような様々な条件によって変わるため、ここで提示するのは控えます。
* プッシュダウンなしのデータ統合処理（ETL処理）で使用するEC2インスタンスのインスタンスタイプ
* プッシュダウンありのデータ統合処理（ELT処理）で使用するSnowflake仮想ウェアハウスのサイズ
* 入出力するデータ量 etc...

大量のデータを処理する場合はプッシュダウン最適化が効果を発揮する場合が多いでしょう。（[プッシュダウンできない種類の変換処理もあります](https://docs.informatica.com/integration-cloud/data-integration-connectors/current-version/snowflake-data-cloud-connector/part-3--sql-elt-with-snowflake-data-cloud-connector/mappings-in-sql-elt-mode-for-snowflake-data-cloud/transformations-in-mappings-in-sql-elt-mode.html)）

# おわりに

以上、「Informaticaのデータ統合サービスCloud Data IntegrationでSnowflakeにELTしてみた」でした。

**[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>