Definieren Sie eine Pipeline im JSON-Format Hochladen und Aktivieren der Pipeline-Definition

MySQL-Daten über die Befehlszeile kopieren

Sie können eine Pipeline erstellen, um Daten aus einer MySQL-Tabelle in eine Datei in einem Amazon S3-Bucket zu kopieren.

Voraussetzungen

Bevor Sie beginnen, müssen Sie die folgenden Schritte ausführen:

Installieren und konfigurieren Sie eine Befehlszeilenschnittstelle (CLI). Weitere Informationen finden Sie unter Zugriff auf AWS Data Pipeline.
Stellen Sie sicher, dass die IAM-Rollen benannt DataPipelineDefaultResourceRolesind DataPipelineDefaultRoleund existieren. Die AWS Data Pipeline Konsole erstellt diese Rollen automatisch für Sie. Wenn Sie die AWS Data Pipeline Konsole nicht mindestens einmal verwendet haben, müssen Sie diese Rollen manuell erstellen. Weitere Informationen finden Sie unter IAM-Rollen für AWS Data Pipeline.
Richten Sie einen Amazon S3-Bucket und eine Amazon RDS-Instance ein. Weitere Informationen finden Sie unter Bevor Sie beginnen.

Aufgaben

Definieren Sie eine Pipeline im JSON-Format
Hochladen und Aktivieren der Pipeline-Definition

Definieren Sie eine Pipeline im JSON-Format

Dieses Beispielszenario zeigt, wie JSON-Pipeline-Definitionen und die AWS Data Pipeline CLI verwendet werden, um Daten (Zeilen) aus einer Tabelle in einer MySQL-Datenbank in eine CSV-Datei (durch Kommas getrennte Werte) in einem Amazon S3-Bucket in einem bestimmten Zeitintervall zu kopieren.

Dies ist die vollständige Pipeline-Definition-JSON-Datei, gefolgt von einer Erläuterung für jeden ihrer Abschnitte.

Anmerkung

Wir empfehlen, dass Sie einen Texteditor verwenden, mit dem Sie die Syntax von JSON-formatierten Dateien überprüfen und die Datei mit der Dateierweiterung .json benennen können.


{
  "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-Datenknoten

Die MySqlDataNode Eingabe-Pipeline-Komponente definiert einen Speicherort für die Eingabedaten, in diesem Fall eine Amazon RDS-Instance. Die MySqlDataNode Eingabekomponente wird durch die folgenden Felder definiert:


{
  "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: Den benutzerdefinierten Namen, der nur als Referenz dient.
Benutzername: Der Benutzername des Datenbank-Kontos, das über ausreichend Berechtigungen verfügt, um Daten aus der Datenbanktabelle abzurufen. Ersetzen Sie my-username durch den Namen Ihres Benutzers.
Plan: Ein Verweis auf die Zeitplankomponente, die wir in den vorhergehenden Zeilen der JSON-Datei erstellt haben.
Name: Den benutzerdefinierten Namen, der nur als Referenz dient.
*Passwort: Das Passwort für das Datenbankkonto mit dem Sternchen-Präfix, das darauf hinweist, dass AWS Data Pipeline den Passwortwert verschlüsseln muss. Ersetzen Sie my-password durch das richtige Passwort für Ihren Benutzer. Dem Passwortfeld ist das Sternchen-Sonderzeichen vorangestellt. Weitere Informationen finden Sie unter Sonderzeichen.
Tabelle: Der Name der Datenbanktabelle, die die zu kopierenden Daten enthält. Ersetzen Sie table-name durch den Namen Ihrer Datenbanktabelle.
connectionString: Die JDBC-Verbindungszeichenfolge für das CopyActivity Objekt, das eine Verbindung zur Datenbank herstellen soll.
selectQuery: Eine gültige SQL-SELECT-Abfrage, die festlegt, welche Daten aus der Datenbanktabelle kopiert werden sollen. Beachten Sie, dass #{table} ein Ausdruck ist, der den Tabellennamen wiederverwendet, der in den vorhergehenden Zeilen der JSON-Datei durch die Variable "table" angegeben wird.
Typ: Der SqlDataNode Typ, bei dem es sich in diesem Beispiel um eine Amazon RDS-Instance handelt, die MySQL verwendet.
Anmerkung
Der MySqlDataNode-Typ ist veraltet. Sie können es zwar weiterhin verwendenMySqlDataNode, wir empfehlen jedoch, es zu verwendenSqlDataNode.

Amazon S3-Datenknoten

Als Nächstes definiert die S3Output-Pipeline-Komponente einen Speicherort für die Ausgabedatei, in diesem Fall eine CSV-Datei in einem Amazon S3-Bucket-Speicherort. Die DataNode S3-Ausgabekomponente wird durch die folgenden Felder definiert:


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

Id: Die benutzerdefinierte ID, die nur als Referenz dient.
Plan: Ein Verweis auf die Zeitplankomponente, die wir in den vorhergehenden Zeilen der JSON-Datei erstellt haben.
filePath: Den Pfad zu den Daten, die mit dem Datenknoten verknüpft sind, der in diesem Beispiel eine CSV-Ausgabedatei ist.
Name: Den benutzerdefinierten Namen, der nur als Referenz dient.
Typ: Der Pipeline-Objekttyp, der S3 ist, DataNode um dem Ort zu entsprechen, an dem sich die Daten befinden, in einem Amazon S3-Bucket.

Ressource

Dies ist eine Definition der Rechenressource, die die Kopieroperation ausführt. In diesem Beispiel sollte AWS Data Pipeline automatisch eine EC2-Instance erstellen, um die Kopieraufgabe auszuführen und die Ressource zu beenden, nachdem die Aufgabe abgeschlossen wurde. Die hier definierten Felder steuern die Erstellung und Funktion der EC2 Instance, die die Arbeit erledigt. Die EC2Resource ist durch folgende Felder definiert:


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

Id: Die benutzerdefinierte ID, die nur als Referenz dient.
Plan: Der Zeitplan für die Erstellung dieser Rechenressource.
Name: Den benutzerdefinierten Namen, der nur als Referenz dient.
Rolle: Die IAM-Rolle des Kontos, das auf Ressourcen zugreift, z. B. auf einen Amazon S3-Bucket zugreift, um Daten abzurufen.
Typ: Die Art der Rechenressource zur Ausführung der Arbeit; in diesem Fall eine EC2 Instance. Es sind andere Ressourcentypen verfügbar, z. B. ein EmrCluster Typ.
resourceRole: Die IAM-Rolle des Kontos, das Ressourcen erstellt, z. B. für Sie eine EC2 Instance erstellt und konfiguriert. Rolle und ResourceRole können dieselbe Rolle sein, bieten jedoch getrennt voneinander eine größere Granularität in Ihrer Sicherheitskonfiguration.

Aktivität

Der letzte Abschnitt in der JSON-Datei ist die Definition der Aktivität, die die auszuführende Arbeit darstellt. In diesem Fall verwenden wir eine CopyActivity Komponente, um Daten aus einer Datei in einem Amazon S3-Bucket in eine andere Datei zu kopieren. Die CopyActivity-Komponente ist durch folgende Felder definiert:


{
  "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: Die benutzerdefinierte ID, die nur als Referenz dient
Eingabe: Den Speicherort der zu kopierenden MySQL-Daten
Plan: Den Zeitplan für die Ausführung dieser Aktivität
Name: Den benutzerdefinierten Namen, der nur als Referenz dient
runsOn: Die Datenverarbeitungsressource, die die Arbeit ausführt, die diese Aktivität definiert. In diesem Beispiel stellen wir einen Verweis auf die zuvor definierte EC2 Instance bereit. Durch die Verwendung des Felds runsOn wird AWS Data Pipeline veranlasst, die EC2-Instance für Sie zu erstellen. Das Feld runsOn zeigt an, dass die Ressource in der AWS-Infrastruktur vorhanden ist, während der Wert workerGroup angibt, dass Sie Ihre eigenen lokalen Ressourcen zur Ausführung der Arbeit verwenden möchten.
onSuccess: Den SnsAlarm, der versendet werden soll, wenn die Aktivität erfolgreich abgeschlossen wurde.
onFail: Den SnsAlarm, der versendet werden soll, wenn die Aktivität fehlschlägt.
Ausgabe: Der Amazon S3-Speicherort der CSV-Ausgabedatei
Typ: Den Typ der Aktivität, die durchgeführt werden soll.

Hochladen und Aktivieren der Pipeline-Definition

Sie müssen Ihre Pipeline-Definition hochladen und Ihre Pipeline aktivieren. Ersetzen Sie in den folgenden Beispielbefehlen pipeline_name durch ein Label für Ihre Pipeline und pipeline_file durch den vollständig qualifizierten Pfad für die Pipeline-Definitionsdatei. .json

AWS CLI

Verwenden Sie den folgenden Befehl create-pipeline, um Ihre Pipeline-Definition zu erstellen und Ihre Pipeline zu aktivieren. Notieren Sie sich die ID Ihrer Pipeline, da Sie diesen Wert bei den meisten CLI-Befehlen verwenden werden.


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

Verwenden Sie den folgenden put-pipeline-definitionBefehl, um Ihre Pipeline-Definition hochzuladen.


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

Wenn Ihre Pipeline erfolgreich validiert wurde, ist das validationErrors Feld leer. Sie sollten alle Warnungen überprüfen.

Verwenden Sie den folgenden Befehl activate-pipeline, um Ihre Pipeline zu aktivieren.


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

Mit dem folgenden Befehl list-pipelines können Sie überprüfen, ob Ihre Pipeline in der Pipelineliste erscheint.


aws datapipeline list-pipelines

Warnung JavaScript ist in Ihrem Browser nicht verfügbar oder deaktiviert.

Zur Nutzung der AWS-Dokumentation muss JavaScript aktiviert sein. Weitere Informationen finden auf den Hilfe-Seiten Ihres Browsers.

Dokumentkonventionen

Bevor Sie beginnen

Daten nach Amazon Redshift kopieren