Embulkプラグイン(+ Digdag) memo

Embulkの各プラグインのconfig設定例をメモしていきます。
(MySQL、Bigquery、Redshit、Redis、S3など)
 
プラグインのインストールが必要
例：`embulk gem install embulk-output-bigquery`

embulkをたくさん順序よく叩いて、あいまにちょっとした処理も行いたいので、
digdagも試します。

# Embulkプラグインのconfig(liquid前提)
## Input
### [embulk-input-s3](https://github.com/embulk/embulk-input-s3)
```
in:
  type: s3
  path_prefix: yyy/xxxxxx
  file_ext: .zip
  access_key_id: {{ env.AWS_ACCESS_KEY }}
  secret_access_key: {{ env.AWS_SECRET_KEY }}
  iam_user_name: yyy-xxxxxx-access
  bucket: {{ env.S3_YY_BUCKET }}
  endpoint: s3-ap-northeast-1.amazonaws.com
  decoders:
  - {type: gzip}
  parser:
    charset: UTF-8
    newline: LF
    type: csv
    delimiter: ','
    quote: '"'
    escape: '"'
    max_quoted_size_limit: 134217728
    columns:
    - {name: url, type: string}
    - {name: site, type: string}
    - {name: contents, type: string}
```

### embulk csv
```
in:
  type: file
  path_prefix: "yyy/xxxxxx.csv"
  parser:
    charset: UTF-8
    newline: LF
    type: csv
    default_timezone: "Asia/Tokyo"
    delimiter: ','
    null_string: 'NULL'
    columns:
    - {name: x, type: string}
    - {name: y, type: string}
```

### [embulk-input-gcs](https://github.com/embulk/embulk-input-gcs)
```
in:
  type: gcs
  bucket: xx-yyy-zzz
  path_prefix: xxx/zzz_
  auth_method: private_key
  service_account_email: {{ env.SERVICE_ACCOUNT_EMAIL }} 
  p12_keyfile: ../bq/key/{{ env.P12_FILENAME }}
  application_name: zzzz
  tasks: 1
  parser:
    charset: UTF-8
    newline: LF
    header_line: true
    type: csv
    delimiter: ','
    quote: '"'
    columns:
    - {name: site, type: string}
    - {name: category, type: string}
```

### [embulk-input-redis](https://github.com/komamitsu/embulk-input-redis)
```
in:
  type: redis 
  host: localhost
  port: 6379
  db: 1
  key_prefix: yyy-xxxxxx
  encode: hash
  columns:
  - {name: site, type: string}
  - {name: category, type: string}
```

### [embulk-input-mysql](https://github.com/embulk/embulk-input-jdbc/tree/master/embulk-input-mysql)
```
in:
  type: mysql
  host: {{ env.XXXXX_DB_HOST }}
  user: {{ env.XXXXX_DB_USER }}
  password: {{ env.XXXXX_DB_PASSWORD }} 
  database: {{ env.XXXXX_DB_NAME }}
  table: xxxx
  select: "aaa, bbb, ccc"
  where: "status = 0 AND id = 12345"
```

### [embulk-input-redshift](https://github.com/embulk/embulk-input-jdbc/tree/master/embulk-input-redshift)
```
  type: redshift
  host: {{ env.XXXXX_RS_DB_HOST }}
  user: {{ env.XXXXX_RS_DB_USER }}
  password: {{ env.XXXXX_DATABASE_PASSWORD }}
  database: {{ env.XXXXX_DATABASE_NAME }}
  table: xxx
  select: "name,url,title"
  fetch_rows: 1000
```

## Output
### [embulk-output-bigquery](https://github.com/embulk/embulk-output-bigquery)
```
out:
  type: bigquery
  mode: append
  auth_method: private_key
  service_account_email: {{ env.SERVICE_ACCOUNT_EMAIL }}
  p12_keyfile: key/{{ env.P12_FILENAME }}
  path_prefix: tmp/
  file_ext: csv.gz
  source_format: CSV
  project: {{ env.XXXX_PROJECT }}
  dataset: xxx_yyyy_zzz
  auto_create_table: true
  table: xxxes
  schema_file: schema/yyyy_zzzz_schema.json
  prevent_duplicate_insert: true
  formatter:
    type: csv
    header_line: false
    timezone: Asia/Tokyo
  encoders:
  - {type: gzip}
  allow_quoted_newlines: 1
```

大量データを投入する場合にはtimeout_secとopen_timeout_secを設定
```
  timeout_sec: 6000
  open_timeout_sec: 6000
```
そうしないと
```
Caused by: org.jruby.exceptions.RaiseException: (TransmissionError) execution expired
```


### [embulk-output-td](https://github.com/treasure-data/embulk-output-td)
```
out:
  type: td
  apikey: {{ env.TD_API_KEY }}
  endpoint: api.treasuredata.com
  database: xxx_yyyy_zzz
  table: xxxes
  default_timezone: Asia/Tokyo 
```

### [embulk-output-redis](https://github.com/komamitsu/embulk-output-redis)
```
out:
  type: redis
  host: localhost
  port: 6379
  db: 0
  key_prefix: "yyy-xxxxxx"
  encode: hash
  key: url
  columns:
  - {name: url, type: string}
  - {name: category, type: string}
```

### [embulk-output-dynamodb](https://github.com/sakama/embulk-output-dynamodb)
```
out:
  type: dynamodb
  mode: upsert
  region: ap-northeast-1
  auth_method: basic
  access_key_id: {{ env.AWS_ACCESS_KEY }}
  secret_access_key: {{ env.AWS_SECRET_KEY }}
  auto_create_table: false
  table: xxxes
  primary_key: id
  primary_key_type: Number
  write_capacity_units:
    normal: 5
    raise: 20
  read_capacity_units:
    normal: 6
    raise: 30
```

### [embulk-output-mysql](https://github.com/embulk/embulk-output-jdbc/tree/master/embulk-output-mysql)
```
out:
  type: mysql
  host: {{ env.XXX_DB_HOST }}
  user: {{ env.XXX_DB_USER }}
  password: {{ env.XXX_DB_PASSWORD }}
  database: {{ env.XXX_DB_NAME }}
  table: yyyy
  mode: truncate_insert
  default_timezone: "Japan"
```

### embulk csv
```
out:
  type: file
  path_prefix: tmp/yyy
  file_ext: txt
  formatter:
    type: csv
    charset: UTF-8
    delimiter: '\'
    header_line: false
    newline: LF
    quote: ''
    quote_policy: NONE
    escape: '\'
```

### [embulk-output-redshift](https://github.com/embulk/embulk-output-jdbc/tree/master/embulk-output-redshift)
```
out:
  type: redshift
  host: {{ env.XXXXX_RS_DB_HOST }}
  user: {{ env.XXXXX_RS_DB_USER }}
  password: {{ env.XXXXX_RS_DB_PASSWORD }}
  database: {{ env.XXXXX_RS_DB_NAME }}
  table: xxxes_tmp
  access_key_id: {{ env.AWS_ACCESS_KEY }}
  secret_access_key: {{ env.AWS_SECRET_KEY }}
  iam_user_name: yyy-xxxxxx-access
  s3_bucket: {{ env.S3_BUCKET }}
  s3_key_prefix: redshift
  mode: truncate_insert
  default_timezone: "Japan"
```

## Filter
### [embulk-filter-ruby_proc](https://github.com/joker1007/embulk-filter-ruby_proc)

```
filters:
  - type: ruby_proc
    requires:
      - cgi
    variables:
      multiply: 1
    before:
      - proc: |
          -> do
            puts "before proc"
            @started_at = Time.now
          end
    after:
      - proc: |
          -> do
            puts "after proc"
            p Time.now - @started_at
          end
    columns:
      - name: words
        proc: |
          ->(words) do
            words # ここにrubyで処理を書く
          end
        type: string
```

## parser
### [embulk-parser-none](https://github.com/sonots/embulk-parser-none)
parseさせたくないときもあります。

```
in:
  type: file
  path_prefix: example/example.jsonl
  parser:
    type: none
    column_name: payload
out:
  type: bigquery
  payload_column_index: 0 # or, payload_column: payload
```


## etc
### max_quoted_size_limit
`The size of the quoted value exceeds the limit size (131072)`
http://www.embulk.org/docs/built-in.html
134217728などの大きめの値にしたら、エラーが出なくなった


# Digdag 

## install 
```
sudo curl -o /usr/local/bin/digdag --create-dirs -L "https://dl.digdag.io/digdag-latest"
sudo chmod +x /usr/local/bin/digdag
```

javaのバージョンがJava8以上でないとだめ

```
sudo yum -y remove java
sudo yum -y install java-1.8.0-openjdk-devel
java -version
javac -version
```

## ワークフローの初期化・作成
```
digdag init workflow
```

```
2016-07-06 15:14:26 +0900: Digdag v0.8.3
  Creating workflow/.gitignore
  Creating workflow/tasks/shell_sample.sh
  Creating workflow/tasks/repeat_hello.sh
  Creating workflow/tasks/__init__.py
  Creating workflow/workflow.dig
Done. Type `cd workflow` and then `digdag run workflow.dig` to run the workflow. Enjoy!
```

```yaml:workflow.dig
timezone: Asia/Tokyo

+step1:
  sh>: embulk run s3/yyy-xxxxxx.yml.liquid

+last:
  py>: tasks.XXXYYYWorkflow.last_notification
```

```py:tasks/__init__.py
#coding:utf-8

import os
import slackweb
import digdag

class XXXYYYWorkflow(object):
    def xxx_notification(self):
        slack = slackweb.Slack(url=os.environ["SLACK_INCOMING_URL"])
        slack.notify(text="This is a *digdag test*...", \
        channel="#system-log", username="dig_dagger", icon_emoji=":coffee", mrkdwn=True)
```

## 実行
embulkを実行し、最後にslackへ通知した
`> digdag run workflow`

```
Success. Task state is saved at .digdag/status/20160706T000000+0900 directory.
```

## 再実行
一度実行すると.digdag/status下に実行ログ（&ステータス）がたまる。
その下のフォルダやファイルを削除しても再実行できるが、rerunというオプションもある
`> digdag run --rerun workflow`

## architecture
http://www.digdag.io/architecture.html#architecture

# 結論
  - embulkはすばらしい！
  - digdagは今後に期待