AWS Data Pipeline ya no está disponible para nuevos clientes. Los clientes actuales de AWS Data Pipeline pueden seguir utilizando el servicio con normalidad. Más información

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Copia de datos de MySQL mediante la línea de comandos

Cree una canalización para copiar datos de una tabla de MySQL en un archivo en un bucket de Amazon S3.

Definir una canalización en formato JSON

Este escenario de ejemplo muestra cómo utilizar definiciones de la canalización de JSON y la CLI de AWS Data Pipeline para copiar datos (filas) de una tabla en una base de datos MySQL en un archivo CSV (valores separados por comas) en un bucket de Amazon S3 en un intervalo de tiempo especificado.

Este es el archivo JSON de definición de la canalización completo seguido de una explicación de cada una de sus secciones.

{
  "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"
    }
  ]
}

Nodo de datos MySQL

El componente de entrada MySqlDataNode de la canalización define una ubicación para los datos de entrada; en este caso, una instancia de Amazon RDS. El componente MySqlDataNode de la entrada se define por medio de los siguientes campos:

{
  "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"
},

Nodo de datos Amazon S3

A continuación, el componente de canalización S3Output define una ubicación para el archivo de salida; en este caso, un archivo CSV en una ubicación de bucket de Amazon S3. El componente S3DataNode de salida se define por los siguientes campos:

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

Recurso

Esta es una definición del recurso informático que realiza la operación de copia. En este ejemplo, AWS Data Pipeline debe crear automáticamente una instancia EC2 para realizar la tarea de copia y terminar el recurso tras completarse la tarea. Los campos definidos aquí controlan la creación y función de la instancia EC2 que realiza el trabajo. El componente Resource se define por los siguientes campos:

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

Actividad

La última sección del archivo JSON es la definición de la actividad que representa el trabajo que se realizará. En este caso, utilizamos un componente CopyActivity para copiar datos de un archivo en un bucket de Amazon S3 en otro archivo. El componente CopyActivity se define por los siguientes campos:

{
  "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"
},                         

Cargar y activar la definición de canalización

Debe cargar la definición de su canalización y activarla. En los siguientes comandos de ejemplo, reemplace pipeline_name por una etiqueta para su canalización y pipeline_file por la ruta completa para el archivo de definición de canalización..json

AWS CLI

Para crear su definición de canalización y activarla, use el siguiente comando create-pipeline. Anote el ID de su canalización, ya que utilizará este valor con la mayoría de los comandos de la CLI.

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

Para cargar su definición de la canalización, utilice el comando siguiente: put-pipeline-definition.

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

Si la canalización se valida correctamente, el campo validationErrors estará vacío. Debe revisar todas las advertencias.

Para activar la canalización, utilice el siguiente comando activate-pipeline:

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

Puede comprobar que su canalización aparece en la lista de canalizaciones mediante el siguiente comando list-pipelines.

aws datapipeline list-pipelines