JSON 形式でパイプラインを定義するパイプライン定義をアップロードし、アクティブ化する

コマンドラインを使用して CSV データをコピーする

パイプラインを作成し、これを使用して、Amazon S3 バケット間でデータをコピーできます。

前提条件

開始する前に、次のステップを完了しておく必要があります。

コマンドラインインターフェイス (CLI) をインストールして設定します。詳細については、「AWS Data Pipelineへのアクセス」を参照してください。
DataPipelineDefaultRole と DataPipelineDefaultResourceRole という名前の IAM ロールが存在していることを確認します。AWS Data Pipeline コンソールにより、自動的にこれらのロールが作成されます。AWS Data Pipeline コンソールをまだ 1 回も使用したことがない場合、これらのロールを手動で作成する必要があります。詳細については、「AWS Data Pipeline の IAM ロール」を参照してください。

JSON 形式でパイプラインを定義する

この例のシナリオでは、JSON パイプライン定義と AWS Data Pipeline CLI を使用して、2 つの Amazon S3 バケット間でデータを特定の間隔でコピーするようにスケジュールする方法を説明します。これは、完全なパイプライン定義の JSON ファイルであり、その後に各セクションの説明を示します。

注記

JSON 形式のファイルの構文を検証できるテキストエディタを使用し、.json ファイル拡張子を使用してファイルに名前を付けることをお勧めします。

この例では、わかりやすくするために、オプションフィールドを省略し、必須フィールドのみを示しています。この例の完全なパイプラインの JSON ファイルは次のようになります。


{
  "objects": [
    {
      "id": "MySchedule",
      "type": "Schedule",
      "startDateTime": "2013-08-18T00:00:00",
      "endDateTime": "2013-08-19T00:00:00",
      "period": "1 day"
    },
    {
      "id": "S3Input",
      "type": "S3DataNode",
      "schedule": {
        "ref": "MySchedule"
      },
      "filePath": "s3://example-bucket/source/inputfile.csv"
    },
    {
      "id": "S3Output",
      "type": "S3DataNode",
      "schedule": {
        "ref": "MySchedule"
      },
      "filePath": "s3://example-bucket/destination/outputfile.csv"
    },
    {
      "id": "MyEC2Resource",
      "type": "Ec2Resource",
      "schedule": {
        "ref": "MySchedule"
      },
      "instanceType": "m1.medium",
      "role": "DataPipelineDefaultRole",
      "resourceRole": "DataPipelineDefaultResourceRole"
    },
    {
      "id": "MyCopyActivity",
      "type": "CopyActivity",
      "runsOn": {
        "ref": "MyEC2Resource"
      },
      "input": {
        "ref": "S3Input"
      },
      "output": {
        "ref": "S3Output"
      },
      "schedule": {
        "ref": "MySchedule"
      }
    }
  ]
}

スケジュール

パイプラインでは、このパイプラインのアクティビティを実行する頻度を決定する期間と共に、開始日と終了日を持つスケジュールを定義します。


{
  "id": "MySchedule",
  "type": "Schedule",
  "startDateTime": "2013-08-18T00:00:00",
  "endDateTime": "2013-08-19T00:00:00",
  "period": "1 day"
},

Amazon S3 データノード

次に、入力 S3DataNode パイプラインコンポーネントによって、入力ファイルの場所を定義します。この例では、Amazon S3 バケットの場所です。入力 S3DataNode コンポーネントは、次のフィールドで定義されます。


{
  "id": "S3Input",
  "type": "S3DataNode",
  "schedule": {
    "ref": "MySchedule"
  },
  "filePath": "s3://example-bucket/source/inputfile.csv"
},

ID: 入力場所のユーザー定義名（参照字にのみ使用されるラベル）。
タイプ: パイプラインコンポーネントの型。Amazon S3 バケット内のデータが存在する場所と一致する "S3DataNode" です。
スケジュール: JSON ファイルの先行部分で作成したスケジュールコンポーネント "MySchedule" 参照します。
[Path] (パス): データノードに関連付けられたデータへのパス。データノードの構文はそのタイプによって決まります。例えば、Amazon S3 パスの構文は、データベーステーブルに適した構文とは別の構文に従います。

次に、出力 S3DataNode コンポーネントで、データの出力先の場所を定義します。これは、入力 S3DataNode コンポーネントと同じ形式に従います。ただし、コンポーネントの名前とターゲットファイルを指定するパスは異なります。


{
  "id": "S3Output",
  "type": "S3DataNode",
  "schedule": {
    "ref": "MySchedule"
  },
  "filePath": "s3://example-bucket/destination/outputfile.csv"
},

リソース

これは、コピー操作を実行するコンピューティングリソースの定義です。この例では、AWS Data Pipeline は、コピータスクを実行するための EC2 インスタンスを自動的に作成し、コピータスクが完了するとリソースを終了します。ここで定義されているフィールドが、作業を行う EC2 インスタンスの作成と機能を制御します。EC2Resource は、次のフィールドで定義されます。


{
  "id": "MyEC2Resource",
  "type": "Ec2Resource",
  "schedule": {
    "ref": "MySchedule"
  },
  "instanceType": "m1.medium",
  "role": "DataPipelineDefaultRole",
  "resourceRole": "DataPipelineDefaultResourceRole"
},

ID: パイプラインスケジュールのユーザー定義の名前。これは参照時にのみ使用されるラベルです。
タイプ: 作業を実行するコンピューティングリソースの種類。この例では、 EC2 インスタンス。EmrCluster タイプなど、その他のリソースタイプも使用できます。
スケジュール: このコンピューティングリソースを作成するスケジュール。
instanceType: 作成する EC2 インスタンスのサイズ。AWS Data Pipeline で実行する作業の負荷に最適な EC2 インスタンスのサイズを設定します。ここでは、m1.medium の EC2 インスタンスを設定します。さまざまなインスタンスタイプとそれぞれの用途の詳細については、Amazon EC2 インスタンスタイプに関するトピック (http://aws.amazon.com/ec2/instance-types/) を参照してください。
ロール: リソースにアクセスするアカウントの IAM ロール (データを取得するための Amazon S3 バケットへのアクセスなど)。
resourceRole: リソースを作成するアカウントの IAM ロール（お客様に代わって EC2 インスタンスを作成および設定するなど）。Role と ResourceRole は同じロールにすることもできますが、個別に設定することによって、セキュリティ設定での詳細度が向上します。

アクティビティ

この JSON ファイルの最後のセクションは、実行する作業を表すアクティビティの定義です。この例では、CopyActivity を使用して、http://aws.amazon.com/ec2/instance-types/ のバケット間で CSV ファイルのデータをコピーします。CopyActivity コンポーネントは、以下のフィールドで定義されます。


{
  "id": "MyCopyActivity",
  "type": "CopyActivity",
  "runsOn": {
    "ref": "MyEC2Resource"
  },
  "input": {
    "ref": "S3Input"
  },
  "output": {
    "ref": "S3Output"
  },
  "schedule": {
    "ref": "MySchedule"
  }
}

ID: アクティビティのユーザー定義の名前。これは参照時にのみ使用されるラベルです。
タイプ: MyCopyActivity など、実行するアクティビティのタイプ。
runsOn: このアクティビティが定義する作業を実行するコンピューティングリソース。この例では、先に定義した EC2 インスタンスへの参照を指定します。runsOn フィールドを使用すると、AWS Data Pipeline が EC2 インスタンスを自動的に作成します。runsOn フィールドは、リソースが AWS インフラストラクチャに存在することを示し、一方、workerGroup 値は、独自のオンプレミスリソースを使用して作業を実行することを示しています。
入力: コピーするデータの場所。
出力: ターゲットデータの場所。
スケジュール: このアクティビティを実行するスケジュール。

パイプライン定義をアップロードし、アクティブ化する

パイプライン定義をアップロードし、パイプラインをアクティブ化する必要があります。以下のコマンド例では、pipeline_name をパイプラインのラベルに置き換え、pipeline_file をパイプライン定義 .json ファイルの完全修飾パスに置き換えます。

AWS CLI

パイプライン定義を作成してパイプラインをアクティブ化するには、以下の create-pipeline コマンドを使用します。パイプラインの ID をメモします。この値は、ほとんどの CLI コマンドで使用するからです。


aws datapipeline create-pipeline --name pipeline_name --unique-id token
{
    "pipelineId": "df-00627471SOVYZEXAMPLE"
}

パイプライン定義を更新するには、以下の put-pipeline-definition コマンドを使用します。


aws datapipeline put-pipeline-definition --pipeline-id df-00627471SOVYZEXAMPLE --pipeline-definition file://MyEmrPipelineDefinition.json

パイプラインが正常に検証された場合、validationErrors フィールドは空です。警告を確認する必要があります。

パイプラインをアクティブ化するには、以下の activate-pipeline コマンドを使用します。


aws datapipeline activate-pipeline --pipeline-id df-00627471SOVYZEXAMPLE

以下の list-pipelines コマンドを使用して、パイプラインリストにパイプラインが表示されていることを確認できます。


aws datapipeline list-pipelines

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

開始する前に

Amazon S3 への MySQL データのエクスポート