Esegui un job Dataflow in un container personalizzato

L'esecuzione di job Dataflow in container personalizzati consente di personalizzare l'ambiente di runtime dei worker e di preinstallare dipendenze di terze parti. Questo approccio contribuisce a garantire un'esecuzione coerente della pipeline nei tuoi workload.

Per istruzioni sulla creazione di un'immagine, consulta Crea immagini container personalizzate per Dataflow. Per evitare errori dovuti a dipendenze incompatibili quando avvii la pipeline, utilizza una versione dell'SDK Apache Beam e una versione del linguaggio che corrispondano all'SDK nell'immagine container personalizzata.

Esegui test in locale

Prima di eseguire la pipeline in Dataflow, è consigliabile testare l'immagine container localmente, in modo da velocizzare i test e il debug.

Per scoprire di più sull'utilizzo specifico di Apache Beam, consulta la guida di Apache Beam per l'esecuzione di pipeline con immagini container personalizzate.

Test di base con PortableRunner

Per verificare che le immagini container remote possano essere estratte e possano eseguire una pipeline di base, utilizza l'PortableRunner di Apache Beam. Quando utilizzi PortableRunner, l'invio dei job avviene nell'ambiente locale e l'esecuzione di DoFn avviene nell'ambiente Docker.

Quando utilizzi le GPU, il container Docker potrebbe non avere accesso alle GPU. Per testare il container con le GPU, utilizza il runner diretto e segui i passaggi per testare un'immagine container su una VM autonoma con GPU nella sezione Esegui il debug con una VM autonoma della pagina "Utilizza le GPU".

Di seguito viene eseguita una pipeline di esempio:

Java

mvn compile exec:java -Dexec.mainClass=com.example.package.MyClassWithMain \
    -Dexec.args="--runner=PortableRunner \
    --jobEndpoint=REGION \
    --defaultEnvironmentType=DOCKER \
    --defaultEnvironmentConfig=IMAGE_URI \
    --inputFile=INPUT_FILE \
    --output=OUTPUT_FILE"

Python

python path/to/my/pipeline.py \
  --runner=PortableRunner \
  --job_endpoint=REGION \
  --environment_type=DOCKER \
  --environment_config=IMAGE_URI \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE

Vai

go path/to/my/pipeline.go \
  --runner=PortableRunner \
  --job_endpoint=REGION \
  --environment_type=DOCKER \
  --environment_config=IMAGE_URI \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE

Sostituisci quanto segue:

  • REGION: la regione del servizio di job da utilizzare, sotto forma di indirizzo e porta. Ad esempio: localhost:3000. Utilizza embed per eseguire un servizio di job in-process.
  • IMAGE_URI: l'URI dell'immagine container personalizzata.
  • INPUT_FILE: un file di input che può essere letto come file di testo. Questo file deve essere accessibile dall'immagine container dell'SDK Harness
    , precaricata sull'immagine container o un file remoto.
  • OUTPUT_FILE: un percorso in cui scrivere l'output. Questo percorso è un percorso remoto o un percorso locale sul container.

Al termine della pipeline, esamina i log della console per verificare che la pipeline sia stata completata correttamente e che venga utilizzata l'immagine remota, specificata da IMAGE_URI.

Dopo aver eseguito la pipeline, i file salvati nel container non si trovano nel file system locale e il container viene arrestato. Puoi copiare i file dal file system del container arrestato utilizzando docker cp.

In alternativa:

  • Fornire output a un file system remoto come Cloud Storage. Potresti dover configurare manualmente l'accesso a scopo di test, anche per i file delle credenziali o per le Credenziali predefinite dell'applicazione.
  • Per un debug rapido, aggiungi il logging temporaneo.

Utilizza Direct Runner

Per test locali più approfonditi dell'immagine container e della pipeline, utilizza il Direct Runner di Apache Beam.

Puoi verificare la pipeline separatamente dal container eseguendo i test in un ambiente locale corrispondente all'immagine container o avviando la pipeline su un container in esecuzione.

Java

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  mvn compile exec:java -Dexec.mainClass=com.example.package.MyClassWithMain ...

Python

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  python path/to/my/pipeline.py ...

Vai

docker run -it --entrypoint "/bin/bash" IMAGE_URI
...
# On docker container:
root@4f041a451ef3:/#  go path/to/my/pipeline.go ...

Sostituisci IMAGE_URI con l'URI dell'immagine container personalizzata.

Gli esempi presuppongono che tutti i file della pipeline, inclusa la pipeline stessa, si trovino nel container personalizzato, siano stati montati da un file system locale o siano remoti e accessibili da Apache Beam e dal container. Ad esempio, per utilizzare Maven (mvn) per eseguire l'esempio Java precedente, è necessario eseguire lo stage di Maven e delle sue dipendenze nel container. Per saperne di più, consulta Storage e docker run nella documentazione di Docker.

L'obiettivo del test sul runner diretto è testare la pipeline nell'ambiente del container personalizzato, non testare l'esecuzione del container con il relativo ENTRYPOINT predefinito. Modifica ENTRYPOINT (ad esempio, docker run --entrypoint ...) per eseguire direttamente la pipeline o per consentire l'esecuzione manuale dei comandi sul container.

Se utilizzi una configurazione specifica basata sull'esecuzione del container su Compute Engine, puoi eseguirlo direttamente su una VM di Compute Engine. Per saperne di più, consulta Container su Compute Engine.

Avvia il job Dataflow

Quando avvii la pipeline Apache Beam su Dataflow, specifica il percorso dell'immagine container. Non utilizzare il tag :latest con le tue immagini personalizzate. Tagga le build con una data o un identificatore univoco. Se qualcosa va storto, l'utilizzo di questo tipo di tag potrebbe consentire di ripristinare l'esecuzione della pipeline a una configurazione di lavoro nota in precedenza e consentire un'ispezione delle modifiche.

Java

Utilizza --sdkContainerImage per specificare un'immagine container dell'SDK per il runtime Java.

Per attivare il runner portatile, utilizza il seguente esperimento in base alla versione dell'SDK:

  • SDK Apache Beam 2.74 o versioni successive: --experiments=enable_portable_runner.
  • SDK Apache Beam 2.73 o versioni precedenti: --experiments=use_runner_v2.

Python

Se utilizzi l'SDK versione 2.30.0 o successive, utilizza l'opzione della pipeline --sdk_container_image per specificare un'immagine container dell'SDK.

Per le versioni precedenti dell'SDK, utilizza l'opzione della pipeline --worker_harness_container_image per specificare la posizione dell'immagine container da utilizzare per l'harness dei worker.

I container personalizzati sono supportati solo per Dataflow Portable Runner. Se stai avviando una pipeline Python batch, imposta il flag dell'esperimento in base alla versione dell'SDK:

  • SDK Apache Beam 2.74 o versioni successive: --experiments=enable_portable_runner.
  • SDK Apache Beam 2.73 o versioni precedenti: --experiments=use_runner_v2.

Se stai avviando una pipeline Python in modalità flusso, non è necessario specificare l'esperimento, perché le pipeline Python in modalità flusso utilizzano il runner portatile per impostazione predefinita.

Vai

Se utilizzi l'SDK versione 2.40.0 o successive, utilizza l'opzione della pipeline --sdk_container_image per specificare un'immagine container dell'SDK.

Per le versioni precedenti dell'SDK, utilizza l'opzione della pipeline --worker_harness_container_image per specificare la posizione dell'immagine container da utilizzare per l'harness dei worker.

I container personalizzati sono supportati su tutte le versioni dell'SDK Go perché utilizzano il runner portatile di Dataflow per impostazione predefinita.

L'esempio seguente mostra come avviare l'esempio batch WordCount con un container personalizzato.

Java

mvn -Pdataflow-runner compile exec:java \
    -Dexec.mainClass=org.apache.beam.examples.WordCount \
    -Dexec.args="--runner=DataflowRunner \
                 --inputFile=INPUT_FILE \
                 --output=OUTPUT_FILE \
                 --project=PROJECT_ID \
                 --region=REGION \
                 --gcpTempLocation=TEMP_LOCATION \
                 --diskSizeGb=DISK_SIZE_GB \
                 --experiments=use_runner_v2 \
                 --sdkContainerImage=IMAGE_URI"

Python

Utilizzando l'SDK Apache Beam per Python versione 2.30.0 o successive:

python -m apache_beam.examples.wordcount \
  --input=INPUT_FILE \
  --output=OUTPUT_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --temp_location=TEMP_LOCATION \
  --runner=DataflowRunner \
  --disk_size_gb=DISK_SIZE_GB \
  --experiments=use_runner_v2 \
  --sdk_container_image=IMAGE_URI

Vai

wordcount --input gs://dataflow-samples/shakespeare/kinglear.txt \
          --output gs://<your-gcs-bucket>/counts \
          --runner dataflow \
          --project your-gcp-project \
          --region your-gcp-region \
          --temp_location gs://<your-gcs-bucket>/tmp/ \
          --staging_location gs://<your-gcs-bucket>/binaries/ \
          --sdk_container_image=IMAGE_URI

Sostituisci quanto segue:

  • INPUT_FILE: il percorso di input di Cloud Storage letto da Dataflow durante l'esecuzione dell'esempio.
  • OUTPUT_FILE: il percorso di output di Cloud Storage in cui scrive la pipeline di esempio. Questo file contiene i conteggi delle parole.
  • PROJECT_ID: l'ID del tuo progetto Google Cloud.
  • REGION: la regione in cui eseguire il deployment del job Dataflow.
  • TEMP_LOCATION: il percorso di Cloud Storage in cui Dataflow eseguirà il provisioning dei file di job temporanei creati durante l'esecuzione della pipeline.
  • DISK_SIZE_GB: (Facoltativo) Se il container è di grandi dimensioni, valuta la possibilità di aumentare le dimensioni del disco di avvio predefinite per evitare di esaurire lo spazio su disco.
  • IMAGE_URI: l'URI dell'immagine container personalizzata dell'SDK. Utilizza sempre un tag o un SHA del container con controllo delle versioni. Non utilizzare il tag :latest o un tag modificabile.

Streaming di immagini container

Puoi migliorare la latenza di avvio e scalabilità automatica della pipeline Dataflow attivando lo streaming di immagini. Questa funzionalità è utile se il container personalizzato contiene contenuti estranei o non utilizza tutti i contenuti in ogni passaggio. Ad esempio, il container potrebbe includere contenuti accessori, come il codice della libreria basato su CPU per l'inferenza basata su GPU. Analogamente, potresti avere un container che esegue pipeline di ML con più modelli che utilizzano un solo modello in ogni passaggio, quindi i suoi contenuti non sono necessari tutti contemporaneamente. L'attivazione dello streaming di immagini contribuirebbe a migliorare la latenza in questi casi.

Java

--dataflowServiceOptions=enable_image_streaming

Python

--dataflow_service_options=enable_image_streaming

Vai

--dataflow_service_options=enable_image_streaming

Lo streaming di immagini recupererà parti del container personalizzato in base alle esigenze del codice della pipeline, anziché scaricare l'intero container in anticipo. Le parti del container che non vengono utilizzate non devono mai essere scaricate.

Per usufruire dello streaming delle immagini, devi aver abilitato l' API Container File System.