AWS Data Pipeline

解決する課題

データを定期的にある場所から別の場所へ移し、途中で変換する「バッチ処理のワークフロー」を自前で組むと、スケジュール実行、ジョブ間の依存関係、失敗時のリトライ、計算リソースの確保といった作業を個別に作り込む必要があります。AWS Data Pipeline はこれらを1つのパイプライン定義に集約します。

• データの移動・変換の処理をスケジュールに従って定期実行する
• 処理どうしの依存関係を定義し、前段が完了してから後段を走らせる
• 失敗時のリトライや通知を組み込み、一時的な障害を吸収する
• 処理に必要な EC2 や EMR クラスタを自動で起動・終了し、計算リソースを用意する

S3・RDS・DynamoDB・Redshift・EMR の間で定期的にデータを動かすバッチ、ログの集計、DynamoDB のバックアップといった用途で使われてきました。なお現在は新規採用が推奨されず、後継サービスへの移行が基本である点に注意が必要です。

主要概念と用語

• パイプライン（Pipeline）: データ処理ワークフロー全体を表す入れ物。複数のアクティビティと依存関係をまとめて定義する
• パイプライン定義（Definition）: パイプラインの構成を JSON 形式で記述したもの。各構成要素とその関係を宣言的に書く
• アクティビティ（Activity）: 実際の処理単位。データのコピー、SQL の実行、EMR ジョブ、シェルコマンド実行などの種類がある
• データノード（DataNode）: 入力元・出力先のデータの場所。S3、DynamoDB テーブル、RDS/Redshift のデータベースなどを指す
• 前提条件（Precondition）: アクティビティを始める前に満たすべき条件。対象データの存在確認などで、満たされるまで待機・再評価する
• リソース（Resource）: アクティビティを実行する計算基盤。EC2 インスタンスや EMR クラスタを必要なときに自動で起動・終了する
• スケジュール（Schedule）: パイプラインの実行間隔や開始・終了時刻を定義する
• Task Runner: リソース上で動き、割り当てられたタスクを取得して実行するエージェント。自前のオンプレ機にも導入できる

仕様・制限・クォータ

• パイプラインは JSON のパイプライン定義で表現し、コンソール・CLI・SDK から作成・有効化する
• 入出力には S3、DynamoDB、RDS、Redshift などのデータノードを使い、オンプレミスのデータソースも Task Runner 経由で扱える
• 処理に使う EC2 / EMR リソースをマネージドに起動・終了でき、必要時だけ計算資源を確保する
• アカウントあたりのパイプライン数や、1パイプラインのオブジェクト数などにサービスクォータがあり、必要に応じて引き上げを申請する
• リージョンによっては提供されない場合があり、新規パイプライン作成が制限される環境もある
• 具体的な上限値・単価は変動するため、利用時に公式の最新情報を確認する

内部の仕組み

Data Pipeline は「スケジューラ」と「実行エージェント」を分離した構成で動きます。

• ユーザーが有効化したパイプライン定義をマネージドなサービスが解釈し、スケジュールと依存関係に従って実行すべきタスクを生成する
• 各アクティビティは前提条件が満たされたかを評価し、満たされて初めて実行対象になる
• 実行は Task Runner が担う。Data Pipeline がマネージドに起動した EC2/EMR 上で動くほか、ユーザーが自前のサーバーに Task Runner を導入して実行させることもできる
• タスクが失敗すると、定義したリトライ回数まで再実行し、最終的に失敗すれば設定に応じて SNS 通知などを発火する
• 計算リソースはアクティビティのために起動され、処理が終わると自動的に終了するため、待機中のインスタンス費用を抑えられる

設計パターン / ベストプラクティス

• 依存関係を明示する: アクティビティ間の前後関係をパイプライン定義に書き、順序の取り違えや競合を防ぐ
• 前提条件を活用する: 入力データの存在を確認してから処理を開始し、空振りや不完全な入力での実行を避ける
• リトライと通知を設定する: 一時障害はリトライで吸収し、最終的な失敗は SNS で運用者へ通知する
• オンデマンドにリソースを使う: 処理に必要なときだけ EC2/EMR を起動・終了させ、待機コストを抑える
• 冪等な処理にする: リトライや再実行で同じデータを二重処理しても結果が壊れないよう、出力を上書きや分割で設計する
• 後継への移行を見据える: 新規ワークフローは Glue・Step Functions・MWAA で組み、既存パイプラインも段階的に移す

運用・監視

• パイプラインの実行状況・各アクティビティの成否はコンソールや API でステータスとして確認する
• 失敗時は SNS で通知を飛ばし、運用者が早期に検知できるようにする
• リトライ回数やタイムアウトを設定し、一時障害を吸収しつつ無限の再試行を防ぐ
• Task Runner やリソース上の処理ログを残し、失敗時の原因調査に使う
• API 操作の監査は CloudTrail で行う
• メンテナンス段階のサービスのため、長期運用では後継サービスへの移行計画を運用タスクに含める

コスト

• 課金は主にパイプライン中のアクティビティ実行頻度（実行が低頻度か高頻度か）に基づく従量制で、有効でも非アクティブなパイプラインには別の料金区分がある
• パイプラインが起動する EC2 や EMR の利用料は、それぞれのサービス料金として別途発生する
• 処理を必要なときだけ動かし、リソースを使い終わったら終了させるほど、無駄なインスタンス費用を抑えられる
• 変動する具体的な単価は公式の料金ページで確認する

セキュリティ

• アクセス制御は IAM で行い、パイプラインの操作権限と、リソースが処理対象にアクセスするためのロールを最小権限で分けて付与する
• データノードへのアクセスは、対象サービス（S3・DynamoDB・RDS など）側の暗号化と権限設定に従う
• 通信は TLS で保護し、VPC 内のリソースへはネットワーク設定で到達経路を絞る
• データベース接続などの認証情報は直書きせず、安全な方法で受け渡す
• API 操作は CloudTrail で記録し、誰がいつパイプラインを変更したかを追跡する

関連サービス・比較

サーバーレスな ETL とデータ統合を担う AWS Glue とよく比較されます。新規のデータ変換ワークフローでは、運用負荷の低い Glue やワークフロー制御に優れた Step Functions が推奨され、Data Pipeline は既存資産の維持・移行対象という位置づけです。

ハンズオン / CLI例

# 空のパイプラインを作成し、パイプラインIDを取得
aws datapipeline create-pipeline \
  --name my-batch-pipeline \
  --unique-id my-batch-pipeline-token

# JSONで書いたパイプライン定義を登録（定義の構文を検証）
aws datapipeline put-pipeline-definition \
  --pipeline-id df-EXAMPLE1234567890 \
  --pipeline-definition file://pipeline-def.json

# パイプラインを有効化してスケジュール実行を開始
aws datapipeline activate-pipeline \
  --pipeline-id df-EXAMPLE1234567890

# 実行状況（各オブジェクトのステータス）を確認
aws datapipeline list-pipelines