Airflow-Verbindungen verwalten

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Auf dieser Seite wird beschrieben, wie Sie Airflow-Verbindungen in Ihrem und über Ihre DAGs darauf zugreifen.

Airflow-Verbindungen

In Airflow-Verbindungen werden Anmeldedaten und andere Verbindungsinformationen wie Nutzernamen, Verbindungsstrings und Passwörter gespeichert. Ihre DAGs verwenden Verbindungen zu Kommunikation und Zugriff auf Ressourcen in Google Cloud und anderen Diensten aus Ihren DAGs.

Airflow-Operatoren in Ihren DAGs verwenden entweder eine Standardverbindung für den Operator oder Sie geben einen benutzerdefinierten Verbindungsnamen an.

Verbindungssicherheit

Die meisten Airflow-Betreiber akzeptieren Anmeldedaten nicht direkt. Stattdessen werden Airflow-Verbindungen verwendet.

Beim Erstellen einer neuen Umgebung generiert Cloud Composer ein Eindeutiger, dauerhafter Fenet-Schlüssel für die Umgebung und sichert Verbindungsextras. ist standardmäßig aktiviert. Sie können fernet_key auf der Seite Konfiguration in der Airflow-Benutzeroberfläche aufrufen.

Weitere Informationen zum Schutz von Verbindungen und Passwörtern finden Sie in Airflow, siehe Verbindungen sichern und Maskieren sensibler Daten

Verbindungstypen

Airflow verwendet Verbindungen verschiedener Typen, um eine Verbindung zu bestimmten Diensten herzustellen. Beispiel: Der Parameter Google Cloud-Verbindungstyp eine Verbindung zu anderen Diensten in Google Cloud herstellen. Ein weiteres Beispiel: Der S3-Verbindungstyp stellt eine Verbindung zu einem Amazon S3-Bucket her.

Um Airflow einen Verbindungstyp hinzuzufügen, ein PyPI-Paket mit diesem Verbindungstyp installieren. Einige Pakete sind in Ihrer Umgebung vorinstalliert. Für können Sie eine Verbindung von apache-airflow-providers-google verwenden. ohne benutzerdefinierte PyPI-Pakete zu installieren.

Vorkonfigurierte Verbindungen

Cloud Composer konfiguriert die folgenden Standardverbindungen in Ihrem zu verbessern. Sie können diese Verbindungen verwenden, um ohne Konfigurieren auf Ressourcen in Ihrem Projekt zuzugreifen.

  • google_cloud_default
  • bigquery_default
  • google_cloud_datastore_default
  • google_cloud_storage_default

Verbindung in Secret Manager hinzufügen

Sie können eine Verbindung in Secret Manager speichern, an Airflow. Wir empfehlen, diesen Ansatz beim Speichern von Anmeldedaten und oder andere vertrauliche Informationen.

So fügen Sie eine Verbindung in Secret Manager hinzu:

  1. Konfigurieren Sie Secret Manager für Ihre Umgebung.

  2. Fügen Sie ein Secret hinzu, dessen Name dem Muster für Verbindungen entspricht.

    Beispiel: airflow-connections-example_connection. Verwenden Sie in Ihren DAGs den Verbindungsnamen ohne Präfix: example_connection.

  3. Fügen Sie der Verbindung Parameter hinzu:

    JSON-Format

    Fügen Sie die JSON-Darstellung Ihrer Verbindung als Wert des Geheimnisses hinzu. Beispiel:

    {
  "conn_type": "mysql",
  "host": "example.com",
  "login": "login",
  "password": "password",
  "port": "9000"
}

    Weitere Informationen zum JSON-Verbindungsformat finden Sie in der Airflow-Dokumentation.

    URI-Format

    Fügen Sie die URI-Darstellung Ihrer Verbindung als Wert des Secret:

    • Das Secret muss eine URI-Darstellung der Verbindung speichern. Beispiel: mysql://login:password@example.com:9000

    • Der URI muss URL-codiert sein. Für Beispiel: Ein Passwort, das ein Leerzeichen enthält, muss URL-codiert werden. wie folgt: mysql://login:secret%20password@example.com:9000

    Airflow bietet eine einfache Methode zum Generieren von Verbindungs-URIs. Beispiel für das Codieren einer komplexen URL mit JSON-Extras ist verfügbar in der Airflow-Dokumentation

  4. Prüfen Sie, ob alle Verbindungsparameter richtig aus Secret Manager gelesen werden.

Verbindung in Airflow hinzufügen

Als Alternative zum Speichern Ihrer Verbindungen in können Sie sie in Airflow speichern.

So fügen Sie eine Verbindung in Airflow hinzu:

Airflow-Befehlszeile

Führen Sie den connections add Airflow-Befehlszeile über die Google Cloud CLI ausführen. Beispiel:

gcloud composer environments run ENVIRONMENT_NAME \
  --location LOCATION \
  connections add -- \
    --conn-type "mysql" \
    --conn-host "example.com" \
    --conn-port "9000" \
    --conn-login "login" \
    --conn-password "password" \
    example_connection

Sie können auch das Argument --conn-uri verwenden:

gcloud composer environments run ENVIRONMENT_NAME \
  --location LOCATION \
  connections add -- \
    --conn-uri "mysql://login:password@example.com:9000" \
    example_connection

Ersetzen Sie Folgendes:

  • ENVIRONMENT_NAME: der Name Ihrer Umgebung
  • LOCATION: die Region, in der sich die Umgebung befindet.

Airflow-UI

Folgen Sie der Anleitung zum Erstellen von Verbindungen in Airflow.

Prüfen, ob Airflow eine Verbindung richtig liest

Sie können den Airflow-Befehl connections get über die Google Cloud CLI ausführen, um zu prüfen, ob eine Verbindung richtig gelesen wird. Wenn Sie beispielsweise eine Verbindung in Secret Manager speichern, können Sie prüfen, ob alle Parameter einer Verbindung von Airflow aus einem Secret gelesen werden.

gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    connections get \
    -- CONNECTION_NAME

Ersetzen Sie:

  • ENVIRONMENT_NAME durch den Namen der Umgebung.
  • LOCATION durch die Region, in der sich die Umgebung befindet.
  • CONNECTION_NAME durch den Namen der Verbindung. Wenn Ihre Verbindung in Secret Manager gespeichert ist, verwenden Sie den Verbindungsnamen ohne Verbindungspräfix. Geben Sie z. B. example_connection anstelle von airflow-connections-example_connection_json.

Beispiel:

gcloud composer environments run example-environment \
    --location us-central1 \
    connections get \
    -- example_connection -o json

Airflow-Verbindungen in DAGs verwenden

In diesem Abschnitt wird beschrieben, wie Sie über einen DAG auf Ihre Verbindung zugreifen.

Secret Manager-Verbindung verwenden

Verwenden Sie den Namen der Verbindung ohne Präfix. Wenn beispielsweise Ihr Secret heißt airflow-connections-aws_s3, bitte geben Sie aws_s3.

transfer_dir_from_s3 = S3ToGCSOperator(
    task_id='transfer_dir_from_s3',
    aws_conn_id='aws_s3',
    prefix='data-for-gcs',
    bucket='example-s3-bucket-transfer-operators',
    dest_gcs='gs://us-central1-example-environ-361f4221-bucket/data/from-s3/')

Wenn Sie eine Standardverbindung im Secret Manager speichern, können Sie den Verbindungsnamen weglassen. In der Airflow-Dokumentation für einen bestimmten Operator finden Sie den Standardverbindungsnamen, der von einem Operator verwendet wird. Der Airflow-Operator S3ToGCSOperator verwendet beispielsweise standardmäßig die Verbindung aws_default. Sie können diese Standardverbindung in einem Secret mit dem Namen airflow-connections-aws_default

In Airflow gespeicherte Verbindung verwenden

Verwenden Sie den Namen der Verbindung, wie er in Airflow definiert ist:

transfer_dir_from_s3 = S3ToGCSOperator(
    task_id='transfer_dir_from_s3',
    aws_conn_id='aws_s3',
    prefix='data-for-gcs',
    bucket='example-s3-bucket-transfer-operators',
    dest_gcs='gs://us-central1-example-environ-361f4221-bucket/data/from-s3/')

Wenn Sie die Standardverbindung für einen Operator verwenden möchten, lassen Sie den Verbindungsnamen weg. Weitere Informationen finden Sie unter Airflow-Dokumentation für einen bestimmten Operator zum Abrufen der Standardverbindung der von einem Operator verwendet wird. Beispiel: Der Airflow-Operator S3ToGCSOperator verwendet standardmäßig die aws_default-Verbindung.

Fehlerbehebung

Wenn Ihre Umgebung nicht auf das im Secret Manager gespeicherte Secret zugreifen kann:

  1. Secret Manager muss in Ihrem zu verbessern.

  2. Prüfen Sie, ob der Name der Verbindung in Secret Manager dem entspricht mit der von Airflow verwendeten Verbindung. Bei einer Verbindung mit dem Namen example_connection lautet der Secret-Name beispielsweise airflow-connections-example_connection.

  3. Prüfen Sie, ob Airflow eine Verbindung richtig liest.

Nächste Schritte