Digdag+Embulkで発生したエラーと対策

はじめに

EC2にDigdagとEmbulkをインストールした環境にMacからssh接続して実行していると複数のエラーが発生しました。
それらの対応策を備忘録として書いていきます。

環境

Mac OSX 10.9.5 Amazon EC2 Amazon Linux
Digdag 0.8.13
Embulk 0.8.13

目的

S3に配置してあるGZIP圧縮されたTSVファイル（ヘッダー無し）をRedshiftにロードする。
DigdagとEmbulkの実行はEC2インスタンス上。

実行時の状態

Digdag実行ファイル

Digdagで実行する処理内容をYAMLで記述しています。
guessで「guess-redshift-output.yml」を読込み、追記修正したものを「config-redshift-output.yml」に上書き保存し、Embulkで読込んでいます。

timezone: UTC
 
+guess:
  sh>: embulk guess embulk/guess-redshift-output.yml -o embulk/config-redshift-output.yml
 
+load:
  embulk>: embulk/config-redshift-output.yml

設定ファイル： guess-redshift-output.yml

こちらもYAMLでS3とRedshiftの接続情報を記述。

in:
  type: s3
  bucket: S3のバケット名
  path_prefix: S3内の対象ファイル名（プレフィックスで読込み可能）
  endpoint: s3-ap-northeast-1.amazonaws.com
  access_key_id: XXXXX
  secret_access_key: XXXXX
out:
  type: redshift
  host: Redshiftのエンドポイント
  user: Redshiftのマスターユーザー名
  password: Redshiftのマスターパスワード
  database: データベース名
  schema: スキーマ名
  table: テーブル名
  access_key_id: XXXXX
  secret_access_key: XXXXX
  iam_user_name: XXXXX
  s3_bucket: S3バケット名
  s3_key_prefix: temp/redshift
  mode: insert

実行

Digdagの実行

$ digdag run 実行ファイル名

実行結果

Redshiftにロードが完了すれば下記のログが最後に表示されます。

Success. Task state is saved at .digdag/status/20161130T000000+0000 directory.
  * Use --session <daily | hourly | "yyyy-MM-dd[ HH:mm:ss]"> to not reuse the last session time.
  * Use --rerun, --start +NAME, or --goal +NAME argument to rerun skipped tasks.

失敗した場合はこんな表示です。

error: 
  * +csack+load:
    Command failed with code 1

Task state is saved at .digdag/status/20161130T000000+0000 directory.
  * Use --session <daily | hourly | "yyyy-MM-dd[ HH:mm:ss]"> to not reuse the last session time.
  * Use --rerun, --start +NAME, or --goal +NAME argument to rerun skipped tasks.

発生した問題

データソースにより、現状の設定では複数のエラーや想定外の結果となったケースが発生したので、問題別の対策を行いました。
参考リンク：Embulk 0.8 documentation

OutOfMemoryError

エラーログ。

2016-12-01 10:04:47 +0000 [ERROR] (task-thread-0): Uncaught exception. Task heartbeat for at-least-once task execution is not implemented yet.
java.lang.OutOfMemoryError: Java heap space

発生しても一応処理は進みましたが、問題なのでヒープサイズの最大値を設定しました。
1024の倍数で設定しますが、オペレーター（sh>:などの事）がembulkだとオプションが実行できない様なので、sh>:に変更しました。

timezone: UTC
 
+guess:
  sh>: embulk guess embulk/guess-redshift-output.yml -o embulk/config-redshift-output.yml

+load:
  sh>: embulk -J-Xmx2048m run embulk/config-redshift-output.yml

ロード件数の不一致 ＆ 既存のテーブルにロード失敗

件数は下記で確認しました。

データソースの行数確認

$ wc -l データソースのパス

ロードされた件数確認

SELECT count(*) FROM テーブル名;

対策

データソースにカラム名のヘッダーを追記しました。方法としては、ヘッダーのみのファイルを作成し、データソースと結合します。ファイルサイズが大きいと、エディターで記述しようと開いたり保存するだけでも時間がかかるのでコマンドを使うのが効率的かと思いました。
注意点として、ヘッダーファイルは最後に改行しておきます。そうしないとヘッダーを記述した１行目のお尻にデータソースが結合されてしまいます。 また、1行目はヘッダーとして扱うので、プレフィックスで複数のデータソースを読み込む場合は全てにヘッダーを結合しておきます。

$ cat ヘッダーファイル データソース名 > 結合したファイル名

YAMLの作成ミス

データソースによってはカラムの型の設定が誤認識されることが有りました。
誤認識の内容は、圧縮形式、ヘッダー有無、カラムの型設定でした。

対応

カラムの方はとりあえずstringにして事なきを得ました。
一応、実行ログを確認して、「skipped line」と出て来る場合は、表示された行が型に適応していないので、適切な型を設定しなおします。

in:
  type: s3
  bucket: S3のバケット名
  path_prefix: S3内の対象ファイル名（プレフィックスで読込み可能）
  endpoint: s3-ap-northeast-1.amazonaws.com
  access_key_id: XXXXX
  secret_access_key: XXXXX
  decoders:
  - {type: gzip}
  parser:
    charset: UTF-8
    newline: CRLF
    type: csv
    delimiter: "\t"
    quote: '"'
    escape: '"'
    trim_if_not_quoted: false
    skip_header_lines: 1
    null_string: ""
    allow_extra_columns: false
    allow_optional_columns: false
    columns:
    - {name: ID, type: string}
    - {name: AMOUNT, type: string}
  ...

また、Digdagの実行ファイルもguessから実行しない様に修正する必要があります。

timezone: UTC
 
+load:
  sh>: embulk -J-Xmx2048m run embulk/config-redshift-output.yml

Digdag再実行時の注意点

前回の実行で成功したタスクはスキップされるので、全てのタスクを再実行する場合は --rerun もしくは -a を追記して実行します。
参考リンク：Digdag | Command reference

$ digdag run 実行ファイル名 --rerun

さいごに

今後も問題が発生すれば記事にしていこうと思います。