EmbulkでS3→BigQueryにデータを流してredashで可視化する

S3に置いてあるcsvデータをBigQueryにつっこんで可視化したいという要件があったので、Embulkを使ってやってみた時の備忘録。
Embulk触りたいと思ってたけど中々機会なくて、初めてだったけどとても簡単にやりたい事が出来ました。素晴らしいプロダクトだなーと実感！！
最近は、社内でredash熱も高まってるし併せていい機会だった。

##今回のテーマ

AWSの請求情報レポートで出せる「xxxxxx-aws-billing-detailed-line-items-with-resources-and-tags-2016-08.csv.zip」の詳細データを対象にしています。※bigqueryのスキーマjsonのカラム名は、このcsvファイルのカラム名をそのまま利用しています。
具体的には、s3のデータ転送量のみをトラッキングしてみたというところです。

##準備

• Embulk本体のインストール (手元のmac)

$  curl --create-dirs -o ~/bin/embulk -L http://dl.embulk.org/embulk-latest.jar

$ chmod +x ~/bin/embulk

$ cat << 'EOF' >> ~/.bash_profile
export PATH=$PATH:$HOME/bin
EOF

$ source ~/.bash_profile

$ embulk
Embulk v0.8.13
Usage ....
....

• 必要なプラグインのインストール

#in_s3
$ embulk gem install embulk-input-s3

#out_bigquery
$ embulk gem install embulk-output-bigquery

#decorder (awsの請求情報がzip形式でs3に置かれるのでそれを解凍する為)
$ embulk gem install embulk-decoder-commons-compress

##設定ファイル(embulk,bigqueryスキーマ)

Embulkは、基本的に下記のようなyamlの設定ファイルを書くだけでOK。大体のプラグインが揃っているので、やりたいことはプラグインを入れるだけで設定ゴニョゴニョ書けばすぐにデータ投入が出来る！

config.yaml

in:
  type: s3
  bucket: [バケット名]
  endpoint: s3.amazonaws.com
  access_key_id: [アクセスキー] 
  secret_access_key: [シークレットキー]
  path_prefix: xxxxxx-aws-billing-detailed-line-items-with-resources-and-tags-
  decoders:
    - type: commons-compress
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: ','
    quote: '"'
    escape: '"'
    columns:
    - {name: InvoiceID, type: string}
    - {name: PayerAccountId, type: string}
    - {name: LinkedAccountId, type: string}
    - {name: RecordType, type: string}
    - {name: RecordId, type: string}
    - {name: ProductName, type: string}
    - {name: RateId, type: string}
    - {name: SubscriptionId, type: string}
    - {name: PricingPlanId, type: string}
    - {name: UsageType, type: string}
    - {name: Operation, type: string}
    - {name: AvailabilityZone, type: string}
    - {name: ReservedInstance, type: string}
    - {name: ItemDescription, type: string}
    - {name: UsageStartDate, type: timestamp, format: '%Y-%m-%d %H:%M:%S'}
    - {name: UsageEndDate, type: timestamp, format: '%Y-%m-%d %H:%M:%S'}
    - {name: UsageQuantity, type: double}
    - {name: BlendedRate, type: double}
    - {name: BlendedCost, type: double}
    - {name: UnBlendedRate, type: double}
    - {name: UnBlendedCost, type: double}
    - {name: ResourceId, type: string}

out:
  type: bigquery
  mode: append
  auth_method: private_key
  service_account_email: [サービスアカウント]
  p12_keyfile: [p12形式鍵の置き場]
  project: [プロジェクトID]
  dataset: [dataset名]
  table: [テーブル名]
  auto_create_table: true
  schema_file: [BigQueryのスキーマファイル(json)置き場]

###bigqueryのテーブルスキーマは下記の感じで。Embulk parserで各カラムにつけた型と合わせるだけ。

bq-billing-aws.json

[
  {
    "name": "InvoiceID",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "PayerAccountId",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "LinkedAccountId",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "RecordType",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "RecordId",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "ProductName",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "RateId",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "SubscriptionId",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "PricingPlanId",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "UsageType",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "Operation",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "AvailabilityZone",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "ReservedInstance",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "ItemDescription",
    "type": "string",
    "mode": "nullable"
  },
  {
    "name": "UsageStartDate",
    "type": "timestamp",
    "mode": "nullable"
  },
  {
    "name": "UsageEndDate",
    "type": "timestamp",
    "mode": "nullable"
  },
  {
    "name": "UsageQuantity",
    "type": "float",
    "mode": "nullable"
  },
  {
    "name": "BlendedRate",
    "type": "float",
    "mode": "nullable"
  },
  {
    "name": "BlendedCost",
    "type": "float",
    "mode": "nullable"
  },
  {
    "name": "UnBlendedRate",
    "type": "float",
    "mode": "nullable"
  },
  {
    "name": "UnBlendedCost",
    "type": "float",
    "mode": "nullable"
  },
  {
    "name": "ResourceId",
    "type": "string",
    "mode": "nullable"
  }
]

##動かす

一応新規のconfigはdry-run的なpreviewを使ってconfigのテストを実施

#config.yamlのtypo
$ embulk preview config.yaml
2016-08-25 16:07:33.486 +0900: Embulk v0.8.13
while scanning a simple key
 in 'string', line 48, column 3:
      schema_file:~/opt/gits/shoichiro ...
      ^
could not found expected ':'
 in 'string', line 49, column 1:


#プラグインの必須項目セットされてないとか
$ embulk preview config.yaml
Error: com.fasterxml.jackson.databind.JsonMappingException: Field 'path_prefix' is required but not set

dry-run成功(エラー出ない)したら、実行

$ embulk run config.yaml

無事インポートが出来ている。

##Re:dash

Re:dashの基盤が既にあったのでこれを使って簡単なグラフを生成

• 日毎のS3の転送量を集計するクエリ

AWSの請求書を見ながら、トラッキングしたい項目にたどりつくのに少し時間がかかった。(課金体系の項目が多すぎてツライ。。。)

select
  DATE(UsageStartDate) as day,
  sum(UsageQuantity) as Traffic_GB,
  sum(BlendedCost) as Cost_USD
from
  [プロジェクトID:InfraTeam.aws_main_billing] 
where 
  UsageType = 'APN1-DataTransfer-Out-Bytes'
  and ProductName = 'Amazon Simple Storage Service'
group by day
order by day

• 時系列のchartを作る

データ転送量とコストが一目瞭然に!

##※ハマったこと

• bigqueryはテーブル名に、-(ハイフン)はだめ ※よく忘れるがカラム名にもハイフンは使えない
• 最初、zip形式というのを忘れてて、embulkのcsvパーサがひたすらエラーを履き続けた。。

##所感

Embulkは、fluentdと同じプラガブルなアーキテクチャでとても使いやすかった。Fluentdはガシガシ使っていたからか、学習コストほぼ0でやりたい事出来た！
うちの場合は、Redash環境が既にあるのでこれを使って可視化したがスプレッドシートで可視化するのもありだと思う。
最近は、Elasticsearchよく使ってたからそれとの比較すると、JSONでデータ引くのに比べてやはり、sqlでデータ出せるのは楽でよいと実感。

##参照

http://qiita.com/y-ken/items/31890f4f188321cb88a7
http://www.embulk.org/docs/built-in.html