CSA Data Uploaderが「Snowflake内部ステージへのデータ連携」に対応しました

はじめに

クラスメソッドが展開しているデータ統合基盤サービス『CSアナリティクス』(以降"CSA")のプロダクト群の1つ、『Data Uploader』に、この度Snowflakeの内部ステージへ出力する機能を追加しました。

前回に引き続き、今回はSnowflakeの内部ステージ(以降"内部ステージ")へ接続し、データをアップロードする方法について紹介いたします。

事前準備など

CSA Data UploaderでSnowflake内部ステージへの接続設定はできているものとします。接続設定の手順については前回の記事を参考にしてください。

データアップロード

データアップロードはCSA Data Uploaderの「データアップロードジョブ」を使い、出力先として内部ステージを指定することで行います。「データアップロードジョブ」については以下の記事を参考にしてください。

『Data Uploader』で実践！データアップロードジョブ(MySQLのデータを抽出→変換処理→Amazon S3にアップロード) #データ統合基盤 #CSアナリティクス

以下、「データアップロードジョブ」を作成して内部ステージにアップロードする方法となります。

ジョブの作成

CSA Data Uploaderのジョブ一覧画面にて「ジョブを作成」を押下し、ジョブ種別に「データアップロード」を指定する形でジョブを作成します。

ジョブの枠が出来ました。

入力処理の作成

内部ステージにアップロードするファイルを入力ファイルとして指定します。ジョブの入力側のボタンを押下し、表示される画面の入力種別に「ローカル接続」を指定します。

以下のような形で入力ファイルを指定します。

変換処理の作成

データアップロードジョブの場合、変換処理で入力ファイルのデータを変換し、アップロードすることができます。今回は任意の列のみをアップロードする「列選択」を指定します。

変換の「追加」ボタンを押下し、「変換種別」で「列選択」を指定します。

列選択の画面でアップロードしたい列のみを選択して「保存」ボタンを押下します。

出力処理の作成 : Snowflake内部ステージへアップロード

次に出力処理を作成します。今回はアップロード先に「Snowflake内部ステージアップロード」を指定します。

ファイル種別、区切り文字

データアップロードジョブの場合、前回のファイルアップロードとは異なり、アップロードするファイルの「ファイル種別」と「区切り文字」を選ぶことができます。

それぞれの内容は以下の通りです。

• ファイル種別 : ファイルの種別を「CSV/TSV」形式なのか「JSONL」形式なのかを選択します。
• 区切り文字 : 「ファイル種別」で「CSV/TSV」を選択した場合、カンマ区切りなのかタブ区切りなのかを選択します。

今回はTSVでアップロードするため「CSV/TSV」「タブ区切り(TSV)」を選択しました。

Snowflake接続名

「Snowflake接続名」では前回の「Snowflake内部ステージ接続設定の準備」にて作成したSnowflake接続名を指定します。

アップロードステージ - ユーザーステージ

「アップロードステージ」ではアップロード先の内部ステージとして「ユーザーステージ」「テーブルステージ」「名前付きステージ」のいずれかを選択します。まずは「ユーザーステージ」を選択します。

項目毎の内容は以下の通りです。

• パス : ユーザーステージ配下のフォルダのパスを記載します。指定内容は既存パス、またアップロード時点では存在していないパスのいずれも指定可能です。パス名にはCSA Data Uploaderで提供している「変数」を使うことも出来ます。
• ファイル名 : アップロード対象のファイル名をどうするかを指定します。「入力時のファイル名を使用する」「任意のファイル名を入力」が選択出来ます。

今回はパスとして当日の値を組み込み、ファイル名はそのままアップロードする形で指定してみました。

アップロードステージ - テーブルステージ

「アップロードステージ」にて「テーブルステージ」を選択します。

項目毎の内容は以下の通りです。

• データベース : アップロード先のテーブルが存在するデータベースを指定します。
• スキーマ : アップロード先のテーブルが存在するスキーマを指定します。
• テーブル : アップロード先のテーブルを指定します。

「パス」と「ファイル名」については「ユーザーステージ」などと同じであり、テーブルステージ内の入力したパスに指定したファイル名でアップロードされます。

アップロードステージ - 名前付きステージ

「アップロードステージ」にて「名前付きステージ」を選択します。

項目毎の内容は以下の通りです。

• ステージ名 : アップロードするステージを指定します。

「パス」と「ファイル名」については「ユーザーステージ」などと同じであり、名前付きステージ内の入力したパスに指定したファイル名でアップロードされます。

ここまで入力すると「ユーザーステージ」「テーブルステージ」「名前付きステージ」のいずれでも「アップロード先 ファイル名イメージ」に想定されるアップロード先ファイルパスが表示されるので、「保存」ボタンを押下します。（以下は「ユーザーステージ」での例となります。）

ジョブ詳細

これで一連のジョブ設定が出来上がりました。ジョブ詳細画面を開くと作成したジョブが表示されます。

データアップロードジョブの実行&結果確認

ここまでで作成したジョブを実行し、内部ステージへのアップロードを実行してみます。上記の「アップロードステージ」で「ユーザーステージ」を選択して入力した状態で実行しています。

ジョブ詳細で「保存して実行」を押下し、実行種別は「現在の日時で実行」で試してみます。

処理が完了すると画面右上に完了通知メッセージが表示されます。

実行結果は履歴として確認可能です。メニューの「参照」→「ジョブ実行履歴」に遷移し、対象ジョブの履歴表示アイコンを選択します。

Snowflakeに接続し、「アップロード先 ファイル名イメージ」に出ていたパスのファイルが存在するかを確認します。

まとめ

CSA Data UploaderでSnowflake内部ステージにファイルを変換してアップロードする手順の紹介でした。CSA Data Uploaderは1ヶ月間のトライアル利用が可能となっています。興味をお持ち頂いた方は是非無料版ダウンロードページからインストーラを入手頂き、お試し頂けますと幸いです。

• カスタマーストーリーアナリティクス（CSA）の記事一覧
• CSA: Data Uploader の記事一覧 | DevelopersIO