Configurare i modelli flessibili

Questa pagina documenta varie opzioni di configurazione dei modelli flessibili Dataflow, tra cui:

• Autorizzazioni
• Variabili di ambiente Dockerfile
• Dipendenze dei pacchetti per Python
• Immagini Docker
• Opzioni pipeline
• Località temporanee e di staging

Per configurare un modello flessibile di esempio, consulta il tutorial sui modelli flessibili.

Informazioni sulle autorizzazioni dei modelli flessibili

Quando lavori con i modelli flessibili, hai bisogno di tre set di autorizzazioni:

• Autorizzazioni per creare risorse
• Autorizzazioni per creare un modello flessibile
• Autorizzazioni per eseguire un modello flessibile

Autorizzazioni per creare risorse

Per sviluppare ed eseguire una pipeline di modelli flessibili, devi creare varie risorse (ad esempio, un bucket di staging). Per le attività di creazione di risorse una tantum, puoi utilizzare il ruolo Proprietario di base.

Autorizzazioni per creare un modello flessibile

In qualità di sviluppatore di un modello flessibile, devi creare il modello per renderlo disponibile agli utenti. La creazione prevede il caricamento di una specifica del modello in un bucket Cloud Storage e il provisioning di un'immagine Docker con il codice e le dipendenze necessari per eseguire la pipeline. Per creare un modello flessibile, devi disporre dell'accesso in lettura e scrittura a Cloud Storage e dell'accesso in scrittura ad Artifact Registry al tuo repository Artifact Registry. Puoi concedere queste autorizzazioni assegnando i seguenti ruoli:

• Amministratore Storage (roles/storage.admin)
• Editor Cloud Build (roles/cloudbuild.builds.editor)
• Artifact Registry Writer (roles/artifactregistry.writer)

Autorizzazioni per eseguire un modello flessibile

Quando esegui un modello flessibile, Dataflow crea un job per te. Per creare il job, l'account di servizio Dataflow deve disporre della seguente autorizzazione:

• dataflow.serviceAgent

Quando utilizzi Dataflow per la prima volta, il servizio assegna questo ruolo per te, quindi non devi concedere questa autorizzazione.

Per impostazione predefinita, il account di servizio Compute Engine viene utilizzato per le VM di avvio e le VM worker. Il account di servizio deve avere i seguenti ruoli e capacità:

• Storage Object Admin (roles/storage.objectAdmin)
• Visualizzatore (roles/viewer)
• Dataflow Worker (roles/dataflow.worker)
• Accesso in lettura e scrittura al bucket di staging
• Accesso in lettura all'immagine del modello flessibile

Per concedere l'accesso in lettura e scrittura al bucket di staging, puoi utilizzare il ruolo Amministratore oggetti Storage (roles/storage.objectAdmin). Per saperne di più, consulta Ruoli IAM per Cloud Storage.

Per concedere l'accesso in lettura all'immagine del modello flessibile, puoi utilizzare il ruolo Storage Object Viewer (roles/storage.objectViewer). Per saperne di più, vedi Configurazione del controllo dell'accesso.

Imposta le variabili di ambiente Dockerfile richieste

Se vuoi creare il tuo Dockerfile per un job Flex Template, specifica le seguenti variabili di ambiente:

ENV FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE="${WORKDIR}/requirements.txt"
ENV FLEX_TEMPLATE_PYTHON_PY_FILE="${WORKDIR}/streaming_beam.py"

Dipendenze dei pacchetti per Python

Quando una pipeline Dataflow Python utilizza dipendenze aggiuntive, potresti dover configurare il modello flessibile per installare dipendenze aggiuntive sulle VM worker Dataflow.

Quando esegui un job Dataflow Python che utilizza i modelli flessibili in un ambiente che limita l'accesso a internet, devi precompilare le dipendenze quando crei il modello.

Utilizza una delle seguenti opzioni per precompilare le dipendenze Python.

Per istruzioni sulla gestione delle dipendenze delle pipeline in Java e Go, consulta Gestire le dipendenze delle pipeline in Dataflow.

Utilizza un file requirements e precompila le dipendenze con il modello

Se utilizzi il tuo Dockerfile per definire l'immagine del modello flessibile, segui questi passaggi:

1. Crea un file requirements.txt che elenca le dipendenze della pipeline.

   COPY requirements.txt /template/
   ENV FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE="/template/requirements.txt"
   

2. Installa le dipendenze nell'immagine del modello flessibile.

   RUN pip install --no-cache-dir -r $FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE
   

3. Scarica le dipendenze nella cache dei requisiti locali, che viene trasferita ai worker Dataflow all'avvio del modello.

   RUN pip download --no-cache-dir --dest /tmp/dataflow-requirements-cache -r $FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE
   

Quando utilizzi questo approccio, le dipendenze del file requirements.txt vengono installate sui worker Dataflow in fase di runtime. Un approfondimento nella scheda dei consigli della console Google Cloud potrebbe segnalare questo comportamento. Per evitare di installare le dipendenze in fase di runtime, utilizza un'immagine container personalizzata.

Di seguito è riportato un esempio di codice che utilizza un file requirements nel modello Flex.

# Copyright 2020 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

FROM gcr.io/dataflow-templates-base/python3-template-launcher-base

# Configure the Template to launch the pipeline with a --requirements_file option.
# See: https://beam.apache.org/documentation/sdks/python-pipeline-dependencies/#pypi-dependencies
ENV FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE="/template/requirements.txt"
ENV FLEX_TEMPLATE_PYTHON_PY_FILE="/template/streaming_beam.py"

COPY . /template

RUN apt-get update \
    # Install any apt packages if required by your template pipeline.
    && apt-get install -y libffi-dev git \
    && rm -rf /var/lib/apt/lists/* \
    # Upgrade pip and install the requirements.
    && pip install --no-cache-dir --upgrade pip \
    # Install dependencies from requirements file in the launch environment.
    && pip install --no-cache-dir -r $FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE \
    # When FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE  option is used,
    # then during Template launch Beam downloads dependencies
    # into a local requirements cache folder and stages the cache to workers.
    # To speed up Flex Template launch, pre-download the requirements cache
    # when creating the Template.
    && pip download --no-cache-dir --dest /tmp/dataflow-requirements-cache -r $FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE

# Set this if using Beam 2.37.0 or earlier SDK to speed up job submission.
ENV PIP_NO_DEPS=True

ENTRYPOINT ["/opt/google/dataflow/python_template_launcher"]

Strutturare la pipeline come pacchetto e utilizzare pacchetti locali

Quando utilizzi più file o moduli Python locali, struttura la pipeline come pacchetto. La struttura del file potrebbe essere simile a quella dell'esempio seguente:

main.py
pyproject.toml
setup.py
src/
  my_package/
    __init__.py
    my_custom_dofns_and_transforms.py
    my_pipeline_launcher.py
    other_utils_and_helpers.py

1. Posiziona il punto di ingresso di primo livello, ad esempio il file main.py, nella directory principale. Inserisci il resto dei file in una cartella separata nella directory src, ad esempio my_package.

2. Aggiungi i file di configurazione del pacchetto alla directory principale con i dettagli e i requisiti del pacchetto.

   pyproject.toml

   [project]
   name = "my_package"
   version = "package_version"
   dependencies = [
     # Add list of packages (and versions) that my_package depends on.
     # Example:
     "apache-beam[gcp]==2.54.0",
   ]
   

   setup.py

     """An optional setuptools configuration stub for the pipeline package.
   
     Use pyproject.toml to define the package. Add this file only if you must
     use the --setup_file pipeline option or the
     FLEX_TEMPLATE_PYTHON_SETUP_FILE configuration option.
     """
   
     import setuptools
     setuptools.setup()
   

   Per ulteriori informazioni su come configurare il pacchetto locale, consulta Packaging Python Projects.

3. Quando importi moduli o file locali per la pipeline, utilizza il nome del pacchetto my_package come percorso di importazione.

   from my_package import word_count_transform
   

4. Installa il pacchetto della pipeline nell'immagine del modello flessibile. Il Dockerfile del modello flessibile potrebbe includere contenuti simili a quelli dell'esempio seguente:

   Dockerfile

   ENV FLEX_TEMPLATE_PYTHON_PY_FILE="${WORKDIR}/main.py"
   ENV FLEX_TEMPLATE_PYTHON_SETUP_FILE="${WORKDIR}/setup.py"
   
   # Copy pipeline, packages and requirements.
   WORKDIR ${WORKDIR}
   COPY main.py .
   COPY pyproject.toml .
   COPY setup.py .
   COPY src src
   
   # Install local package.
   RUN pip install -e .
   

Quando utilizzi questo approccio, le dipendenze del file requirements.txt vengono installate sui worker Dataflow in fase di runtime. Un approfondimento nella scheda dei consigli della console Google Cloud potrebbe segnalare questo comportamento. Per evitare di installare le dipendenze in fase di runtime, utilizza un'immagine container personalizzata.

Per un esempio che segue l'approccio consigliato, consulta l'esercitazione Modello flessibile per una pipeline con dipendenze e un'immagine container personalizzata su GitHub.

Utilizza un container personalizzato che preinstalli tutte le dipendenze

Per evitare l'installazione delle dipendenze in fase di runtime, utilizza container personalizzati. Questa opzione è preferibile per le pipeline eseguite in ambienti senza accesso a internet.

Per utilizzare un container personalizzato:

1. Crea un'immagine container personalizzata che preinstalli le dipendenze necessarie.

2. Preinstalla le stesse dipendenze nel Dockerfile del modello flessibile.

   Per impedire l'installazione delle dipendenze in fase di runtime, non utilizzare le opzioni FLEX_TEMPLATE_PYTHON_REQUIREMENTS_FILE o FLEX_TEMPLATE_PYTHON_SETUP_FILE nella configurazione del modello flessibile.

   Un modello flessibile Dockerfile modificato potrebbe avere il seguente aspetto:

   FROM gcr.io/dataflow-templates-base/python3-template-launcher-base
   ENV FLEX_TEMPLATE_PYTHON_PY_FILE="/template/main.py"
   COPY . /template
   # If you use a requirements file, pre-install the requirements.txt.
   RUN pip install --no-cache-dir -r /template/requirements.txt
   # If you supply the pipeline in a package, pre-install the local package and its dependencies.
   RUN pip install -e /template
   

   Quando utilizzi questo approccio, esegui le seguenti operazioni:

   • crea l'immagine del modello flessibile
   • crea l'immagine container dell'SDK personalizzato
   • installare le stesse dipendenze in entrambe le immagini

   In alternativa, per ridurre il numero di immagini da gestire, utilizza la tua immagine container personalizzata come immagine di base per il modello flessibile.

3. Se utilizzi l'SDK Apache Beam versione 2.49.0 o precedente, aggiungi l'opzione pipeline --sdk_location=container nel launcher della pipeline. Questa opzione indica alla pipeline di utilizzare l'SDK dal container personalizzato anziché scaricarlo.

   options = PipelineOptions(beam_args, save_main_session=True, streaming=True, sdk_location="container")
   

4. Imposta il parametro sdk_container_image nel comando flex-template run. Ad esempio:

   gcloud dataflow flex-template run $JOB_NAME \
      --region=$REGION \
      --template-file-gcs-location=$TEMPLATE_PATH \
      --parameters=sdk_container_image=$CUSTOM_CONTAINER_IMAGE \
      --additional-experiments=use_runner_v2
   

   Per ulteriori informazioni, consulta Utilizzare i container personalizzati in Dataflow.

Scegliere un'immagine di base

Puoi utilizzare un'immagine di base fornita da Google per creare pacchetti di immagini container del modello utilizzando Docker. Scegli il tag più recente da Immagini di base per i modelli flessibili. Ti consigliamo di utilizzare un tag immagine specifico anziché latest. Le immagini di base sono ospitate all'indirizzo gcr.io/dataflow-templates-base.

Specifica l'immagine di base nel seguente formato:

gcr.io/dataflow-templates-base/IMAGE_NAME:TAG

Sostituisci quanto segue:

• IMAGE_NAME: un'immagine di base fornita da Google
• TAG: il nome di una versione dell'immagine di base, che si trova nel riferimento alle immagini di base dei modelli flessibili

Utilizzare immagini container personalizzate

Se la tua pipeline utilizza un'immagine container personalizzata, ti consigliamo di utilizzare l'immagine personalizzata come immagine di base per l'immagine Docker del modello flessibile. Per farlo, copia il binario di avvio del modello flessibile dall'immagine di base del modello fornita da Google sull'immagine personalizzata.

Un esempio di Dockerfile per un'immagine che può essere utilizzata sia come immagine container SDK personalizzata sia come modello flessibile, potrebbe avere il seguente aspetto:

FROM gcr.io/dataflow-templates-base/IMAGE_NAME:TAG as template_launcher
FROM apache/beam_python3.10_sdk:2.66.0

# RUN <...Make image customizations here...>
# See: https://cloud.google.com/dataflow/docs/guides/build-container-image

# Configure the Flex Template here.
COPY --from=template_launcher /opt/google/dataflow/python_template_launcher /opt/google/dataflow/python_template_launcher
COPY my_pipeline.py /template/
ENV FLEX_TEMPLATE_PYTHON_PY_FILE="/template/my_pipeline.py"

Sostituisci quanto segue:

• IMAGE_NAME: un'immagine di base fornita da Google. Ad esempio: python311-template-launcher-base.
• TAG: un tag di versione per l'immagine di base trovata nel riferimento Immagini di base per i modelli flessibili. Per una migliore stabilità e risoluzione dei problemi, evita di utilizzare latest. Blocca invece un tag di versione specifico.

Per un esempio che segue questo approccio, consulta il tutorial Modello flessibile per una pipeline con dipendenze e un'immagine container personalizzata.

Utilizzare un'immagine di un registro privato

Puoi creare un'immagine del modello flessibile archiviata in un registro Docker privato, se il registro privato utilizza HTTPS e ha un certificato valido.

Per utilizzare un'immagine da un registro privato, specifica il percorso dell'immagine e un nome utente e una password per il registro. Il nome utente e la password devono essere archiviati in Secret Manager. Puoi fornire il segreto in uno dei seguenti formati:

• projects/{project}/secrets/{secret}/versions/{secret_version}
• projects/{project}/secrets/{secret}

Se utilizzi il secondo formato, poiché non specifica la versione, Dataflow utilizza l'ultima versione.

Se il registro utilizza un certificato autofirmato, devi anche specificare il percorso del certificato autofirmato in Cloud Storage.

La tabella seguente descrive le opzioni di gcloud CLI che puoi utilizzare per configurare un registro privato.

Ecco un esempio di comando Google Cloud CLI che crea un modello flessibile utilizzando un'immagine in un registro privato con un certificato autofirmato.

gcloud dataflow flex-template build gs://example-bucket/custom-pipeline-private-repo.json
--sdk-language=JAVA
--image="gcp.repository.example.com:9082/registry/example/image:latest"
--image-repository-username-secret-id="projects/example-project/secrets/username-secret"
--image-repository-password-secret-id="projects/example-project/secrets/password-secret/versions/latest"
--image-repository-cert-path="gs://example-bucket/self-signed.crt"
--metadata-file=metadata.json

Per creare un modello flessibile, devi sostituire i valori di esempio e potresti dover specificare opzioni diverse o aggiuntive. Per saperne di più, consulta le seguenti risorse:

• gcloud dataflow flex-template build
• Utilizzo dei modelli flessibili

Specifica le opzioni della pipeline

Per informazioni sulle opzioni della pipeline supportate direttamente dai modelli flessibili, vedi Opzioni della pipeline.

Puoi anche utilizzare indirettamente qualsiasi opzione della pipeline Apache Beam. Se utilizzi un file metadata.json per il job del modello flessibile, includi queste opzioni della pipeline nel file. Questo file di metadati deve seguire il formato indicato in TemplateMetadata.

In caso contrario, quando avvii il job del modello flessibile, passa queste opzioni della pipeline utilizzando il campo dei parametri.

Per le opzioni sperimentali di runtime o le opzioni comuni della pipeline di runtime che non sono esposte esplicitamente nei metadati del modello, passa queste opzioni utilizzando i campi additional-experiments o additional-pipeline-options.

{
  "jobName": "my-flex-template-job",
  "parameters": {
    "option_defined_in_metadata": "value"
  },
  "environment": {
    "additionalExperiments": [
      "use_runner_v2"
    ],
    "additionalPipelineOptions": {
      "common_pipeline_option": "value"
    }
  }
}

Quando utilizzi i modelli flessibili, puoi configurare alcune opzioni della pipeline durante l'inizializzazione, ma altre non possono essere modificate. Se gli argomenti della riga di comando richiesti dal modello flessibile vengono sovrascritti, il job potrebbe ignorare, sostituire o eliminare le opzioni della pipeline trasmesse dal launcher del modello. Il job potrebbe non essere avviato o potrebbe essere avviato un job che non utilizza il modello flessibile. Per saperne di più, vedi Impossibile leggere il file del job.

Durante l'inizializzazione della pipeline, non modificare le seguenti opzioni della pipeline:

Blocca le chiavi SSH del progetto dalle VM che utilizzano chiavi SSH basate su metadati

Puoi impedire alle VM di accettare le chiavi SSH archiviate nei metadati del progetto bloccando le chiavi SSH del progetto dalle VM. Utilizza il flag additional-experiments con l'opzione di servizio block_project_ssh_keys:

--additional-experiments=block_project_ssh_keys

Per ulteriori informazioni, consulta Opzioni del servizio Dataflow.

Metadati

Puoi estendere il modello con metadati aggiuntivi in modo che i parametri personalizzati vengano convalidati quando viene eseguito il modello. Se vuoi creare metadati per il tuo modello, segui questi passaggi:

1. Crea un file metadata.json utilizzando i parametri in Parametri dei metadati.

   Per visualizzare un esempio, consulta File di metadati di esempio.

2. Archivia il file di metadati in Cloud Storage nella stessa cartella del modello.

{
  "name": "Streaming Beam SQL",
  "description": "An Apache Beam streaming pipeline that reads JSON encoded messages from Pub/Sub, uses Beam SQL to transform the message data, and writes the results to a BigQuery",
  "parameters": [
    {
      "name": "inputSubscription",
      "label": "Pub/Sub input subscription.",
      "helpText": "Pub/Sub subscription to read from.",
      "regexes": [
        "[a-zA-Z][-_.~+%a-zA-Z0-9]{2,}"
      ]
    },
    {
      "name": "outputTable",
      "label": "BigQuery output table",
      "helpText": "BigQuery table spec to write to, in the form 'project:dataset.table'.",
      "isOptional": true,
      "regexes": [
        "[^:]+:[^.]+[.].+"
      ]
    }
  ]
}

{
  "name": "Streaming beam Python flex template",
  "description": "Streaming beam example for python flex template.",
  "parameters": [
    {
      "name": "input_subscription",
      "label": "Input PubSub subscription.",
      "helpText": "Name of the input PubSub subscription to consume from.",
      "regexes": [
        "projects/[^/]+/subscriptions/[a-zA-Z][-_.~+%a-zA-Z0-9]{2,}"
      ]
    },
    {
      "name": "output_table",
      "label": "BigQuery output table name.",
      "helpText": "Name of the BigQuery output table name.",
      "isOptional": true,
      "regexes": [
        "([^:]+:)?[^.]+[.].+"
      ]
    }
  ]
}

Informazioni sulla posizione di staging e sulla posizione temporanea

Google Cloud CLI fornisce le opzioni --staging-location e --temp-location quando esegui un modello flessibile. Analogamente, l'API REST Dataflow fornisce i campi stagingLocation e tempLocation per FlexTemplateRuntimeEnvironment.

Per i modelli flessibili, la posizione di gestione temporanea è l'URL di Cloud Storage in cui vengono scritti i file durante il passaggio di gestione temporanea dell'avvio di un modello. Dataflow legge questi file di staging per creare il grafico del modello. La posizione temporanea è l'URL di Cloud Storage in cui vengono scritti i file temporanei durante il passaggio di esecuzione.

Aggiornare un job modello flessibile

La seguente richiesta di esempio mostra come aggiornare un job di streaming del modello utilizzando il metodo projects.locations.flexTemplates.launch. Se vuoi utilizzare gcloud CLI, consulta Aggiornare una pipeline esistente.

Se vuoi aggiornare un modello classico, utilizza projects.locations.templates.launch in alternativa.

1. Segui i passaggi per creare un job di streaming da un modello flessibile. Invia la seguente richiesta HTTP POST con i valori modificati:

   POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/flexTemplates:launch
   {
       "launchParameter": {
         "update": true
         "jobName": "JOB_NAME",
         "parameters": {
           "input_subscription": "projects/PROJECT_ID/subscriptions/SUBSCRIPTION_NAME",
           "output_table": "PROJECT_ID:DATASET.TABLE_NAME"
         },
       "containerSpecGcsPath": "STORAGE_PATH"
       },
   }
   

   • Sostituisci PROJECT_ID con l'ID progetto.
   • Sostituisci REGION con la regione Dataflow del job che stai aggiornando.
   • Sostituisci JOB_NAME con il nome esatto del job che vuoi aggiornare.
   • Imposta parameters sull'elenco di coppie chiave-valore. I parametri elencati sono specifici di questo esempio di modello. Se utilizzi un modello personalizzato, modifica i parametri in base alle esigenze. Se utilizzi il modello di esempio, sostituisci le seguenti variabili.
     • Sostituisci SUBSCRIPTION_NAME con il nome dell'abbonamento Pub/Sub.
     • Sostituisci DATASET con il nome del tuo set di dati BigQuery.
     • Sostituisci TABLE_NAME con il nome della tabella BigQuery.
   • Sostituisci STORAGE_PATH con la posizione Cloud Storage del file modello. La posizione deve iniziare con gs://.

2. Utilizza il parametro environment per modificare le impostazioni dell'ambiente. Per ulteriori informazioni, vedi FlexTemplateRuntimeEnvironment.

3. (Facoltativo) Per inviare la richiesta utilizzando curl (Linux, macOS o Cloud Shell), salva la richiesta in un file JSON, quindi esegui il seguente comando:

   curl -X POST -d "@FILE_PATH" -H "Content-Type: application/json" -H "Authorization: Bearer $(gcloud auth print-access-token)"  https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/REGION/flexTemplates:launch
   

   Sostituisci FILE_PATH con il percorso del file JSON contenente il corpo della richiesta.

4. Utilizza l'interfaccia di monitoraggio di Dataflow per verificare che sia stato creato un nuovo job con lo stesso nome. Questo lavoro ha lo stato Aggiornato.

Limitazioni

Ai job Flex Template si applicano le seguenti limitazioni:

• Per creare pacchetti dei container utilizzando Docker, devi utilizzare un'immagine di base fornita da Google. Per un elenco delle immagini applicabili, consulta Immagini di base per i modelli flessibili.
• Il programma che crea la pipeline deve uscire dopo la chiamata di run per avviare la pipeline.
• waitUntilFinish (Java) e wait_until_finish (Python) non sono supportati.

Passaggi successivi

• Per saperne di più sui modelli classici, sui modelli flessibili e sui relativi scenari di utilizzo, consulta Modelli Dataflow.
• Per informazioni sulla risoluzione dei problemi relativi ai modelli flessibili, consulta l'articolo Risolvere i problemi di timeout dei modelli flessibili.
• Per ulteriori architetture di riferimento, diagrammi e best practice, esplora il Cloud Architecture Center.