Definire una pipeline in formato JSON Caricamento e attivazione della definizione della pipeline

Copia dei dati su MySQL tramite la riga di comando

Puoi creare una pipeline per copiare dati da una tabella MySQL a un file in un bucket Amazon S3.

Prerequisiti

Prima di iniziare , devi completare le fasi seguenti:

Installa e configura un'interfaccia a riga di comando (CLI). Per ulteriori informazioni, consulta Accedere AWS Data Pipeline.
Assicurati che i ruoli IAM siano denominati DataPipelineDefaultRoleed DataPipelineDefaultResourceRoleesistano. La AWS Data Pipeline console crea questi ruoli automaticamente. Se non hai usato la AWS Data Pipeline console almeno una volta, devi creare questi ruoli manualmente. Per ulteriori informazioni, consulta Ruoli IAM per AWS Data Pipeline.
Configura un bucket Amazon S3 e un'istanza Amazon RDS. Per ulteriori informazioni, consulta Prima di iniziare.

Processi

Definire una pipeline in formato JSON
Caricamento e attivazione della definizione della pipeline

Definire una pipeline in formato JSON

Questo scenario di esempio mostra come utilizzare le definizioni della pipeline JSON e l'interfaccia a AWS Data Pipeline riga di comando per copiare dati (righe) da una tabella in un database MySQL a un file CSV (valori separati da virgole) in un bucket Amazon S3 a un intervallo di tempo specificato.

Questo è il file JSON completo di definizione della pipeline seguito da una spiegazione per ciascuna delle sue sezioni.

Nota

È consigliabile utilizzare un editor di testo che può aiutare a verificare la sintassi di file in formato JSON e nominare il file utilizzando l'estensione del file .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"
    }
  ]
}

Nodo di dati MySQL

Il componente della MySqlDataNode pipeline di input definisce una posizione per i dati di input; in questo caso, un'istanza Amazon RDS. Il MySqlDataNode componente di input è definito dai seguenti campi:


{
  "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: Il nome definito dall'utente, un'etichetta solo di riferimento.
Username: Il nome utente dell'account di database che dispone di autorizzazioni sufficienti per recuperare i dati dalla tabella di database. Sostituisci my-username con il nome del tuo utente.
Pianificazione: Un riferimento al componente di pianificazione che abbiamo creato nelle righe precedenti del file JSON.
Nome: Il nome definito dall'utente, un'etichetta solo di riferimento.
*Password: La password per l'account di database con il prefisso asterisco per confermare che AWS Data Pipeline deve crittografare il valore della password. Sostituisci my-password con la password corretta per il tuo utente. Il campo password è preceduto dal carattere speciale dell'asterisco. Per ulteriori informazioni, consulta Caratteri speciali.
Tabella: Il nome della tabella del database che contiene i dati da copiare. Sostituire table_name con il nome della tabella del database
connectionString: La stringa di connessione JDBC per la connessione CopyActivity dell'oggetto al database.
selectQuery: Una query SQL SELECT valida che specifichi quali dati copiare dalla tabella di database. Si noti che #{table} è un'espressione che riutilizza il nome della tabella fornito dalla variabile "tabella" nelle righe precedenti del file JSON.
Type (Tipo): Il SqlDataNode tipo, che è un'istanza Amazon RDS che utilizza MySQL in questo esempio.
Nota
Il tipo MySqlDataNode è obsoleto. Sebbene tu possa ancora utilizzarloMySqlDataNode, ti consigliamo di utilizzarloSqlDataNode.

Nodo dati Amazon S3

Successivamente, il componente della pipeline S3Output definisce una posizione per il file di output; in questo caso un file CSV in una posizione del bucket Amazon S3. Il DataNode componente S3 di output è definito dai seguenti campi:


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

Id: L'ID definito dall'utente, un'etichetta solo di riferimento.
Pianificazione: Un riferimento al componente di pianificazione che abbiamo creato nelle righe precedenti del file JSON.
filePath: Il percorso ai dati associati al nodo di dati, cioè il file di output CSV in questo esempio.
Nome: Il nome definito dall'utente, un'etichetta solo di riferimento.
Type (Tipo): Il tipo di oggetto della pipeline, che corrisponde DataNode alla posizione in cui risiedono i dati, è S3 in un bucket Amazon S3.

Risorsa

Questa è una definizione della risorsa di calcolo che esegue l'operazione di copia. In questo esempio, AWS Data Pipeline deve creare automaticamente un'istanza EC2 per eseguire l'attività di copia e terminare la risorsa dopo aver completato l'attività. I campi definiti qui controllano la creazione e la funzione dell'istanza EC2 che esegue il lavoro. EC2Resource è definito dai campi seguenti:


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

Id: L'ID definito dall'utente, un'etichetta solo di riferimento.
Pianificazione: La pianificazione su cui creare questa risorsa di calcolo.
Nome: Il nome definito dall'utente, un'etichetta solo di riferimento.
Ruolo: Il ruolo IAM dell'account che accede alle risorse, ad esempio l'accesso a un bucket Amazon S3 per recuperare i dati.
Type (Tipo): Il tipo di risorsa di calcolo per eseguire il lavoro, in questo caso, un'istanza EC2. Sono disponibili altri tipi di risorse, ad esempio un EmrCluster tipo.
resourceRole: Il ruolo IAM dell'account che crea le risorse, ad esempio la creazione e la configurazione di un'istanza EC2 a tuo nome. Ruolo e ResourceRole possono avere lo stesso ruolo, ma forniscono separatamente una maggiore granularità nella configurazione di sicurezza.

Attività

L'ultima sezione del file JSON è la definizione dell'attività che rappresenta il lavoro da eseguire. In questo caso utilizziamo un CopyActivity componente per copiare i dati da un file in un bucket Amazon S3 a un altro file. Questo componente CopyActivity è definito dai campi seguenti:


{
  "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: L'ID definito dall'utente, un'etichetta solo di riferimento
Input: Posizione dei dati MySQL da copiare
Pianificazione: La pianificazione su cui eseguire questa attività
Nome: Il nome definito dall'utente, un'etichetta solo di riferimento
runsOn: La risorsa di calcolo che esegue il lavoro definito dall'attività. In questo esempio, forniamo un riferimento all'istanza EC2 definita in precedenza. Utilizzando il campo runsOn, AWS Data Pipeline crea l'istanza EC2 per l'utente. Il campo runsOn indica che la risorsa è disponibile nell'infrastruttura AWS, mentre il valore workerGroup indica che si desidera utilizzare le proprie risorse locali per eseguire il lavoro.
onSuccess: SnsAlarm da inviare se l'attività viene completata correttamente
onFail: SnsAlarm da inviare se l'attività non viene completata correttamente
Output: La posizione Amazon S3 del file di output CSV
Type (Tipo): Il tipo di attività da eseguire.

Caricamento e attivazione della definizione della pipeline

Devi caricare la definizione della pipeline e attivare la pipeline. Nei seguenti comandi di esempio, sostituisci pipeline_name con un'etichetta per la tua pipeline e pipeline_file con il percorso completo per il file di definizione della pipeline. .json

AWS CLI

Per creare la definizione della pipeline e attivare la pipeline, usa il seguente comando create-pipeline. Nota l'ID della tua pipeline, perché utilizzerai questo valore con la maggior parte dei comandi CLI.


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

Per caricare la definizione della pipeline, usa il seguente put-pipeline-definitioncomando.


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

Se la pipeline viene convalidata correttamente, il validationErrors campo è vuoto. È necessario esaminare eventuali avvertenze.

Per attivare la pipeline, usa il seguente comando activate-pipeline.


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

Puoi verificare che la tua pipeline appaia nell'elenco delle pipeline utilizzando il seguente comando list-pipelines.


aws datapipeline list-pipelines

Avvertimento JavaScript è disabilitato o non è disponibile nel tuo browser.

Per usare la documentazione AWS, JavaScript deve essere abilitato. Consulta le pagine della guida del browser per le istruzioni.

Convenzioni dei documenti

Prima di iniziare

Copiare dati su Amazon Redshift