EmbulkでTreasure DataのテーブルからAWS S3にCSV出力する
はじめに
EmbulkでTreasure Data(以降、TD)のテーブルからAWS S3にCSV出力しなくても、TDの機能で出力すればいいでしょ、と思いますが、TDの機能でCSV formatterなど制約があって少し不便な時があるようです。
本記事では、EmbulkでTDテーブルからAWS S3にCSV出力する方法を記載しました。そして、TDで現状できないダブルクォーテーションでデータを囲ったCSVの出力を確認します。
前提
- バージョン
- Embulk v0.9.23
- embulk-input-td v0.2.3
- embulk-output-s3 v1.5.0
- その他
- TDをまあ知っている。
- Embulkをまあ知っている。
TDでの制約
TD WorkflowのResult Setting、QueriesのExport Resultsでは以下の制約があります。
- CSV formatterのquote_policyがMINIMAL固定。
この制約で、CSVファイルのデータ全てをダブルクォーテーション「"」で囲いたいときに対応できません。(ファイル連携先のシステムから、「"」で囲ってあるCSVデータを要求された場合など困ります。)
Embulkインストール
業務を前提とするなら、AWSのEC2にインストールするのが良いと思います。また、EC2を使う場合はTDとリージョンを合わせることでネットワークトラフィック費用を抑えることができます。
Embullkのインストール方法は公式ページに記載があります。
Pluginのインストール
今回は、2つのPluginを使用します。Pluginのインストールは簡単です。
(コマンド例はmacのプロンプトとしています。)
embulk-input-td
embulk-input-tdのインストールは以下のコマンドとなります。
% embulk gem install embulk-input-td
embulk-output-s3
embulk-output-s3のインストールは以下のコマンドとなります。
% embulk gem install embulk-output-s3
Embulkのymlファイルの設定
TDテーブルからS3へのcsv出力設定です。
TDテーブル(インプット)
インプットのembulk-input-td設定はこんな簡単です。
in:
type: td
apikey: XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
endpoint: api.treasuredata.com
database: sample_db
query: |
SELECT * FROM sample_tb
- apikeyは、SELECTするテーブルがあるデータベースの権限を持っているユーザーのものをセットします。
- endpointはTDのリージョンによって変わります。(TD Documentation Sites and Endpoints
) - queryには出力するクエリを記述します。
S3(アウトプット)
アウトプットでは、embulk-output-s3の設定とCSV formatterの設定を記述します。
out:
type: s3
path_prefix: sample/csv_test
file_ext: .csv
sequence_format: '-%02d'
bucket: sample-bucket
endpoint:
access_key_id: XXXXXXXXXXXXXXXXXX
secret_access_key: XXXXXXXXXXXXXXXXXXXXXXXXXXX
formatter:
type: csv
delimiter: ","
newline: LF
newline_in_field: LF
charset: UTF-8
quote_policy: ALL
quote: '"'
escape: '\'
null_string: ""
default_timezone: 'UTC'
# column_options:
# mycol1: {format: '%Y-%m-%d %H:%M:%S'}
- sequence_formatは出力されるcsvファイルが分割されるため、シーケンス番号をファイル名に振るフォーマットを指定します。(デフォルトは'.%03d.%02d')本記事では「-XX」と振るように設定しています。
- 実行タスク数で、ファイルが分割されます。
- endpointは未入力でOKです。
- quote_policyは、ALL、MINIMAL、NONEの設定が可能です。(本記事はALL指定)
- 必要ならcolumn_optionでフォーマット指定します。
- その他設定は、embulk-output-s3の記載をご確認ください。
ymlファイル
結果、ymlファイルはこんな感じの中身になります。
ファイル:sample.yml
in:
type: td
apikey: XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
endpoint: api.treasuredata.com
database: sample_db
query: |
SELECT * FROM sample_tb
out:
type: s3
path_prefix: sample/csv_test
file_ext: .csv
sequence_format: '-%02d'
bucket: sample-bucket
endpoint:
access_key_id: XXXXXXXXXXXXXXXXXX
secret_access_key: XXXXXXXXXXXXXXXXXXXXXXXXXXX
formatter:
type: csv
delimiter: ","
newline: LF
newline_in_field: LF
charset: UTF-8
quote_policy: ALL
quote: '"'
escape: '"'
null_string: ""
default_timezone: 'UTC'
embulk実行
embulkの実行は、作成したymlファイルのディレクトリで、以下のコマンドを実行します。
% embulk run sample.yml
S3出力結果
本記事では、quote_policy=ALLの出力結果を確認します。
S3
CSVファイルが分割されて、シーケンス番号が振られます。
CSVファイル
CSVファイルはquote_policy=ALL設定のため、データがダブルクォーテーション「"」で囲まれています。
タスク数
実行時のタスク数はEmbulkのログに出力されています。今回は8タスクで実行されています。ですので、CSVファイルは8(分割)ファイルとなっています。
終わりに
TDからデータをダブルクォーテーションで囲ったCSVファイルを出力したい場合、TDが対応してくれるまではEmbulkが楽だと思います。
謝辞
Embulk 及び Embulk Pluginを開発された方々、大変便利なツールをありがとうございます。
Discussion