定型データ移動処理のスケジュール自動化設定サービス『AWS DataPipeline』の構成要素をひと通り整理してみた

AWSをお使いの皆さん、『AWS Data Pipeline』というサービスを御存知でしょうか？そしてAWSをお使いの皆さん、AWS Data Pipelineをお使いになられているでしょうか！ ( °д°) ｸﾜｯ!!

『AWS Data Pipeline』のフレーズで検索を掛けてみると、国内はもとより海外でも、そこまで(AWSの他サービスと比べて)言及数や情報共有されている形跡が無いようなイメージです(笑) 昨今、ビッグデータや統計・分析がもてはやされている(とは言え最近は以前よりは落ち着いて来たでしょうか)中で、データ連携の重要性がより一層認識されている現在、比較的リリースされてから時間の経つこのサービスがクローズアップされていないのは勿体無いな〜と思っておりました。

思っておりましたが、私自身も興味は持ちつつそこまでAWS Data Pipelineに対する知識や利用経験・ノウハウは持ち合わせておりませんでした。そこでこのタイミングを機に、当エントリでAWS Data Pipelineはどんなものなのか、現状何が出来て何が出来ないのか、実利用を進める上で気を付ける点にはどのようなものがあるのか、使い勝手等について当ブログ上で検証・確認・横展開して行ければな〜と思っております。どうか長い目で見守って頂ければと思います。m(_ _)m

目次

• AWS Data Pipelineとは
• AWS Data Pipeline 現在の対応リージョン
• AWS Data PipelineがVPC環境内で使えるようになったのは、実は最近のお話。
• AWS Data Pipelineの構成要素
• タスクランナー(TaskRunner)
• データノード(DataNodes)
  • DynamoDBDataNode
  • MySqlDataNode
  • RedshiftDataNode
  • S3DataNode
• データベース(Databases)
  • JdbcDatabase
  • RdsDatabase
  • RedshiftDatabase
• アクティビティ(Activities)
  • CopyActivity
  • EmrActivity
  • HiveActivity
  • HiveCopyActivity
  • PigActivity
  • RedshiftCopyActivity
  • ShellCommandActivity
  • SqlActivity
• 前提条件(Preconditions)
  • DynamoDBDataExists
  • DynamoDBTableExists
  • S3KeyExists
  • S3PrefixNotEmpty
  • Exists
  • ShellCommandPreCondition
• スケジュール(Schedules)
  • 時系列形式のスケジュールとCron形式のスケジュール
  • 埋め戻し(Backfill)タスク
  • スケジュールを使用した最大限のリソース効率化
  • タイムゾーン
  • データの上書き防止
• リソース(Resources)
  • • Ec2Resource
    • EmrCluster

• アクション(Actions)
  • SnsAlarm
  • Terminate
• Roles and Permissions(役割と権限)
• データフォーマット(DataFormat)
  • CSV
  • Custom
  • DynamoDBDataFormat
  • DynamoDBExportDataFormat
  • RegEx
  • TSV
• オプションフィールド(Optional Field)
• 式と関数
• まとめ

AWS Data Pipelineとは

• Amazon Data Pipeline ( データワークフローオーケストレーションサービス) | アマゾン ウェブ サービス（AWS 日本語）

AWS Data Pipelineは各種データ移動や定型処理を自動化したり、自動化したタスク間の制御等も行えるサービスとなっています。S3やRDS、DynamoDB、Redshiftと言ったデータソース等の連携もサポートしているので、AWS環境上で日々実行しているバッチ処理等がある場合はAWS Data Pipelineへの移行も検討の価値があるのかなと思います。

AWS Data Pipeline 現在の対応リージョン

現在AWS Data Pipelineが利用可能なリージョンは以下の通りです。東京もバッチリ対応しています！

• US East (N.Virginia) / us-east-1
• US West (Oregon) / us-west-2
• EU (Ireland) / eu-west-1
• Asia Pacific (Tokyo) / ap-northeast-1
• Asia Pacific (Sydney) / ap-southeast-2

AWS Data PipelineがVPC環境内で使えるようになったのは、実は最近のお話。

...がしかし。

VPC環境内でAWS Data Pipelineを使おうとした場合、一部制約により十分な利用が出来ない状況でした。下記フォーラムで通知があったように、AWS Data Pipelineが(現在のAWSでのEC2起動時構成のデフォルトとなっている)VPCに対応したのは2014年03月以降(2014年03月16日)のお話。実際自分も(VPC対応前の時期に)使ってみようと思ったものの、EC2リソースを伴う処理を組もうとしてこの壁に当たり、断念した経緯がありました。VPC対応が為されてから1年経っていない、という状況ももしかしたら情報やノウハウの横展開が十分でない要因の一つなのかも知れません。

• AWS Developer Forums: AWS Data Pipeline announces support for Amazon VPC

AWS Data Pipelineの構成要素

ここからはAWS Data Pipelineの中身について見て行きたいと思います。AWS Data Pipelineで定義されているオブジェクトは以下の通り。それぞれ階層構造を持った形で整理されています。

• Pipeline Object Reference - AWS Data Pipeline

主なオブジェクトとしては以下の通り。以降のスペースで各種オブジェクトについて内容を解説して行きます。

データノード(Data Nodes):

タスクに対する入力データ、または出力データが格納される場所。

アクティビティ(Activities):

計算リソースと入出力データノードを用いてスケジュールに基づいて作業を実行ための定義。

前提条件(Preconditions):

処理実行時に真(true)となるべき条件文。

スケジュール(Schedules):

『アクティビティがいつ実行するか』のような、スケジューリングされたイベントの日付時刻定義。

リソース(Resources):

パイプラインが定義した作業を遂行するための計算リソース。

アクション(Actions):

『アクティビティ失敗時』のように、指定の条件に合致した際に実行されるアクション。

TaskRunner(タスクランナー)

タスクに対応したAWS DataPipelineをポーリングし、それらのタスクを実行するアプリケーション。AWSで提供されているTaskRunnerまたはカスタムTaskRunnerを選ぶ事が出来ます。

• Task Runners - AWS Data Pipeline
  • Task Runner on AWS Data Pipeline-Managed Resources
  • Task Runner on User-Managed Resources
  • Custom Task Runners
  • Task Runner and Custom AMIs

• Task Runner の操作 - AWS Data Pipeline
  • Task Runner のインストール - AWS Data Pipeline
  • Task Runner の起動 - AWS Data Pipeline
  • Task Runner のログの検証 - AWS Data Pipeline
  • Task Runner のスレッド - AWS Data Pipeline
  • 長時間実行される前提条件 - AWS Data Pipeline
  • Task Runner 設定オプション - AWS Data Pipeline

DataNodes(データノード)

AWS Data Pipelineでは、データノード(DataNodes)が、pipelineアクティビティが入力または出力として使うロケーションとデータタイプを定義しています。AWS DataPipelineは以下のデータノードをサポートしています。

Databases(データベース)

AWS DataPipelineでは以下のデータベースをサポートしています。

Activities(アクティビティ)

アクティビティ(Activity)は、実行する作業を定義するコンポーネントです。AWS Data Pipelineでは、『ある場所から別の場所へデータを移動する』『Hiveクエリを実行する』と言った様な、一般的なシナリオに対応する幾つかの事前パッケージされたアクティビティを提供しています。

アクティビティは拡張可能です。独自のカスタムスクリプトを作成する事も出来ますので、アイデア次第では無限の組み合わせがサポート可能です。AWS DataPipelineでは以下のアクティビティをサポートしています。(幾つかのアクティビティについてはステージング用のデータとデータベーステーブルをサポートしているものもあるようです。詳細については以下ページをご参照ください)

• Staging Data and Tables with Pipeline Activities - AWS Data Pipeline

前提条件(Preconditions)

AWS DataPipelineでは、前提条件(Precondition)も用意されています。アクティビティ実行前に『真(true)であるべき』条件文を含むパイプラインコンポーネントです。例えば、あるアクティビティかそれをコピー処理する場合、その対象となるソースデータが存在するか？という前提条件をチェック出来ます。

AWS Data Pipelineでは『共通のAmazon S3のキーが存在するか』『データベーステーブルが存在するか』と言ったシナリオに対応するための幾つかの事前パッケージ済みされた前提条件を提供しています。また、事前条件は拡張可能です。

前提条件には『システム管理の前提条件』と『ユーザー管理の前提条件』の2種類があります。

• システム管理の前提条件：AWS Data Pipeline Webサービス側であなたに代わって実行を行います。処理に用いるリソースは必要ありません。
• ユーザー管理の前提条件：利用ユーザがrunsOnやworkerGroupフィールドを指定するリソース上で実行を行います。

AWS Data Pipelineでは以下の『前提条件』を提供しています。

パイプラインコンポーネント間の依存について

AWS Data Pipelineでは、『他のシステムやタスクの作業に依存するタスク』の様な、依存性を持つパイプラインを作成する事が可能です。

パイプラインが特定のリソースを必要とする場合、前提条件(precondition)を使ってそれらの依存性をパイプラインに追加します。前提条件にてデータノードとタスクを関連付ける事で、よりデバッグが簡単になり弾力性も得られます。可能であれば、依存は単一のパイプラインに留めておいた方が良いでしょう。クロスパイプラインのトラブルシューティングは厄介です。

スケジュール(Schedules)

• Schedule - AWS Data Pipeline
• スケジュール - AWS Data Pipeline

スケジュール(Schedule)は、『アクティビティをいつ実行するか』と言った、スケジューリングされたイベントの時間を定義します。AWS Data Pipelineはスケジュールパイプラインコンポーネントを介して、この情報を公開しています。

時系列形式のスケジュールとCron形式のスケジュール

スケジュール(Schedule)は、『アクティビティをいつ実行するか』と言った、スケジューリングされたイベントの時間を定義します。AWS Data Pipelineはスケジュールパイプラインコンポーネントを介して、この情報を公開しています。AWS Data Pipelineでは、[時系列(Time Series)形式]と[Cron形式]の2種類のパイプラインスケジュールを提供しています。

スケジュールタイプを使用する事で、パイプラインコンポーネントインスタンスが、インターバル(またはピリオド(period))の開始をいつ始めて、いつ終わらせれば良いのかを指定出来るようになります。

時系列形式のスケジューリングはそれぞれのインターバルの終わりにスケジュールが予定される事を意味し、Cron形式のスケジューリングはそれぞれのインターバルの始めにスケジュールが予定される事を意味します。例えば、時系列形式のスケジューリングで開始時刻が22:00 UTCでインターバル/期間が30分に設定されていた場合、パイプラインコンポーネントインスタンスはまず最初に22:30 UTCに実行されます。(22:00 UTCではありません)

もしインスタンスを期間/インターバルの最初に実行させたい場合(先の例で行くと22:00 UTCに実行させたい場合)、Cron形式を使います。

CLIでは、時系列形式：timeseries、Cron形式：cronで扱われています。

スケジュール形式を無視するリソースについて

AWS Data Pipelineはパイプラインのスケジュール形式設定(時系列orCron形式)に依存する形でスケジュールの開始・終了時にアクティビティやデータノードインスタンスを作成します。

しかし、AWS Data PipelineEC2ResourceやEmrClusterのようにパイプラインスケジュールの形式内容に関わらず、インターバルの開始時にリソースインスタンスを作成するものも存在します。

パイプラインが時系列形式のスケジューリングに設定されている場合、AWS Data Pipelineはリソースインスタンスを作成し、WAITING_ON_DEPENDENCIESステータスを、アクティビティ又はデータノードが開始するよりも早く設定するかも知れません。(ここでの時間＝スケジュール間隔の長さです)

埋め戻し(Backfill)タスク

過去時間での開始時刻でスケジューリングされたパイプラインを定義した場合、AWS Data Pipelineはパイプラインでの『埋戻し(backfill)』作業を行います。 この状況下に於いて、AWS Data Pipelineはそれらのタスクが開始〜終了時刻の間に実行するであろう実行回数に追い付くために直ちに多くのタスクインスタンスを起動します。これらの事が起きると、パイプラインを作成した際の指定した期間での値よりも高い頻度でパイプラインコンポーネントが実行される様を見る事になるでしょう。AWS Data Pipelineは過去実行数に追い付いた時に、定義された期間にパイプラインを返します。

開発環境またはテストフェーズでの埋め戻し作業を最小化するためには、startDateTime..endDateTimeに比較的短い間隔を設定します。

AWS Data PipelineはパイプラインコンポーネントのscheduledStartTimeが1日前よりも早く設定されている場合、パイプラインの活性化(activation)によって偶発的な埋め戻し作業を防ごうと試みます。この動きを上書きするには、CLI経由で--forceパラメータを使います。

パイプラインをすぐに起動する場合、開始日時を過去1日(以内？)の日付に指定します。AWS Data Pipelineは作業のバックログとして知覚するものを対処するための試みで『延滞(past due)』作業をすぐに実行するために起動を開始します。この埋戻し作業は、AWS Data Pipelineが最初のクラスタ起動を表示するのには待つ時間が無い事を意味します。

スケジュールを使用した最大限のリソース効率化

AWS Data Pipelineを使用すると、リソースやリソースに関連するアクティビティに対する異なるスケジュール期間をサポートする事が出来るのでリソースの効率性を最大化する事が出来ます。

例えば、20分のスケジュール期間でのアクティビティを考えてみましょう。もしアクティビティのリソースが20分スケジュール期間で設定されていた場合、AWS Data Pipelineは1時間で3つのリソースインスタンスを作成し、タスクに必要な3つのリソースを消費します。代わりに、AWS Data Pipelineを使用すると(例えば、1時間のスケジュールに対して)様々なスケジュールとリソースを構成出来るようになります。

20分のスケジュールでアクティビティを組み合わせる場合、AWS Data Pipelineはリソースを1つだけ作成します。1時間でアクティビティの3つ全てのインスタンスにサービスを提供する事で、リソースの使用率を最大化するのです。

タイムゾーン

AWS Data PipelineはデフォルトでUTC/GMTのYYYY-MM-DDTHH:MM:SSの日付時刻フォーマットをサポートしています。例えば、以下はScheduleオブジェクトのstartDateTimeフィールドに1/15/2012, 11:59 p.m.の値をUTC/GMTタイムゾーンで設定した内容になります。

"startDateTime" : "2012-01-15T23:59:00"

式(Expression)を使うと、異なるタイムゾーンを扱うDateTimeオブジェクトを作成する事が出来ます。(アカウント内でサマータイムを扱う場合等も含む)以下記述例はタイムゾーンを使用したものです。

#{inTimeZone(myDateTime,'America/Los_Angeles')}

以下例はDateTimeの値の中に式(Expression)を使用したものです。

"2011-05-24T10:10:00 America/Los_Angeles"

AWS Data PipelineではJoda Time APIを使っています。詳細は以下サイトをご参照ください。

• Joda-Time - Java date and time API - Time Zones

データの上書き防止

AWS Data Pipelineを使い、繰り返しインポートを行う事を考えてみましょう。その処理では1日複数回実行し、実行毎に同じAmazon S3の場所へ出力を行います。この時、日付ベースの式(Expression)を使用しない限り、誤って出力データを上書きしてしまう可能性があります。

S3Output.DirectoryPathに対してs3://myBucket/#{@scheduledStartTime}のような日付ベースの式を指定する事で、各期間での処理毎にディレクトリを指定出来ます。詳細はScheduleをご参照ください。

Resources(リソース)

AWS Data Pipelineではリソースはパイプラインアクティビティが指定する作業を実行する為の計算リソースとして使われます。AWS Data Pipelineは以下のリソースをサポートしています。

リソースがその作業データセットに対する作業は、同じリージョンでも、また異なるリージョンでさえも実行可能です。詳細についてはUsing a Pipeline with Resources in Multiple Regionsをご参照ください。

アクション(Actions)

AWS Data Pipelineのアクションは、成功、失敗、または遅延アクティビティ(？)のような特定のイベントが発生する際にパイプラインコンポーネントが取る事になるステップです。

アクティビティのイベントフィールドはEmrActivityのonLateActionフィールドに於けるTerminateへの参照のような、アクションを指します。AWS Data Pipelineは以下のアクションを提供しています。

役割と権限(Roles and Permissions)

AWS Data Pipelineでは、IAMロールが、パイプラインが何にアクセス出来て、何を実行出来るかを決定します。加えて、パイプラインがEC2インスタンスのようなリソースを作成する際、IAMロールは許可されたリソースとアクションを決定します。

パイプラインを作成する際は、パイプラインを管理するIAMロールと、その他にパイプラインのリソースを管理するIAMロール(リソースロールと呼ばれる)を指定します。これらはどちらも同じ役割を果たします。

パイプラインが作業を実行出来るように、必要最低限のアクセス許可を慎重に検討し、それに応じてIAMロールを定義します。控えめに定義したパイプラインでさえも、リソースやAWSの様々な領域へのアクションへのアクセスが必要となる点は押さえておきましょう。例えば以下の様な操作です。

• Amazon S3のファイルへのアクセス
• Amazon EMRクラスタの作成と管理
• Amazon EC2インスタンスの作成と管理
• Amazon RDSやDynamoDBへのデータアクセス
• Amazon SNSを用いての通知送信

コンソールを使用する場合、AWS Data Pipelineは信頼出来るエンティティのリストを含む、必要となるIAMロールやポリシーを作成します。しかし、同じ作業をCLIやSDKユーザーで行う場合、所定の手順を実行する必要があります。詳細についてはSetting the Required IAM Rolesをご参照ください。

データフォーマット(DataFormat)

AWS DataPipelineで定義されているデータフォーマットは以下の通り。

オプションフィールド(Optional Field)

これまでご紹介してきた各種要素の他にも、AWS Data Pipelineでは『オプションフィールド』(Optional Field)として様々な要素を提供しています。以下ざっくりとご紹介。

Attempt Status

確実なデータ管理を実現する為に、AWS Data Pipelineは失敗した操作についてリトライを試みます。許可された再試行回数のMAXに達するまで設定が可能です。Attemptオブジェクトは様々な試みの内容、結果や(該当する場合は)失敗理由等について追跡を行います。本質的にはそれらは"回数"のインスタンスです。Amazon EMRクラスタとEC2インスタンスのような過去に実行を試みたリソースを使ってAWS Data Pipelineはリトライを遂行します。

...っていう内容がPipeline Components, Instances, and Attemptsに記載としてあったのは確認出来たけど、じゃあこの『Status』にはどのような文字列が入るのか、何を入れておくとどう作用するのか？という所まではいまいち分からず。

要素を展開すると表示されるツールチップや、そのリンクから展開されるEMRアクティビティの該当箇所文言(attemptStatus/The status most recently reported from the object./String →『オブジェクトから通知される最も最近のステータス』)から内容は読み解く事は出来ますが...うーむこの辺はひとまず保留で。

Attempt Timeout

一転してこちらは分かり易いですね。関連する要素の実行に関するタイムアウト設定が行えるようです。

Depends On

要素が他の実行可能な要素に依存する場合、指定します。

Input / Output

実行に入出力要素を伴う場合、各々指定します。

Late After Timeout

オブジェクトの実行を開始しなければならない期間。オブジェクトがスケジュールされた開始時間＋この時間間隔以内に実行を開始しない場合、それは"遅い"とみなされます。(みなされるとどうなる？)

• EmrActivity - AWS Data Pipeline

Max Active Instances

コンポーネントにおける同時実行アクティブインスタンスの最大数。アクティビティにてこの値を"1"に設定すると、インスタンスは厳密に時系列に沿って実行を行います。2以上を設定すると異なるインスタンスを同時に実行する事ができます。(設定の際は、指定したアクティビティが同時実行可能であるかを確認する必要はあるでしょう。)

Maximum Retries

リトライ回数の最大数。

On Success / On Fail / On Late Action

所定の操作が条件に合致した場合に連動させるアクションをそれぞれ指定可能。

• On Success：処理成功時
• On Fail：処理失敗時
• On Late Action：設定したトリガーが実行されず、処理完了しない場合

現状、提供されている実行可能なアクションは以下2つとなっているようです。

• SnsAlarm - AWS Data Pipeline
• Terminate - AWS Data Pipeline

Parent

現在対象となっているオブジェクトが継承する要素。

PreCondition

実行に必要な前提条件。

Report Progress Timeout

ReportTaskProgress APIに対する、タスクランナーからの連続呼び出しの期間。もしタスクランナーやその他タスクを処理するコードが指定した期間内に進捗状況を報告していない場合は、オブジェクト試行をリトライ出来ます。

Retry Delay

リトライする試み(Attempt)間のタイムアウトとする実行間隔を指定。

Runs On

オブジェクトが実行する際の環境を指定。

Schedule Type

オブジェクトが実行を行う際に従う、スケジュールの形式。

Step

実行するクラスタに対する、1つ以上のステップ。2個以上の指定も(255個まで)可能で、JARファイル名の後にカンマ区切りで追記する事で対応が可能となる模様。(例)s3://example-bucket/MyWork.jar,arg1,arg2,arg3

Worker Group

タスクをルーティングする為に使用されます。runsOnの値が設定してあり且つこの値が存在する場合は、この値は無視されます。

式と関数

• Pipeline Expressions and Functions - AWS Data Pipeline
• パイプラインの式と関数 - AWS Data Pipeline

AWS Data Pipelineを扱う上でキーとなるのがこの『式と関数(Pipeline Expressions and Functions)』だと思っています。S3からのデータコピーや各種バッチ処理、SQL処理を行う上で、全ての処理が『全件削除、全件投入』で済むタスクであれば、それらを実現するリソースの組み合わせ、処理構築は比較的楽かも知れません。しかし現実は業務仕様やその環境下で実現可能な対応策というものも限られてしまい、任意の条件で処理を行う必要性が必ず出てくる事と思います。その際にこの『式と関数』を上手く活用する事がキーになると考えます。

このテーマについては別途深堀りして行く必要があると思っていますので、機会を見て進めてみたいと思います。

まとめ

以上、AWS Data Pipelineを構成する要素についてひと通り眺めてみました。ひと通りの要素を1エントリにまとめたところ画面スクロールがクッソ長くなってしまった点については御了承頂けますと幸いです...|ω・)

このエントリを書いていくに当たって、個人的にはどのような要素があるのかという点についてある程度把握出来たような気がします。GUIベースでの処理フロー設計が行えるというのもAWSの他サービスには無い特徴ですし、我々が思い描くアイコンを並べて処理を繋げて...という様なパイプライン作成の際の"コツ"についても、各種オブジェクトの特徴や機能を把握した上で設計を置き換えるような、ある種の『慣れ』が必要になってくるのかな、と関連ドキュメントを読み解き、実業務を置き換えてみよう、となった際に感じた思いでした。

今後は個別のオブジェクトで設定出来る内容、及び実際にどこまで細やかな設定／対応で行いたい処理が組めるか、が気になるところ。その辺りについても今後のエントリで言及して行ければな〜と思います。