コマンドラインを使用して MySQL データをコピーする - AWS Data Pipeline
JSON 形式でパイプラインを定義するパイプライン定義をアップロードし、アクティブ化する

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

MySQL のテーブルから Amazon S3 バケットのファイルにデータをコピーするパイプラインを作成することができます。

前提条件

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

  1. コマンドラインインターフェイス (CLI) をインストールして設定します。詳細については、「AWS Data Pipelineへのアクセス」を参照してください。

  2. DataPipelineDefaultRole と DataPipelineDefaultResourceRole という名前の IAM ロールが存在していることを確認します。AWS Data Pipeline コンソールにより、自動的にこれらのロールが作成されます。AWS Data Pipeline コンソールをまだ 1 回も使用したことがない場合、これらのロールを手動で作成する必要があります。詳細については、「AWS Data Pipeline の IAM ロール」を参照してください。

  3. Amazon S3 バケットと Amazon RDS インスタンスを設定します。詳細については、「開始する前に」を参照してください。

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

この例では、JSON パイプライン定義と AWS Data Pipeline CLI を使用して、指定した時間間隔で MySQL データベーステーブルのデータ (行) を Amazon S3 バケットの CSV (Comma Separated Value) ファイルにコピーする方法を示します。

これは、完全なパイプライン定義の JSON ファイルであり、その後に各セクションの説明を示します。

注記

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

{
  "objects": [
    {
      "id": "ScheduleId113",
      "startDateTime": "2013-08-26T00:00:00",
      "name": "My Copy Schedule",
      "type": "Schedule",
      "period": "1 Days"
    },
    {
      "id": "CopyActivityId112",
      "input": {
        "ref": "MySqlDataNodeId115"
      },
      "schedule": {
        "ref": "ScheduleId113"
      },
      "name": "My Copy",
      "runsOn": {
        "ref": "Ec2ResourceId116"
      },
      "onSuccess": {
        "ref": "ActionId1"
      },
      "onFail": {
        "ref": "SnsAlarmId117"
      },
      "output": {
        "ref": "S3DataNodeId114"
      },
      "type": "CopyActivity"
    },
    {
      "id": "S3DataNodeId114",
      "schedule": {
        "ref": "ScheduleId113"
      },
      "filePath": "s3://example-bucket/rds-output/output.csv",
      "name": "My S3 Data",
      "type": "S3DataNode"
    },
    {
      "id": "MySqlDataNodeId115",
      "username": "my-username",
      "schedule": {
        "ref": "ScheduleId113"
      },
      "name": "My RDS Data",
      "*password": "my-password",
      "table": "table-name",
      "connectionString": "jdbc:mysql://your-sql-instance-name.id.region-name.rds.amazonaws.com:3306/database-name",
      "selectQuery": "select * from #{table}",
      "type": "SqlDataNode"
    },
    {
      "id": "Ec2ResourceId116",
      "schedule": {
        "ref": "ScheduleId113"
      },
      "name": "My EC2 Resource",
      "role": "DataPipelineDefaultRole",
      "type": "Ec2Resource",
      "resourceRole": "DataPipelineDefaultResourceRole"
    },
    {
      "message": "This is a success message.",
      "id": "ActionId1",
      "subject": "RDS to S3 copy succeeded!",
      "name": "My Success Alarm",
      "role": "DataPipelineDefaultRole",
      "topicArn": "arn:aws:sns:us-east-1:123456789012:example-topic",
      "type": "SnsAlarm"
    },
    {
      "id": "Default",
      "scheduleType": "timeseries",
      "failureAndRerunMode": "CASCADE",
      "name": "Default",
      "role": "DataPipelineDefaultRole",
      "resourceRole": "DataPipelineDefaultResourceRole"
    },
    {
      "message": "There was a problem executing #{node.name} at for period #{node.@scheduledStartTime} to #{node.@scheduledEndTime}",
      "id": "SnsAlarmId117",
      "subject": "RDS to S3 copy failed",
      "name": "My Failure Alarm",
      "role": "DataPipelineDefaultRole",
      "topicArn": "arn:aws:sns:us-east-1:123456789012:example-topic",
      "type": "SnsAlarm"
    }
  ]
}

MySQL データノード

入力 MySqlDataNode パイプラインコンポーネントは、入力データの場所を定義します。この例では、Amazon RDS インスタンスです。入力 MySqlDataNode コンポーネントは、次のフィールドで定義されます。

{
  "id": "MySqlDataNodeId115",
  "username": "my-username",
  "schedule": {
    "ref": "ScheduleId113"
  },
  "name": "My RDS Data",
  "*password": "my-password",
  "table": "table-name",
  "connectionString": "jdbc:mysql://your-sql-instance-name.id.region-name.rds.amazonaws.com:3306/database-name",
  "selectQuery": "select * from #{table}",
  "type": "SqlDataNode"
},
ID

ユーザー定義の名前。これは参照時にのみ使用されるラベルです。

ユーザーネーム

データベーステーブルからデータを取得するのに十分なアクセス許可を持つデータベースアカウントのユーザー名。my-username をユーザーアカウントの名前に置き換えます。

スケジュール

JSON ファイルの先行部分で作成したスケジュールコンポーネントを参照します。

名前

ユーザー定義の名前。これは参照時にのみ使用されるラベルです。

*Password

データベースアカウントのパスワード。先頭のアスタリスク（*）は、AWS Data Pipeline がパスワード値を暗号化する必要があることを示します。my-passwordをユーザーの正しいパスワードに置き換えます。パスワードフィールド名の先頭には、特殊文字のアスタリスク（*）が付きます。詳細については、「特殊文字」を参照してください。

テーブル

コピーするデータを含むデータベーステーブルの名前。table-name をデータベーステーブルの名前に置き換えます。

connectionString

データベースに接続する CopyActivity オブジェクト用の JDBC 接続文字列。

selectQuery

データベーステーブルからコピーするデータを指定する有効な SQL の SELECT クエリ。#{table} は、JSON ファイルの先行部分の table 変数によって指定されたテーブル名を再利用する式です。

タイプ

SqlDataNode 型。この例の場合、これは MySQL を使用する Amazon RDS インスタンスです。

注記

MySqlDataNode 型は廃止されました。現時点では MySqlDataNode も使用できますが、SqlDataNode の使用をお勧めします。

Amazon S3 データノード

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

{
  "id": "S3DataNodeId114",
  "schedule": {
    "ref": "ScheduleId113"
  },
  "filePath": "s3://example-bucket/rds-output/output.csv",
  "name": "My S3 Data",
  "type": "S3DataNode"
},
ID

ユーザー定義の ID。これは参照時にのみ使用されるラベルです。

スケジュール

JSON ファイルの先行部分で作成したスケジュールコンポーネントを参照します。

filePath

データノードに関連付けられているデータへのパス。この例では CSV 出力ファイルです。

名前

ユーザー定義の名前。これは参照時にのみ使用されるラベルです。

タイプ

パイプラインオブジェクトの型。Amazon S3 バケット内のデータが存在する場所と一致する S3DataNode です。

リソース

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

{
  "id": "Ec2ResourceId116",
  "schedule": {
    "ref": "ScheduleId113"
  },
  "name": "My EC2 Resource",
  "role": "DataPipelineDefaultRole",
  "type": "Ec2Resource",
  "resourceRole": "DataPipelineDefaultResourceRole"
},
ID

ユーザー定義の ID。これは参照時にのみ使用されるラベルです。

スケジュール

このコンピューティングリソースを作成するスケジュール。

名前

ユーザー定義の名前。これは参照時にのみ使用されるラベルです。

ロール

リソースにアクセスするアカウントの IAM ロール (データを取得するための Amazon S3 バケットへのアクセスなど)。

タイプ

作業を実行するコンピューティングリソースの種類。この例では、 EC2 インスタンス。EmrCluster タイプなど、その他のリソースタイプも使用できます。

resourceRole

リソースを作成するアカウントの IAM ロール（お客様に代わって EC2 インスタンスを作成および設定するなど）。Role と ResourceRole は同じロールにすることもできますが、個別に設定することによって、セキュリティ設定での詳細度が向上します。

アクティビティ

この JSON ファイルの最後のセクションは、実行する作業を表すアクティビティの定義です。この例では、CopyActivity コンポーネントを使用して、Amazon S3 バケットのファイルから別のファイルにデータをコピーします。CopyActivity コンポーネントは、次のフィールドで定義されます。

{
  "id": "CopyActivityId112",
  "input": {
    "ref": "MySqlDataNodeId115"
  },
  "schedule": {
    "ref": "ScheduleId113"
  },
  "name": "My Copy",
  "runsOn": {
    "ref": "Ec2ResourceId116"
  },
  "onSuccess": {
    "ref": "ActionId1"
  },
  "onFail": {
    "ref": "SnsAlarmId117"
  },
  "output": {
    "ref": "S3DataNodeId114"
  },
  "type": "CopyActivity"
},
ID

ユーザー定義の ID。これは参照時にのみ使用されるラベルです。

入力

コピーする MySQL データの場所。

スケジュール

このアクティビティを実行するスケジュール。

名前

ユーザー定義の名前。これは参照時にのみ使用されるラベルです。

runsOn

このアクティビティが定義する作業を実行するコンピューティングリソース。この例では、先に定義した EC2 インスタンスへの参照を指定します。runsOn フィールドを使用すると、AWS Data Pipeline が EC2 インスタンスを自動的に作成します。runsOn フィールドは、リソースが AWS インフラストラクチャに存在することを示し、一方、workerGroup 値は、独自のオンプレミスリソースを使用して作業を実行することを示しています。

onSuccess

アクティビティが正常終了した場合に送信する SnsAlarm。

onFail

アクティビティが失敗した場合に送信する SnsAlarm。

出力

CSV 出力ファイルの Amazon S3 での場所。

タイプ

実行するアクティビティのタイプ。

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

パイプライン定義をアップロードし、パイプラインをアクティブ化する必要があります。以下のコマンド例では、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