システムソリューション部の佐藤奏です。
業務でAWS Data Pipelineを結構ヘビーに使ったので、調べにくいところやハマりどころをQ&A形式でご紹介します。
サービスの概要について少しだけコメントします。その後はひたすら細かい話になります。なお、以下はもっぱらリソースとしてEC2を使う場合の記述です（EC2の他に、EMRクラスタを起動することもできますが、筆者は使ったことがありません）。
概要について少しだけ
Q. Data Pipelineは何がいいの？ cronの方が簡単そうなんだけれど。
A. ジョブを定期実行させる だけ なら、cronの方が簡単なのですが、Data Pipelineは それに付随するもろもろをサポートしてくれます 。例えば下記のようなことを「cron＋シェルスクリプト」だけでやろうとすると結構面倒ですが、Data Pipelineにはそのための仕組みが準備されています。
- ログの記録、ローテーション（パイプラインのログの表示）
- 失敗時のリトライ、リトライしてもダメだった場合の後続処理キャンセル（カスケードの失敗と再実行）
- 入力ファイルが配置されるまで処理を待機（前提条件）
- 正常終了・失敗の通知（アクション）
一方で、Data Pipeline上では条件分岐のようなフロー制御はできないので、複雑な処理には向きません。典型的にはETL処理などに適しています1。
Q. Data Pipelineってデータ処理にしか使っちゃいけないの？
A. そうでもありません 。確かに、S3へのファイル入出力、DBへのSQL発行などのデータ処理に特化したアクティビティが多いですが、ShellCommandActivityを使えば任意のシェルスクリプトが実行できるので、cronの代替としてもいろいろ使えます。弊社ではRedshiftクラスターの落とし上げの自動化にも使っていたりします（AWS CLIをキックするEC2の面倒を見なくて済む）。
Data Pipelineで使うコンピューティングリソースは処理実行時に新規作成し、終わったら破棄する仕組みなので、「実行頻度の低い処理」「実行時だけ大量のコンピューティングリソースが必要な処理」などでは経費節減効果がありますし、運用の手間も減ります。
ひたすら細かい話
Q. AWSコンソールではData Pipelineの全機能が使えますか？
A. いいえ 。パラメータ項目が新規作成できなかったりします（編集は可能）。全機能を使うためには定義ファイルをJSON形式で作成して読み込ませる必要があります。
Q. JSONつらいです。
A. たとえばYAMLで書くのはどうですか？ コメントも書けるし、複数行文字列もそのまま書けるので見やすいと思います。筆者は、定義ファイルはYAML形式の状態でリポジトリ管理し、Data Pipelineに登録（デプロイ）する時にJSONへ変換しました。変換にはnpmモジュールのyamljsを利用しました。
本記事末尾に、YAMLで記述した場合の見た目を掲載しています。
Q. ShellCommandActivity/SqlActivityで実行するスクリプトは「定義ファイルに直書き（ command / script フィールド）」「S3にファイルで配置し、定義ファイルにはパスのみ記述（ scriptUri フィールド）」どっちがいいの？
A. 下記の違いがあります （太字がメリット）。筆者個人的には直書きの方がメリットが大きいと感じます。
| 定義ファイルに直書き | S3に配置 | |
|---|---|---|
| #{} でのパラメータ埋め込み | 可能 | 不可能（※1） |
| 管理対象のファイル | 少ない（定義ファイルのみ） | 多い（定義ファイル＋S3へ配置するスクリプト群） |
| 定義ファイルの見やすさ | 見にくい（長い） | 見やすい |
| スクリプトの差し替え | 面倒（※2） | 簡単（※3） |
（※1）
scriptArgument フィールドでの値埋め込みが可能ですが、
#{} に比べると制約が多いです。
（※2）既存パイプラインの編集よりは、新しいスクリプトを埋め込んだ新規パイプラインをデプロイする方が管理も単純化できるでしょう。
（※3）S3上でのファイル差し替えのみ。
Q. ShellCommandActivity内でAWS CLIを使っているんだけど、うまく動かないです。
A. AWS CLIのバージョンが古いのかもしれません 。パイプラインの先頭で sudo yum -y update aws-cli しましょう2。
Q. ShellCommandActivityの command フィールドにシェルスクリプトを記述してますが、長いコマンドを行末バックスラッシュ \ で区切って複数行で書くとなんかうまくいかないんだけど。
A. JSON上では \ は エスケープして \\ にしないといけません 。行末に限らず同様。
Q. SqlActivityで、複数のクエリをまとめて1つのアクティビティで実行できますか？
A. ドライバ依存 です3。筆者個人的には、まとめすぎない方がいいと思います。どのクエリが失敗したか分かりにくくなるので。
Q. 複数のオブジェクトで同内容のフィールドを記述しているんですが、もう少しDRYに書けませんか？
A. それらのフィールドをまとめたオブジェクトを1つ作り、他のオブジェクトから parent フィールドで参照 しましょう（パイプラインのフィールド）。
Q. というか、”Default”オブジェクトは他の全オブジェクトが暗黙のうちに継承しているから、共通で使いたいものは全部Defaultに書けばいいよね。
A. 大体はそれでもいいのですが 時々ダメ です。例えば Ec2Resource オブジェクトに workerGroup フィールドを設定するとエラーになるので、 workerGroup を”Default”オブジェクトに書くことはできません（オブジェクト内の不要なフィールドは無視される場合が多いのですが、たまに無視されずエラーになるものがあってこういうことになります）。
Q. 「Redshiftリストア（ShellCommandActivityでAWS CLIを実行）→クエリ発行（SqlActivity）」というパイプラインを作ったんだけど、クエリがうまくいきません。
A. Redshiftリストア直後に、 リストアにかかる時間ぶんsleepを入れましょう 。AWS CLIでRedshiftをリストアすると、リストア完了を待たずにCLIは終了します。するとShellCommandActivityも終了してしまうので、リストアが完了する前にSqlActivityが実行されてしまいます。
Q. id フィールドとは別に name フィールドも必ず書かなきゃいけないの？
A. name は省略可能 です。
Q. input フィールドには配列で複数の項目を設定できますが、このとき他のフィールドから #{input[1].directoryPath} のような形でn番目の項目を参照できますか？
A. 現状ではできません 4。
Q. 1つのパイプライン定義に入れられるオブジェクトの最大数はいくつですか？
A. 当記事執筆時点では 100 です（AWS Data Pipeline の制限）。
Q. 100オブジェクトを超えてしまってData Pipelineに怒られました。
A. 弱りましたね 。そういう場合は2つのパイプラインに分けるしかないのですが、現状、パイプライン同士の先行・後続関係の定義はできないので、例えば下記のような作りにすることになります。
- 先行パイプラインの最後で「処理完了ファイル」（中身は空でOK）をS3に配置する
- 後続パイプラインの先頭アクティビティで、上記ファイルの存在を「前提条件（S3KeyExists）」とする
Q. 例えば処理を20分周期で実行させたいとすると、EC2が1時間あたり3個立ち上がることになって、料金がかさみませんか？
A. EC2は1時間周期にしておいて、使い回す ということができます（スケジュールを使用したリソースの最大効率）。
Q. AWS CLIのput-pipeline-definitionを使って下記のようにしたんですが、定義ファイル data_pipeline.json 内の日本語が化けた状態で登録されます。
aws datapipeline put-pipeline-definition --pipeline-definition file://data_pipeline.json (後略)
A. file:// を fileb:// （バイナリファイル扱い）にしてみてください。
Q. テスト実行の時、EC2起動でいちいち時間を取られるんだけど、なんとかならない？
A. TaskRunnerを使いましょう 。EC2をひとつ、常時稼働させておいて、そのEC2上で処理を走らせることができます。
awslabs/data-pipeline-samples にある RedshiftToRDS_WithoutRDSCreate.json
をYAMLで記述してみたサンプルです。個人的には括弧が少なくなって見やすいと感じます。
objects:
- id: "RdsDatabase"
name: "RdsDatabase"
type: "RdsDatabase"
databaseName: "#{myRDSDatabaseName}"
'*password': "#{*myRDSPassword}"
rdsInstanceId: "#{myRDSInstanceId}"
username: "#{myRDSUsername}"
- id: "RedshiftCluster"
name: "RedshiftCluster"
type: "RedshiftDatabase"
databaseName: "#{myRedshiftDatabaseName}"
'*password': "#{*myRedshiftPassword}"
clusterId: "#{myRedshiftInstanceId}"
username: "#{myRedshiftUsername}"
- id: "DefaultSchedule"
name: "RunOnce"
type: "Schedule"
occurrences: "1"
period: "1 Day"
startAt: "FIRST_ACTIVATION_DATE_TIME"
- id: "RedshiftToS3"
myComment: This object is a RedshiftCopyActivity. It is used to define the work that will be done to copy the data from Redshift to S3.
name: "RedshiftToS3"
output: { ref: "S3OutputLocation" }
type: "RedshiftCopyActivity"
input: { ref: "RedshiftDataNode" }
schedule: { ref: "DefaultSchedule" }
runsOn: { ref: "Ec2Instance" }
insertMode: "TRUNCATE"
# ... snip ...
parameters:
- id: "myRDSInstanceId"
type: "String"
description: "RDS Instance name (DB Instance)"
- id: "myRDSDatabaseName"
type: "String"
description: "RDS Database name"
- id: "myRDSUsername"
type: "String"
description: "RDS Database Username"
- id: "*myRDSPassword"
type: "String"
description: "RDS MySQL password"
# ... snip ...
YAMLの場合、文字列値の "" は省略可能ですが、パラメータ埋め込み #{} がコメントと解釈されるのを防ぐため、一律 "" で囲うのがよいと思います。
なお、 | +改行 のあとに文字列値を書くこともできます。この場合、文字列内の改行を \n とせずそのまま書けます（ただしこの場合もバックスラッシュ \ 自体はエスケープが必要）。シェルスクリプトやSQLの記述に便利です。例えばこんな感じになります。
- id: "install"
type: "ShellCommandActivity"
# ... snip ...
command: |
sudo yum -y update aws-cli || exit 1
some_command1 arg1 arg2 && \\
some_command2 arg1 arg2 && \\
some_command3 arg1 arg2
- id: "INSERT_activity"
type: "SqlActivity"
# ... snip ...
script: |
INSERT INTO myschema.mydata (
id
, name
)
SELECT
id
, name
FROM dwh.master_user
;
最後になりますが、ALBERTでは、データサイエンティストを積極募集しています。
ぜひ採用ページをご覧ください。
- 参考：AWS Black Belt Tech シリーズ 2015 – AWS Data Pipeline ↩
- 参考：AWS DataPipeline の ShellCommandActivity を使って特定のタグを持つインスタンスを日次停止する – Qiita ↩
-
参考：AWS Developer Forums: More than one SQL statement in Script …
“Unfortunately, it depends on the driver whether multiple statements can be executed in the script parameter for the SQL activity.” ↩ - 参考：variables – AWS data pipeline activity with multiple inputs – Stack Overflow ↩
