Zenn
📝

EmbulkでTreasure DataのテーブルからAWS S3にCSV出力する

2020/11/17に公開
S3
embulk
Treasure Data
tech

はじめに

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のインストール方法は公式ページに記載があります。
https://www.embulk.org/

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