IICSのGoogle BigQuery V2コネクタのpre SQLでDELETE INSERTを試してみた

データアナリティクス事業本部の鈴木です。

今回はGoogle BigQuery V2コネクタでBigQueryのテーブルにデータを格納する際、pre SQLに設定したSQL文を使い、事前にDELETE文を実行し、DELETE INSERTができるか検証しました。

やりたいこと

Informatica Cloud Data Integrationでマッピングを作成し、ターゲットの接続でGoogle BigQuery V2コネクタを使った際、pre SQLにSQL文を設定することで操作を実行する前に事前にSQLを実行することができます。

Google BigQuery V2コネクタでは操作としてInsert以外にもUpsertのような冪等性が担保されるものを選択できますが、なにかしらの理由でDELETE INSERTを行いたい場合にpre SQLで実現できそうか試してみました。

データセットとデータの準備

今回の構成は、SecureAgentをAmazon EC2上にインストールしておき、Amazon S3上のデータを読み込み、VPN経由でBigQueryにデータを格納するというものです。構成の詳細はマッピングおよびマッピングタスクの設定には大きく関与しないと思うので、なにかしらの方法でそれぞれのサービスに疎通ができているという程度の前提とします。

インプット用のデータは、以下のようなCSVファイルを用意し、Amazon S3に配置しました。

"item_id","item_name","filter_datetime"
"1","Apple","2023/03/10 01:00:00"
"2","Orange","2023/03/12 03:00:00"
"3","Grape","2023/03/15 05:00:00"

BigQueryには以下のようにsample_datasetデータセット配下に、データ格納用のsample_tableテーブルを用意しました。

CREATE TABLE `sample_dataset.sample_table`
(
  item_id STRING OPTIONS(description="商品ID"),
  item_name STRING OPTIONS(description="名前"),
  filter_datetime DATETIME OPTIONS(description="フィルター用の日時情報")
)
OPTIONS(
  description="サンプルテーブル"
);

今回はDELETE文をpre SQLに設定し、繰り返し実行しても冪等なマッピングが作成できるか検証します。pre SQLに設定するSQLは後で記載しますが、sample_data.csvの3つのレコードのうち、item_idが1および2の2つを削除するようなものとしました。繰り返し実行して確かにpre SQLが実行されていることを確認できるようにするためです。

pre SQLに設定するSQL文

今回は以下のSQLを設定しました。前述しましたが、S3に配置した3レコードのうち、item_idが1および2の2つについてはこのSQLが実行されると消えるので、マッピングタスクの実行の度に、1件ずつデータが増えることが期待されます。

SQLの意図としては、例えば当日分だけINSERTされるデータについて、何回やってもその日のデータが重複して格納されないように事前にその日のデータは消しておく、というような場合への応用を想定しています。

DELETE FROM sample_dataset.sample_table 
WHERE filter_datetime >= '2023-03-10T00:00:00'
  AND filter_datetime < '2023-03-15T00:00:00';

IICSリソースの準備

マッピングの作成

今回は以下の2つのアセットを用意しました。

特にターゲットは、ブログのタイトル通り、Google BigQuery V2コネクタを使いました。

まずマッピングを作成します。ここで処理の詳細を定義します。S3からデータを読み込み、filter_datetime列を日付型に変更して、BigQueryのsample_tableテーブルに格納します。テーブルへの格納前に、詳細で設定したpre SQLが実行されます。

日付は以下のように変更し、new_filter_datetimeカラムとしました。

ターゲットで以下のようにフィールドマッピングを設定しました。

ターゲットの詳細では、以下のように接続先のデータセットおよびテーブルを設定しました。

ようやくここでpre SQLを設定します。そのほかの注意点として、Truncate target tableはチェックを入れずそのままにしておきます。

マッピングタスクの作成

作成したマッピングを読み込んで、マッピングタスクを作成します。今回は、pre SQLとは直接関係しませんが、エラーになったときに分かりやすいようにと、BigQueryに日付型のデータをInsertするためにで、セッションプロパティを2つ設定しておきました。

実行結果

マッピングタスクを実行してみます。

初回の実行では、もともとsample_tableテーブルは空なので、3件が格納されます。

再度実行すると、pre SQLに設定したSQLが実行され、4件になることが確認できました。

BigQueryのプロジェクト履歴から、確かに2回実行され、LOADの前にDELETE文が実行されていることが分かります。

パラメータファイルの値を使いたい

開発・本番のように環境が分かれている場合、データセット名はパラメータファイルに設定しておくと便利ですが、pre SQLに設定したSQLからパラメータファイルの値が利用できるかも確認してみました。

まず、ナレッジベースでは以下のソリューションが提案されているのを確認し、実現できることは分かりました。

• HOW TO: Create PreSQL & PostSQL parameterization in IICS - Cloud Data Integration

次に、デフォルトでパラメータファイルが参照されるapps/Data_Integration_Server/data/userparametersのような場所にparameter.txtを作成し、以下のように対象となるデータセット名を設定しました。

[Global]
$$target_dataset=sample_dataset

次に、マッピングに以下のように入出力パラメータを設定しました。

加えて、ターゲットからpre SQLに設定したSQLのうち、パラメータファイルの設定値に合わせて、データセットのところを$$target_datasetとしました。

設定したSQLは以下のようになります。

DELETE FROM $$target_dataset.sample_table 
WHERE filter_datetime >= '2023-03-10T00:00:00'
  AND filter_datetime < '2023-03-15T00:00:00';

最後にマッピングタスクで該当のパラメータファイルを読み込むよう、パラメータファイルの場所を設定しました。今回はローカルを選び、パラメータファイル名をparameter.txtとしました。パラメータファイルディレクトリはデフォルトで参照されるところにパラメータファイルを置いていたため今回は設定していませんが、異なる場所に置いている場合は設定が必要です。

マッピングタスクを保存し、再実行するとBigQueryに想定通りデータが格納されました。

ログからも確かに期待したSQLが実行されたことが確認できました。

最後に

今回はGoogle BigQuery V2コネクタでBigQueryのテーブルにデータを格納する際、pre SQLに設定したSQL文を使い、事前にDELETE文を実行し、DELETE INSERTができるか検証しました。

簡単に事前に実行したいSQLを設定できたのに加え、パラメータファイルのパラメータを使えることも確認できました。

CDIを使ったBigQueryへのデータ連携でpre SQLを利用したい方の参考になれば幸いです。

参考にした資料

• Informatica Documentation TO_DATE
• IICS CDIでOracle接続を試す（マッピング編） | DevelopersIO