Definir um pipeline no formato JSON Fazer upload e ativar a definição do pipeline

Copiar dados do MySQL usando a linha de comando

Você pode criar um pipeline para copiar dados de uma tabela do MySQL para um arquivo em um bucket do Amazon S3.

Pré-requisitos

Antes de começar, é necessário concluir as seguintes etapas:

Instale e configure a Interface da linha de comando (CLI). Para obter mais informações, consulte Acessando AWS Data Pipeline.
Certifique-se de que os perfis do IAM chamados DataPipelineDefaultRole e DataPipelineDefaultResourceRole existem. O console do AWS Data Pipeline cria essas funções para você automaticamente. Se você não usou o console do AWS Data Pipeline pelo menos uma vez, deverá criar essas funções manualmente. Para obter mais informações, consulte Funções do IAM para o AWS Data Pipeline.
Configure um bucket do Amazon S3 e uma instância do Amazon RDS. Para obter mais informações, consulte Antes de começar.

Tarefas

Definir um pipeline no formato JSON
Fazer upload e ativar a definição do pipeline

Definir um pipeline no formato JSON

Este cenário de exemplo mostra como usar as definições de pipeline JSON e a CLI do AWS Data Pipeline para copiar dados (linhas) de uma tabela em um banco de dados MySQL para um arquivo CSV (valores separados por vírgulas) de um bucket do Amazon S3 em um intervalo especificado.

Este é o arquivo JSON de definição de pipeline completo, seguido de uma explicação para cada uma das seções.

nota

Recomendamos que você use um editor de texto que possa ajudá-lo a verificar a sintaxe dos arquivos formatados com JSON e nomeie o arquivo usando a extensão de arquivo .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"
    }
  ]
}

Nó de dados do MySQL

O componente de pipeline MySqlDataNode de entrada define um local para os dados de entrada. Nesse caso, uma instância do Amazon RDS. O componente MySqlDataNode de entrada é definido pelos seguintes 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"
},

Id: O nome definido pelo usuário, que é apenas um rótulo para sua referência.
Nome de usuário: O nome de usuário da conta do banco de dados que tem permissão suficiente para recuperar dados da tabela do banco de dados. Substitua my-username pelo nome do seu usuário.
Schedule: Uma referência para o componente de programação que criamos nas linhas anteriores do arquivo JSON.
Nome: O nome definido pelo usuário, que é apenas um rótulo para sua referência.
*Password: A senha da conta do banco de dados com o prefixo de asterisco para indicar que o AWS Data Pipeline precisa criptografar o valor da senha. Substitua my-password pela senha correta do seu usuário. O campo de senha é precedido pelo caractere especial asterisco. Para obter mais informações, consulte Caracteres especiais.
Tabela: O nome da tabela do banco de dados que contém os dados a serem copiados. Substitua table-name pelo nome da tabela do seu banco de dados.
connectionString: A string de conexão JDBC para o objeto CopyActivity que se conectará ao banco de dados.
selectQuery: Uma consulta SQL SELECT válida que especifica quais dados da tabela do banco de dados serão copiados. #{table} é uma expressão que reutiliza o nome da tabela fornecido pela variável "table" nas linhas que precedem o arquivo JSON.
Tipo: O tipo SqlDataNode, que é uma instância do Amazon RDS usando o MySQL neste exemplo.
nota
O tipo MySqlDataNode está obsoleto. Você ainda pode usar MySqlDataNode. No entanto, recomendamos o uso de SqlDataNode.

Nó de dados do Amazon S3

Em seguida, o componente de pipeline S3Output definirá um local para o arquivo de saída. Nesse caso, um arquivo CSV no local de um bucket do Amazon S3. O componente S3DataNode de saída é definido pelos seguintes campos:


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

Id: O ID definido pelo usuário, que é apenas um rótulo para sua referência.
Schedule: Uma referência para o componente de programação que criamos nas linhas anteriores do arquivo JSON.
filePath: O caminho para os dados associados ao nó de dados, que é um arquivo de saída CSV neste exemplo.
Nome: O nome definido pelo usuário, que é apenas um rótulo para sua referência.
Tipo: O tipo de objeto do pipeline, que é S3DataNode, para corresponder com o local em que os dados residem, em um bucket do Amazon S3.

Recurso

Esta é uma definição do recurso computacional que executa a operação de cópia. Neste exemplo, o AWS Data Pipeline deve criar automaticamente uma instância do EC2 para executar a tarefa de cópia e encerrar o recurso após a conclusão da tarefa. Os campos definidos aqui controlam a criação e a função da instância do EC2 que faz o trabalho. O EC2Resource é definido pelos seguintes campos:


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

Id: O ID definido pelo usuário, que é apenas um rótulo para sua referência.
Schedule: A programação para criar este recurso computacional.
Nome: O nome definido pelo usuário, que é apenas um rótulo para sua referência.
Função: O perfil do IAM da conta que acessa os recursos, como acesso ao bucket do Amazon S3 para recuperação de dados.
Tipo: O tipo de recurso computacional para executar o trabalho. Nesse caso, uma instância do EC2. Existem outros tipos de recursos disponíveis, por exemplo, o tipo EmrCluster.
resourceRole: A função do IAM da conta que cria recursos, como criação e configuração de uma instância do EC2 em seu nome. Role e ResourceRole podem ser a mesma função, mas fornecem separadamente uma granularidade maior na sua configuração de segurança.

Atividade

A última seção no arquivo JSON é a definição da atividade que representa o trabalho a ser executado. Neste caso, usamos um componente CopyActivity para copiar dados de um arquivo em um bucket do Amazon S3 para outro arquivo. O componente CopyActivity é definido pelos seguintes 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"
},

Id: O ID definido pelo usuário, que é apenas um rótulo para sua referência
Entrada: O local dos dados do MySQL a serem copiados
Schedule: A programação na qual esta atividade será executada
Nome: O nome definido pelo usuário, que é apenas um rótulo para sua referência
runsOn: O recurso computacional que realiza o trabalho definido por essa atividade. Neste exemplo, fornecemos uma referência à instância do EC2 anteriormente definida. Usar o campo runsOn faz com que o AWS Data Pipeline crie a instância do EC2 para você. O campo runsOn indica que o recurso existe na infraestrutura da AWS, enquanto o valor workerGroup indica que você deseja usar seus próprios recursos locais para executar o trabalho.
onSuccess: SnsAlarm a ser enviado se a atividade for concluída com sucesso
onFail: SnsAlarm a ser enviado se a atividade falhar
Resultado: Local do arquivo CSV de saída no Amazon S3
Tipo: O tipo da atividade a ser executada.

Fazer upload e ativar a definição do pipeline

Você deve fazer o upload da definição do pipeline e ativá-lo. Nos comandos de exemplo a seguir, substitua pipeline_name por um rótulo para seu pipeline e pipeline_file pelo caminho totalmente qualificado para o arquivo .json de definição de pipeline.

AWS CLI

Para criar sua definição de pipeline e ativar seu pipeline, use o seguinte comando: create-pipeline. Observe a ID do seu pipeline, pois você usará esse valor com a maioria dos comandos da CLI.


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

Atualize a definição do pipeline usando o seguinte comando: put-pipeline-definition.


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

Se o pipeline for validado com êxito, o campo validationErrors estará vazio. Você deve revisar todos os avisos.

Para ativar o pipeline, use o seguinte comando: activate-pipeline.


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

Você pode verificar se seu pipeline aparece na lista de pipeline usando o seguinte comando: list-pipelines.


aws datapipeline list-pipelines

Atenção O Javascript está desativado ou não está disponível no seu navegador.

Para usar a documentação da AWS, o Javascript deve estar ativado. Consulte as páginas de Ajuda do navegador para obter instruções.

Convenções do documento

Antes de começar

Copiar dados para o Amazon Redshift