Ein deklarativer Ansatz für Dataflow-Flex-Vorlagen

Google Cloud bietet eine verwaltete Apache Beam-Lösung namens Dataflow an. Seit einiger Zeit verfügt Dataflow über eine Funktion namens Flex Templates. Flex Templates verwenden Docker-Container anstelle der benutzerdefinierten Vorlagen von Dataflow. Der Vorteil ist, dass Docker ein bekannter Standard ist und der Container in verschiedenen Umgebungen ausgeführt werden kann. Allerdings wird immer noch eine benutzerdefinierte Metadaten-JSON-Datei benötigt, die auf das Docker-Image in Ihrer Registrierung verweist.

Sowohl bei der CLI- als auch bei der Terraform-Methode müssen Sie das Docker-Image in eine Registry stellen.

Verwendung der gcloud CLI

Um die JSON-Datei zu erzeugen und hochzuladen, können Sie gcloud dataflow flex-template build ausführen. Die Eingabe für diesen Befehl ist ein Stück JSON, das den Pipeline-Namen und die Parameter definiert:

{
    "name": "My Apache Beam pipeline",
    "parameters": [
        {
            "name": "output_table",
            "label": "BigQuery output table name.",
            "helpText": "Name of the BigQuery output table name.",
            "regexes": ["([^:]+:)?[^.]+[.].+"]
        }
    ]
}

Lassen Sie uns sehen, was sich nach der Ausführung des Befehls in Cloud Storage befindet:

{
    "image": "eu-docker.pkg.dev/my-gcp-project-id/dataflow-templates/example:latest",
    "sdkInfo": {
        "language": "PYTHON"
    },
    "metadata": {
        "name": "My Apache Beam pipeline",
        "parameters": [
            {
                "name": "output_table",
                "label": "BigQuery output table name.",
                "helpText": "Name of the BigQuery output table name.",
                "regexes": ["([^:]+:)?[^.]+[.].+"]
            }
        ]
    }
}

Der Befehl fügt nur den Speicherort des Docker-Images und einige Beam SDK hinzu, bevor er es in den Cloud-Speicher hochlädt.

Terraform verwenden

Das funktioniert zwar gut, widerspricht aber dem deklarativen Ansatz von Terraform und anderen Infrastruktur-als-Code-Tools. Sehen wir uns an, was nötig ist, um diese JSON-Datei mit Metadaten in Terraform zu erzeugen und zu verwalten:

resource "google_storage_bucket_object" "flex_template_metadata" {
    bucket       = "my-unique-bucket"
    name         = "dataflow-templates/example/metadata.json"
    content_type = "application/json"

    content = jsonencode({
        image = "eu-docker.pkg.dev/my-gcp-project-id/dataflow-templates/example:latest"
        sdkInfo = {
            language = "PYTHON"
        }
        metadata = {
            name = "My Apache Beam pipeline"
            parameters = [
                {
                    name = "output_table"
                    label = "BigQuery output table name."
                    helpText = "Name of the BigQuery output table name.",
                    regexes = ["([^:]+:)?[^.]+[.].+"]
                }
            ]
        }
    })
}

Wir können den Pfad der Speicherdatei in unserem Flex-Vorlagenauftrag referenzieren:

resource "google_dataflow_flex_template_job" "flex_template_job" {
    provider = google-beta

    name                    = "example_pipeline"
    region                  = "europe-west4"
    container_spec_gcs_path = "gs://${google_storage_bucket_object.flex_template_metadata.bucket}/${google_storage_bucket_object.flex_template_metadata.name}"

    parameters = {
        output_table = "my-gcp-project-id/example_dataset/example_table"
    }
}

Führen Sie terraform apply aus, um sowohl die Vorlagendatei als auch den Dataflow-Auftrag zu erstellen.

Aktualisieren des Dataflow-Auftrags

Wir haben noch ein Problem. Eine Änderung der Vorlagendaten löst keine Aktualisierung des Dataflow-Jobs aus. Damit dies funktioniert, muss sich ein Attribut der Dataflow-Job-Ressource ändern. Wir können dies erreichen, indem wir einen MD5-Hash des Dateiinhalts in den Speicherpfad aufnehmen:

locals {
    template_content = jsonencode({
        image = "eu-docker.pkg.dev/my-gcp-project-id/dataflow-templates/example:latest"
        sdkInfo = {
            language = "PYTHON"
        }
        metadata = {
            name = "My Apache Beam pipeline"
            parameters = [
                {
                    name = "output_table"
                    label = "BigQuery output table name."
                    helpText = "Name of the BigQuery output table name.",
                    regexes = ["([^:]+:)?[^.]+[.].+"]
                }
            ]
        }
    })
    template_gcs_path = "dataflow-templates/example/${base64encode(md5(local.template_content))}/metadata.json"
}

resource "google_storage_bucket_object" "flex_template_metadata" {
    bucket = "my-unique-bucket"
    name   = local.template_gcs_path
    ...
}

Eine Änderung der Vorlagendaten führt zu einer Änderung des MD5-Hash. Dies führt zu einer Änderung des Speicherpfads der Vorlage, auf den wir auch in der Dataflow-Job-Ressource verweisen. Die Ausführung von terraform apply aktualisiert nun sowohl die JSON-Daten im Speicher als auch den Dataflow-Flex-Job korrekt. Wenn Sie einen Batch-Modus verwenden, wird eine neue Job-Instanz erstellt.

Fazit

Mit Dataflow Flex Templates können Sie Docker-Images für Dataflow-Jobs verwenden. Ein gcloud CLI-Befehl ist erforderlich, um einige JSON-Metadaten zu erstellen und hochzuladen. Wir können dieses Verhalten mit Terraform-Code replizieren. Dies ermöglicht eine 100% deklarative Infrastruktur-als-Code-Lösung.

Ein vollständiges Codebeispiel finden Sie in meinem Dataflow Flex Terraform-Modul auf GitHub.