Connector für Dataflow

Workflow-Connector, der die integrierte Funktion definiert, mit der in einem Workflow auf Dataflow zugegriffen wird.

Weitere Informationen

Eine ausführliche Dokumentation, die dieses Codebeispiel enthält, finden Sie hier:

• Dataflow API-Connector-Übersicht

Codebeispiel

YAML

# This workflow demonstrates how to use the Cloud Dataflow connector.
# The workflow creates a word count job using a Dataflow public job template
# and uses a Cloud Storage bucket as temporary storage for temp files.
# The bucket resource is deleted after the job completes.
# Expected successful output: "SUCCESS"

- init:
    assign:
      - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
      - location: "us-central1"
      - zone: "us-central1-a"
      - bucket_name: "[fill in a bucket name]"
      - job_name: "[fill in a job name]"
      - input_file: "gs://dataflow-samples/shakespeare/kinglear.txt"
      - output_storage_file_prefix: ${"gs://" + bucket_name + "/counts"}
      - temp_location: ${"gs://" + bucket_name + "/counts/temp"}
      - template_path: "gs://dataflow-templates-us-central1/latest/Word_Count"
- create_bucket:
    call: googleapis.storage.v1.buckets.insert
    args:
      project: ${project_id}
      body:
        name: ${bucket_name}
- create_job:
    call: googleapis.dataflow.v1b3.projects.locations.templates.create
    args:
      projectId: ${project_id}
      location: ${location}
      body:
        jobName: ${job_name}
        parameters:
          inputFile: ${input_file}
          output: ${output_storage_file_prefix}
        environment:
          numWorkers: 1
          maxWorkers: 2
          zone: ${zone}
          tempLocation: ${temp_location}
        gcsPath: ${template_path}
- delete_bucket_object1:
    call: googleapis.storage.v1.objects.delete
    args:
      bucket: ${bucket_name}
      object: ${"counts-00000-of-00003"}
- delete_bucket_object2:
    call: googleapis.storage.v1.objects.delete
    args:
      bucket: ${bucket_name}
      object: ${"counts-00001-of-00003"}
- delete_bucket_object3:
    call: googleapis.storage.v1.objects.delete
    args:
      bucket: ${bucket_name}
      object: ${"counts-00002-of-00003"}
- delete_bucket:
    call: googleapis.storage.v1.buckets.delete
    args:
      bucket: ${bucket_name}
- the_end:
    return: "SUCCESS"

Nächste Schritte

Wenn Sie nach Codebeispielen für andere Google Cloud -Produkte suchen und filtern möchten, können Sie den Google Cloud -Beispielbrowser verwenden.