Ausführungen planen

In diesem Dokument wird Folgendes beschrieben:

Hinweise

Wenn Sie Ausführungen mit Workflowkonfigurationen planen oder Ausführungen mit Workflows und Cloud Scheduler planen möchten, gehen Sie so vor:

  1. Öffnen Sie in der Google Cloud Console die Seite Dataform.

    Zu Dataform

  2. Wählen Sie ein Repository aus oder erstellen Sie ein neues.

  3. Erstellen Sie eine Releasekonfiguration.

Wenn Sie Ausführungen mit Cloud Composer planen möchten, gehen Sie so vor:

  1. Wählen Sie ein Dataform-Repository aus oder erstellen Sie ein neues.
  2. Weisen Sie Dataform Zugriff auf BigQuery zu.
  3. Wählen Sie einen Dataform-Arbeitsbereich aus oder erstellen Sie einen.
  4. Erstellen Sie mindestens eine Tabelle.
  5. Erstellen Sie eine Cloud Composer 2-Umgebung.

Erforderliche Rollen

Bitten Sie Ihren Administrator, Ihnen die folgenden IAM-Rollen zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Ausführen der Aufgaben in diesem Dokument benötigen:

Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Sie können die erforderlichen Berechtigungen auch über benutzerdefinierte Rollen oder andere vordefinierte Rollen erhalten.

Wenn Sie ein anderes Dienstkonto als das standardmäßige Dataform-Dienstkonto verwenden möchten, gewähren Sie Zugriff auf das benutzerdefinierte Dienstkonto.

Läufe mit Workflowkonfigurationen planen

In diesem Abschnitt erfahren Sie, wie Sie in Dataform eine Workflowkonfiguration erstellen, um Workflowausführungen zu planen und zu konfigurieren. Mit Workflowkonfigurationen können Sie Dataform-Workflows nach einem Zeitplan ausführen.

Workflowkonfigurationen

Wenn Sie Dataform-Ausführungen aller oder ausgewählter Workflowaktionen in BigQuery planen möchten, können Sie Workflowkonfigurationen erstellen. In einer Workflowkonfiguration wählen Sie eine Build-Release-Konfiguration, Workflowaktionen für die Ausführung und den Ausführungszeitplan aus.

Anschließend werden bei einem geplanten Durchlauf Ihrer Workflowkonfiguration die ausgewählten Aktionen aus dem letzten Kompilierungsergebnis in Ihrer Release-Konfiguration von Dataform in BigQuery bereitgestellt. Sie können die Ausführung einer Workflowkonfiguration auch manuell mit der Dataform API workflowConfigs auslösen.

Eine Dataform-Workflowkonfiguration enthält die folgenden Ausführungseinstellungen:

  • ID der Workflowkonfiguration.
  • Releasekonfiguration.

  • Dienstkonto.

    Das ist das Dienstkonto, das der Workflowkonfiguration zugeordnet ist. Sie können das standardmäßige Dataform-Dienstkonto oder ein Dienstkonto auswählen, das mit Ihrem Google Cloud-Projekt verknüpft ist. Sie können auch manuell ein anderes Dienstkonto eingeben. Standardmäßig verwenden Workflowkonfigurationen dieselben Dienstkonten wie ihre Repositories.

  • Auszuführende Workflow-Aktionen:

    • Alle Aktionen.
    • Auswahl von Aktionen
    • Auswahl von Tags

  • Ausführungszeitplan und Zeitzone

Workflowkonfiguration erstellen

So erstellen Sie eine Dataform-Workflowkonfiguration:

  1. Gehen Sie in Ihrem Repository zu Releases und Planung.
  2. Klicken Sie im Bereich Workflowkonfigurationen auf Erstellen.

  3. Geben Sie im Bereich Workflowkonfiguration erstellen im Feld Konfigurations-ID eine eindeutige ID für die Workflowkonfiguration ein.

    IDs dürfen nur Ziffern, Buchstaben, Bindestriche und Unterstriche enthalten.

  4. Wählen Sie im Menü Releasekonfiguration eine Konfiguration für die Release-Kompilierung aus.

  5. Optional: Geben Sie im Feld Häufigkeit die Häufigkeit der Ausführungen im unix-cron-Format ein.

    Damit Dataform das neueste Kompilierungsergebnis in der entsprechenden Releasekonfiguration ausführt, muss zwischen dem Erstellen des Kompilierungsergebnisses und der geplanten Ausführung mindestens eine Stunde liegen.

  6. Wählen Sie im Menü Dienstkonto ein Dienstkonto für die Workflowkonfiguration aus.

    Im Menü können Sie das Standarddienstkonto von Dataform oder ein beliebiges Dienstkonto auswählen, das mit Ihrem Google Cloud-Projekt verknüpft ist und auf das Sie Zugriff haben. Wenn Sie kein Dienstkonto auswählen, wird in der Workflowkonfiguration das Dienstkonto des Repositories verwendet.

  7. Optional: Wählen Sie im Menü Zeitzone die Zeitzone für die Ausführungen aus.

    Die Standardzeitzone ist UTC.

  8. Wählen Sie die auszuführenden Workflow-Aktionen aus:

    • Wenn Sie den gesamten Workflow ausführen möchten, klicken Sie auf Alle Aktionen.
    • Wenn Sie ausgewählte Aktionen im Workflow ausführen möchten, klicken Sie auf Auswahl von Aktionen und wählen Sie die Aktionen aus.
    • Wenn Sie Aktionen mit ausgewählten Tags ausführen möchten, klicken Sie auf Auswahl von Tags und wählen Sie die entsprechenden Tags aus.
    • Optional: Wenn Sie ausgewählte Aktionen oder Tags und ihre Abhängigkeiten ausführen möchten, wählen Sie die Option Abhängigkeiten einschließen aus.
    • Optional: Wenn Sie ausgewählte Aktionen oder Tags und ihre abhängigen Elemente ausführen möchten, wählen Sie die Option Abhängige Elemente einschließen aus.
    • Optional: Wenn Sie alle Tabellen neu erstellen möchten, wählen Sie die Option Mit vollständiger Aktualisierung ausführen aus.

    Ohne diese Option aktualisiert Dataform inkrementelle Tabellen, ohne sie von Grund auf neu zu erstellen.

  9. Klicken Sie auf Erstellen.

In der folgenden Workflowkonfiguration werden beispielsweise jede Stunde in der Zeitzone CEST Aktionen mit dem Tag hourly ausgeführt:

  • Konfigurations-ID: production-hourly
  • Releasekonfiguration: –
  • Frequenz: 0 * * * *
  • Zeitzone: Central European Summer Time (CEST)
  • Auswahl von Workflow-Aktionen: Auswahl von Tags, hourly-Tag

Workflowkonfiguration bearbeiten

So bearbeiten Sie eine Workflow-Konfiguration:

  1. Gehen Sie in Ihrem Repository zu Releases und Planung.
  2. Klicken Sie neben der Workflowkonfiguration, die Sie bearbeiten möchten, auf das Dreipunkt-Menü  Mehr und dann auf Bearbeiten.
  3. Bearbeiten Sie im Bereich Workflowkonfiguration bearbeiten die Einstellungen für die Releasekonfiguration und klicken Sie dann auf Speichern.

Workflowkonfiguration löschen

So löschen Sie eine Workflowkonfiguration:

  1. Gehen Sie in Ihrem Repository zu Releases und Planung.
  2. Klicken Sie neben der Workflowkonfiguration, die Sie löschen möchten, auf das Dreipunkt-Menü  Mehr und dann auf Löschen.
  3. Klicken Sie im Dialogfeld Releasekonfiguration löschen auf Löschen.

Ausführungen mit Workflows und Cloud Scheduler planen

In diesem Abschnitt erfahren Sie, wie Sie mit Workflows und Cloud Scheduler Ausführungen von Dataform-Workflows planen.

Geplante Workflowausführungen

Sie können die Häufigkeit der Ausführung Ihres Dataform-Workflows festlegen, indem Sie einen Cloud Scheduler-Job erstellen, der einen Workflows-Workflow auslöst. Mit Workflows werden Dienste in einem von Ihnen definierten Orchestrierungsworkflow ausgeführt.

Mit Workflows wird Ihr Dataform-Workflow in zwei Schritten ausgeführt. Zuerst wird der Code Ihres Dataform-Repositorys von Ihrem Git-Anbieter abgerufen und in ein Kompilierungsergebnis kompiliert. Anschließend wird anhand des Kompilierungsergebnisses ein Dataform-Workflow erstellt und mit der von Ihnen festgelegten Häufigkeit ausgeführt.

Geplante Orchestrierungsabläufe erstellen

Wenn Sie Ausführungen Ihres Dataform-Workflows planen möchten, erstellen Sie mit Workflows einen Orchestration-Workflow und fügen Sie einen Cloud Scheduler-Job als Trigger hinzu.

  1. In Workflows werden Dienstkonten verwendet, um Workflows Zugriff aufGoogle Cloud -Ressourcen zu gewähren. Erstellen Sie ein Dienstkonto und gewähren Sie ihm die Identity and Access Management-Rolle Dataform Editor (roles/dataform.editor) sowie die Mindestberechtigungen, die für die Verwaltung des Orchestration-Workflows erforderlich sind. Weitere Informationen finden Sie unter Zugriff auf Google Cloud -Ressourcen für einen Workflow gewähren.

  2. Erstellen Sie einen Orchestration-Workflow und verwenden Sie den folgenden YAML-Quellcode als Workflow-Definition:

    main:
    steps:
    - init:
        assign:
        - repository: projects/PROJECT_ID/locations/REPOSITORY_LOCATION/repositories/REPOSITORY_ID
    - createCompilationResult:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1beta1/" + repository + "/compilationResults"}
            auth:
                type: OAuth2
            body:
                gitCommitish: GIT_COMMITISH
        result: compilationResult
    - createWorkflowInvocation:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1beta1/" + repository + "/workflowInvocations"}
            auth:
                type: OAuth2
            body:
                compilationResult: ${compilationResult.body.name}
        result: workflowInvocation
    - complete:
        return: ${workflowInvocation.body.name}

    Ersetzen Sie Folgendes:

    • PROJECT_ID: die ID Ihres Google Cloud Projekts.
    • REPOSITORY_LOCATION: der Speicherort Ihres Dataform-Repositories.
    • REPOSITORY_ID: der Name Ihres Dataform-Repositorys.
    • GIT_COMMITISH: der Git-Zweig, aus dem Sie Dataform-Code ausführen möchten. Ersetzen Sie bei einem neu erstellten Repository main.

  3. Orchestrierungsworkflow mit Cloud Scheduler planen

Anfrage zum Erstellen von Kompilierungsergebnissen im Dataform-Workflow anpassen

Sie können den vorhandenen Orchestration-Workflow aktualisieren und die Einstellungen für die Anfrage zum Erstellen des Kompilierungsergebnisses des Dataform-Workflows im YAML-Format definieren. Weitere Informationen zu den Einstellungen finden Sie in der projects.locations.repositories.compilationResults REST-Ressourcenreferenz.

Wenn Sie beispielsweise allen Aktionen während der Kompilierung eine _dev-schemaSuffix-Einstellung hinzufügen möchten, ersetzen Sie den createCompilationResult-Schrittkörper durch das folgende Code-Snippet:

    - createCompilationResult:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1beta1/" + repository + "/compilationResults"}
            auth:
                type: OAuth2
            body:
                gitCommitish: GIT_COMMITISH
                codeCompilationConfig:
                    schemaSuffix: dev

Sie können auch zusätzliche Einstellungen als Laufzeitargumente in einer Anfrage zur Workflowausführung übergeben und mithilfe von Variablen auf diese Argumente zugreifen. Weitere Informationen finden Sie unter Laufzeitargumente in einer Ausführungsanfrage übergeben.

Anfrage zum Aufrufen eines Dataform-Workflows anpassen

Sie können den vorhandenen Orchestration-Workflow aktualisieren und die Einstellungen für die Dataform-Workflow-Aufrufanfrage im YAML-Format definieren. Weitere Informationen zu den Einstellungen für Aufrufanfragen finden Sie in der projects.locations.repositories.workflowInvocations REST-Ressourcenreferenz.

Wenn Sie beispielsweise nur Aktionen mit dem hourly-Tag mit allen transitiven Abhängigkeiten ausführen möchten, ersetzen Sie den createWorkflowInvocation-Body durch das folgende Code-Snippet:

    - createWorkflowInvocation:
        call: http.post
        args:
            url: ${"https://dataform.googleapis.com/v1beta1/" + repository + "/workflowInvocations"}
            auth:
                type: OAuth2
            body:
                compilationResult: ${compilationResult.body.name}
                invocationConfig:
                    includedTags:
                    - hourly
                    transitiveDependenciesIncluded: true

Sie können auch zusätzliche Einstellungen als Laufzeitargumente in einer Anfrage zur Workflowausführung übergeben und mithilfe von Variablen auf diese Argumente zugreifen. Weitere Informationen finden Sie unter Laufzeitargumente in einer Ausführungsanfrage übergeben.

Ausführungen mit Cloud Composer planen

Sie können Cloud Composer 2 verwenden, um Dataform-Ausführungen zu planen. Dataform unterstützt Cloud Composer 1 nicht.

Wenn Sie Zeitpläne für Dataform-Ausführungen mit Cloud Composer 2 verwalten möchten, können Sie Dataform-Operatoren in Airflow-gerichteten azyklischen Graphen (Directed Acyclic Graphs, DAGs) verwenden. Sie können einen Airflow-DAG erstellen, mit dem Dataform-Workflowaufrufe geplant werden.

Dataform bietet verschiedene Airflow-Operatoren. Dazu gehören Operatoren zum Abrufen eines Kompilierungsergebnisses, zum Abrufen einer Workflowausführung und zum Abbrechen einer Workflowausführung. Eine vollständige Liste der verfügbaren Dataform Airflow-Operatoren finden Sie unter Google Dataform-Operatoren.

google-cloud-dataform-PyPI-Paket installieren

Wenn Sie Cloud Composer 2 Version 2.0.25 und höher verwenden, ist dieses Paket bereits in Ihrer Umgebung vorinstalliert. Sie müssen ihn also nicht installieren.

Wenn Sie ältere Versionen von Cloud Composer 2 verwenden, installieren Sie das PyPI-Paket google-cloud-dataform.

Geben Sie im Abschnitt „PyPI-Pakete“ die Version ==0.2.0 an.

Airflow-DAG erstellen, der Dataform-Workflowaufrufe plant

Wenn Sie geplante Ausführungen von Dataform-Workflows mit Cloud Composer 2 verwalten möchten, schreiben Sie den DAG mithilfe der Dataform Airflow-Operatoren und laden Sie ihn in den Bucket Ihrer Umgebung hoch.

Das folgende Codebeispiel zeigt einen Airflow-DAG, der ein Dataform-Kompilierungsergebnis erstellt und eine Dataform-Workflowausführung startet:

from datetime import datetime

from airflow import models
from airflow.models.baseoperator import chain
from airflow.providers.google.cloud.operators.dataform import (
    DataformCreateCompilationResultOperator,
    DataformCreateWorkflowInvocationOperator,
)

DAG_ID = "dataform"
PROJECT_ID = "PROJECT_ID"
REPOSITORY_ID = "REPOSITORY_ID"
REGION = "REGION"
GIT_COMMITISH = "GIT_COMMITISH"

with models.DAG(
    DAG_ID,
    schedule_interval='@once',  # Override to match your needs
    start_date=datetime(2022, 1, 1),
    catchup=False,  # Override to match your needs
    tags=['dataform'],
) as dag:

    create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": GIT_COMMITISH,
        },
    )
    create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
        task_id='create_workflow_invocation',
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
         workflow_invocation={
            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}"
        },
    )


create_compilation_result >> create_workflow_invocation

Ersetzen Sie Folgendes:

  • PROJECT_ID: Ihre Google Cloud-Projekt-ID für Dataform.
  • REPOSITORY_ID: der Name Ihres Dataform-Repositorys.
  • REGION: die Region, in der sich das Dataform-Repository befindet.
  • COMPILATION_RESULT: Der Name des Kompilierungsergebnisses, das Sie für diese Workflowausführung verwenden möchten.
  • GIT_COMMITISH: der Git-Commitish im Remote-Git-Repository der Version Ihres Codes, die Sie verwenden möchten, z. B. ein Branch oder ein Git-SHA.

Das folgende Codebeispiel zeigt einen Airflow-DAG, der Folgendes ausführt:

  1. Erstellt ein Dataform-Kompilierungsergebnis.
  2. Startet eine asynchrone Dataform-Workflowausführung.
  3. Der Status des Workflows wird mit DataformWorkflowInvocationStateSensor abgefragt, bis er den erwarteten Status erreicht.
from datetime import datetime

from google.cloud.dataform_v1beta1 import WorkflowInvocation

from airflow import models
from airflow.models.baseoperator import chain
from airflow.providers.google.cloud.operators.dataform import (
    DataformCreateCompilationResultOperator,
    DataformCreateWorkflowInvocationOperator,
)
from airflow.providers.google.cloud.sensors.dataform import DataformWorkflowInvocationStateSensor

DAG_ID = "dataform"
PROJECT_ID = "PROJECT_ID"
REPOSITORY_ID = "REPOSITORY_ID"
REGION = "REGION"
GIT_COMMITISH = "GIT_COMMITISH"

with models.DAG(
    DAG_ID,
    schedule_interval='@once',  # Override to match your needs
    start_date=datetime(2022, 1, 1),
    catchup=False,  # Override to match your needs
    tags=['dataform'],
) as dag:

    create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": GIT_COMMITISH,
        },
    )

create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
    task_id='create_workflow_invocation',
    project_id=PROJECT_ID,
    region=REGION,
    repository_id=REPOSITORY_ID,
    asynchronous=True,
    workflow_invocation={
        "compilation_result": COMPILATION_RESULT
    }
)

is_workflow_invocation_done = DataformWorkflowInvocationStateSensor(
    task_id="is_workflow_invocation_done",
    project_id=PROJECT_ID,
    region=REGION,
    repository_id=REPOSITORY_ID,
    workflow_invocation_id=("{{ task_instance.xcom_pull('create_workflow_invocation')['name'].split('/')[-1] }}"),
    expected_statuses={WorkflowInvocation.State.SUCCEEDED},
)


create_compilation_result >> create_workflow_invocation

Ersetzen Sie Folgendes:

  • PROJECT_ID: Ihre Dataform-Google Cloud-Projekt-ID.
  • REPOSITORY_ID: der Name Ihres Dataform-Repositorys.
  • REGION: die Region, in der sich das Dataform-Repository befindet.
  • COMPILATION_RESULT: Der Name des Kompilierungsergebnisses, das Sie für diese Workflowausführung verwenden möchten.
  • GIT_COMMITISH: der Git-Commitish im Remote-Git-Repository der Version Ihres Codes, die Sie verwenden möchten, z. B. ein Branch oder ein Git-SHA.
  • COMPILATION_RESULT: Der Name des Kompilierungsergebnisses, das Sie für diese Workflowausführung verwenden möchten.

Kompilierungskonfigurationsparameter hinzufügen

Sie können dem create_compilation_result Airflow-DAG-Objekt zusätzliche Kompilierungskonfigurationsparameter hinzufügen. Weitere Informationen zu den verfügbaren Parametern finden Sie in der CodeCompilationConfig Dataform API-Referenz.

  • Wenn Sie dem Airflow-DAG-Objekt create_compilation_result Konfigurationsparameter für die Kompilierung hinzufügen möchten, fügen Sie die ausgewählten Parameter im folgenden Format in das Feld code_compilation_config ein:

        create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": GIT_COMMITISH,
            "code_compilation_config": { "PARAMETER": "PARAMETER_VALUE"}
        },
    )

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Google Cloud-Projekt-ID für Dataform.
    • REPOSITORY_ID: der Name Ihres Dataform-Repositorys.
    • REGION: die Region, in der sich das Dataform-Repository befindet.
    • GIT_COMMITISH: der Git-Commitish im Remote-Git-Repository der Version Ihres Codes, die Sie verwenden möchten, z. B. ein Branch oder ein Git-SHA.
    • PARAMETER: der ausgewählte CodeCompilationConfig-Parameter. Sie können mehrere Parameter hinzufügen.
    • PARAMETER_VALUE: den Wert des ausgewählten Parameters.

Das folgende Codebeispiel zeigt den Parameter defaultDatabase, der dem Airflow-DAG-Objekt create_compilation_result hinzugefügt wurde:

    create_compilation_result = DataformCreateCompilationResultOperator(
        task_id="create_compilation_result",
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        compilation_result={
            "git_commitish": REMOTE_BRANCH,
            "code_compilation_config": { "default_database": "my-custom-gcp-project"}
        },
    )

Konfigurationsparameter für die Workflowauslösung hinzufügen

Sie können dem create_workflow_invocation Airflow-DAG-Objekt zusätzliche Konfigurationsparameter für die Workflowausführung hinzufügen. Weitere Informationen zu den verfügbaren Parametern finden Sie in der InvocationConfig Dataform API-Referenz.

  • Wenn Sie dem create_workflow_invocation-Airflow-DAG-Objekt Konfigurationsparameter für die Workflowausführung hinzufügen möchten, fügen Sie die ausgewählten Parameter im folgenden Format in das Feld invocation_config ein:

        create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
        task_id='create_workflow_invocation',
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        workflow_invocation={
            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}",
            "invocation_config": { "PARAMETER": PARAMETER_VALUE }
        },
    )

    Ersetzen Sie Folgendes:

    • PROJECT_ID: Ihre Google Cloud-Projekt-ID für Dataform.
    • REPOSITORY_ID: der Name Ihres Dataform-Repositorys.
    • REGION: die Region, in der sich das Dataform-Repository befindet.
    • PARAMETER: der ausgewählte InvocationConfig-Parameter. Sie können mehrere Parameter hinzufügen.
    • PARAMETER_VALUE: den Wert des ausgewählten Parameters.

Das folgende Codebeispiel zeigt die Parameter includedTags[] und transitiveDependenciesIncluded, die dem Airflow-DAG-Objekt create_workflow_invocation hinzugefügt wurden:

    create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
        task_id='create_workflow_invocation',
        project_id=PROJECT_ID,
        region=REGION,
        repository_id=REPOSITORY_ID,
        workflow_invocation={
            "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}",
            "invocation_config": { "included_tags": ["daily"], "transitive_dependencies_included": true }
        },
    )

Nächste Schritte