Erstellen einer SSH-Verbindung mit dem SSHOperator

Das folgende Beispiel beschreibt, wie Sie die DAG (SSHOperatorIn a Directed Acyclic Graph) verwenden können, um von Ihrer Amazon Managed Workflows for Apache Airflow-Umgebung aus eine Verbindung zu einer Amazon EC2-Remoteinstanz herzustellen. Sie können einen ähnlichen Ansatz verwenden, um eine Verbindung zu einer beliebigen Remote-Instance mit SSH-Zugriff herzustellen.

Im folgenden Beispiel laden Sie einen geheimen SSH-Schlüssel (.pem) in das dags Verzeichnis Ihrer Umgebung auf Amazon S3 hoch. Anschließend installieren Sie die erforderlichen Abhängigkeiten mithilfe der Benutzeroberfläche requirements.txt und erstellen eine neue Apache Airflow-Verbindung. Schließlich schreiben Sie eine DAG, die eine SSH-Verbindung zur Remote-Instanz herstellt.

Version

• Sie können das Codebeispiel auf dieser Seite mit Apache Airflow v2 und höher in Python 3.10 verwenden.

Voraussetzungen

Um den Beispielcode auf dieser Seite zu verwenden, benötigen Sie Folgendes:

• Eine Amazon MWAA-Umgebung.

• Ein geheimer SSH-Schlüssel. Das Codebeispiel geht davon aus, dass Sie über eine Amazon EC2 EC2-Instance verfügen und sich .pem in derselben Region wie Ihre Amazon MWAA-Umgebung befinden. Wenn Sie keinen Schlüssel haben, finden Sie weitere Informationen unter Erstellen oder Importieren eines key pair im Amazon EC2 EC2-Benutzerhandbuch.

Berechtigungen

• Für die Verwendung des Codebeispiels auf dieser Seite sind keine zusätzlichen Berechtigungen erforderlich.

Voraussetzungen

Fügen Sie den folgenden Parameter hinzurequirements.txt, um das apache-airflow-providers-ssh Paket auf dem Webserver zu installieren. Sobald Ihre Umgebung aktualisiert wurde und Amazon MWAA die Abhängigkeit erfolgreich installiert hat, wird in der Benutzeroberfläche ein neuer SSH-Verbindungstyp angezeigt.

-c https://raw.githubusercontent.com/apache/airflow/constraints-Airflow-version/constraints-Python-version.txt
apache-airflow-providers-ssh

Anmerkung

-cdefiniert die URL der Einschränkungen in. requirements.txt Dadurch wird sichergestellt, dass Amazon MWAA die richtige Paketversion für Ihre Umgebung installiert.

Kopieren Sie Ihren geheimen Schlüssel zu Amazon S3

Verwenden Sie den folgenden AWS Command Line Interface Befehl, um Ihren .pem Schlüssel in das dags Verzeichnis Ihrer Umgebung in Amazon S3 zu kopieren.

$ aws s3 cp your-secret-key.pem s3://your-bucket/dags/

Amazon MWAA kopiert den Inhalt einschließlich des .pem Schlüssels in dags das lokale /usr/local/airflow/dags/ Verzeichnis. Auf diese Weise kann Apache Airflow auf den Schlüssel zugreifen.

Erstellen Sie eine neue Apache Airflow-Verbindung

Um eine neue SSH-Verbindung mit der Apache Airflow-Benutzeroberfläche zu erstellen

1. Öffnen Sie die Seite Umgebungen auf der Amazon MWAA-Konsole.

2. Wählen Sie aus der Liste der Umgebungen Open Airflow UI für Ihre Umgebung aus.

3. Wählen Sie auf der Apache Airflow-UI-Seite in der oberen Navigationsleiste Admin aus, um die Dropdownliste zu erweitern, und wählen Sie dann Verbindungen aus.

4. Wählen Sie auf der Seite Verbindungen auflisten die Option + oder die Schaltfläche Neuen Datensatz hinzufügen, um eine neue Verbindung hinzuzufügen.

5. Fügen Sie auf der Seite „Verbindung hinzufügen“ die folgenden Informationen hinzu:

   1. Geben Sie als Verbindungs-ID einssh_new.

   2. Wählen Sie als Verbindungstyp SSH aus der Dropdownliste aus.

      Anmerkung

      Wenn der SSH-Verbindungstyp nicht in der Liste verfügbar ist, hat Amazon MWAA das erforderliche Paket nicht installiert. apache-airflow-providers-ssh Aktualisieren Sie Ihre requirements.txt Datei, sodass sie dieses Paket enthält, und versuchen Sie es erneut.

   3. Geben Sie für Host die IP-Adresse der Amazon EC2 EC2-Instance ein, zu der Sie eine Verbindung herstellen möchten. z. B. 12.345.67.89.

   4. Geben Sie unter Nutzername ein, ec2-user ob Sie eine Verbindung zu einer Amazon EC2 EC2-Instance herstellen. Ihr Benutzername kann je nach Art der Remote-Instance, zu der Apache Airflow eine Verbindung herstellen soll, unterschiedlich sein.

   5. Geben Sie für Extra das folgende Schlüssel-Wert-Paar im JSON-Format ein:

      { "key_file": "/usr/local/airflow/dags/your-secret-key.pem" }

      Dieses Schlüssel-Wert-Paar weist Apache Airflow an, im lokalen Verzeichnis nach dem geheimen Schlüssel zu suchen. /dags

Codebeispiel

Die folgende DAG verwendet dieSSHOperator, um eine Verbindung zu Ihrer Amazon EC2 EC2-Zielinstanz herzustellen, und führt dann den hostname Linux-Befehl aus, um den Namen der Instanz auszudrucken. Sie können die DAG so ändern, dass sie jeden Befehl oder jedes Skript auf der Remote-Instance ausführt.

1. Öffnen Sie ein Terminal und navigieren Sie zu dem Verzeichnis, in dem Ihr DAG-Code gespeichert ist. Beispielsweise:

   cd dags

2. Kopieren Sie den Inhalt des folgenden Codebeispiels und speichern Sie ihn lokal unterssh.py.

   from airflow.decorators import dag
   from datetime import datetime
   from airflow.providers.ssh.operators.ssh import SSHOperator
   
   @dag(
       dag_id="ssh_operator_example",
       schedule_interval=None,     
       start_date=datetime(2022, 1, 1),
       catchup=False,
       )
   def ssh_dag():
       task_1=SSHOperator(
           task_id="ssh_task",
           ssh_conn_id='ssh_new',
           command='hostname',
       )
   
   my_ssh_dag = ssh_dag()

3. Führen Sie den folgenden AWS CLI Befehl aus, um die DAG in den Bucket Ihrer Umgebung zu kopieren, und lösen Sie die DAG dann mithilfe der Apache Airflow-Benutzeroberfläche aus.

   $ aws s3 cp your-dag.py s3://your-environment-bucket/dags/

4. Bei Erfolg erhalten Sie in den Aufgabenprotokollen für die ssh_operator_example DAG eine Ausgabe, die ssh_task der folgenden ähnelt:

   [2022-01-01, 12:00:00 UTC] {{base.py:79}} INFO - Using connection to: id: ssh_new. Host: 12.345.67.89, Port: None,
   Schema: , Login: ec2-user, Password: None, extra: {'key_file': '/usr/local/airflow/dags/your-secret-key.pem'}
   [2022-01-01, 12:00:00 UTC] {{ssh.py:264}} WARNING - Remote Identification Change is not verified. This won't protect against Man-In-The-Middle attacks
   [2022-01-01, 12:00:00 UTC] {{ssh.py:270}} WARNING - No Host Key Verification. This won't protect against Man-In-The-Middle attacks
   [2022-01-01, 12:00:00 UTC] {{transport.py:1819}} INFO - Connected (version 2.0, client OpenSSH_7.4)
   [2022-01-01, 12:00:00 UTC] {{transport.py:1819}} INFO - Authentication (publickey) successful!
   [2022-01-01, 12:00:00 UTC] {{ssh.py:139}} INFO - Running command: hostname
   [2022-01-01, 12:00:00 UTC]{{ssh.py:171}} INFO - ip-123-45-67-89.us-west-2.compute.internal
   [2022-01-01, 12:00:00 UTC] {{taskinstance.py:1280}} INFO - Marking task as SUCCESS. dag_id=ssh_operator_example, task_id=ssh_task, execution_date=20220712T200914, start_date=20220712T200915, end_date=20220712T200916