Esecuzione di modelli classici

Dopo aver creato e archiviato il modello Dataflow in un'area intermedia, esegui il modello con la Google Cloud console, l'API REST o Google Cloud CLI. Puoi eseguire il deployment dei job del modello Dataflow da molti ambienti, tra cui l'ambiente standard di App Engine, le funzioni di Cloud Run e altri ambienti vincolati.

Nota: oltre al file del modello, le pipeline basate su modelli si basano su file archiviati e a cui è stato fatto riferimento al momento della creazione del modello. Se sposti o rimuovi i file archiviati, il job della pipeline non verrà eseguito.

Utilizzare la Google Cloud console

Puoi utilizzare la Google Cloud console per eseguire modelli Dataflow forniti da Google e personalizzati.

Modelli forniti da Google

Per eseguire un modello fornito da Google:

1. Vai alla pagina Dataflow nella Google Cloud console.
Vai alla pagina Dataflow
2. Fai clic CREA JOB DA MODELLO.



3. Seleziona il modello fornito da Google che vuoi eseguire dal Modello Dataflow menu a discesa.



4. Inserisci un nome del job nel campo Nome job.
5. Inserisci i valori dei parametri nei campi dei parametri forniti. Non hai bisogno della sezione Parametri aggiuntivi quando utilizzi un modello fornito da Google.
6. Fai clic su Esegui job.

Modelli personalizzati

Per eseguire un modello personalizzato:

1. Vai alla pagina Dataflow nella Google Cloud console.
Vai alla pagina Dataflow
2. Fai clic su CREA JOB DA MODELLO.



3. Seleziona Modello personalizzato dal Modello Dataflow menu a discesa.



4. Inserisci un nome del job nel campo Nome job.
5. Inserisci il percorso Cloud Storage del file del modello nel campo del percorso Cloud Storage del modello.
6. Se il modello richiede parametri, fai clic su AGGIUNGI PARAMETRO nella sezione Parametri aggiuntivi. Inserisci il Nome e il Valore del parametro. Ripeti questo passaggio per ogni parametro necessario.
7. Fai clic su Esegui job.

Utilizzare l'API REST

Per eseguire un modello con una richiesta dell'API REST, invia una richiesta HTTP POST con l'ID progetto. Questa richiesta richiede l'autorizzazione.

Consulta il riferimento dell'API REST per projects.locations.templates.launch per saperne di più sui parametri disponibili.

Nel corpo della richiesta JSON, imposta i valori dell'ambiente di runtime nella proprietà environment, non nella proprietà parameters. La environment proprietà definisce l'ambiente di runtime per il job.

Creare un job batch del modello personalizzato

Questa richiesta projects.locations.templates.launch di esempio crea un job batch da un modello che legge un file di testo e scrive un file di testo di output. Se la richiesta ha esito positivo, il corpo della risposta contiene un'istanza di LaunchTemplateResponse.

Modifica i seguenti valori:

• Sostituisci YOUR_PROJECT_ID con l'ID progetto.
• Sostituisci LOCATION con la regione Dataflow che preferisci.
• Sostituisci JOB_NAME con un nome del job a tua scelta.
• Sostituisci YOUR_BUCKET_NAME con il nome del tuo bucket Cloud Storage.
• Imposta gcsPath sul percorso Cloud Storage del file del modello.
• Imposta parameters sull'elenco di coppie chiave-valore.
• Imposta tempLocation su una località in cui hai l'autorizzazione di scrittura permesso. Questo valore è obbligatorio per eseguire i modelli forniti da Google.

    POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=gs://YOUR_BUCKET_NAME/templates/TemplateName
    {
        "jobName": "JOB_NAME",
        "parameters": {
            "inputFile" : "gs://YOUR_BUCKET_NAME/input/my_input.txt",
            "output": "gs://YOUR_BUCKET_NAME/output/my_output"
        },
        "environment": {
            "tempLocation": "gs://YOUR_BUCKET_NAME/temp",
            "zone": "us-central1-f"
        }
    }

Creare un job di streaming del modello personalizzato

Questa richiesta projects.locations.templates.launch di esempio crea un job di streaming da un modello classico che legge da un abbonamento Pub/Sub e scrive in una tabella BigQuery. Se vuoi avviare un modello flessibile, utilizza projects.locations.flexTemplates.launch invece. Il modello di esempio è un modello fornito da Google. Puoi modificare il percorso nel modello in modo che punti a un modello personalizzato modello. La stessa logica viene utilizzata per avviare i modelli forniti da Google e personalizzati. In questo esempio, la tabella BigQuery deve già esistere con lo schema appropriato. In caso di esito positivo, il corpo della risposta contiene un'istanza di LaunchTemplateResponse.

Modifica i seguenti valori:

• Sostituisci YOUR_PROJECT_ID con l'ID progetto.
• Sostituisci LOCATION con la regione Dataflow che preferisci.
• Sostituisci JOB_NAME con un nome del job a tua scelta.
• Sostituisci YOUR_BUCKET_NAME con il nome del tuo bucket Cloud Storage.
• Sostituisci GCS_PATH con il percorso Cloud Storage del file del modello. Il percorso deve iniziare con gs://
• Imposta parameters sull'elenco di coppie chiave-valore. I parametri elencati sono specifici per 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 YOUR_SUBSCRIPTION_NAME con il nome dell'abbonamento Pub/Sub.
  • Sostituisci YOUR_DATASET con il set di dati BigQuery e sostituisci YOUR_TABLE_NAME con il nome della tabella BigQuery.
• Imposta tempLocation su una località in cui hai l'autorizzazione di scrittura permesso. Questo valore è obbligatorio per eseguire i modelli forniti da Google.

    POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=GCS_PATH
    {
        "jobName": "JOB_NAME",
        "parameters": {
            "inputSubscription": "projects/YOUR_PROJECT_ID/subscriptions/YOUR_SUBSCRIPTION_NAME",
            "outputTableSpec": "YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME"
        },
        "environment": {
            "tempLocation": "gs://YOUR_BUCKET_NAME/temp",
            "zone": "us-central1-f"
        }
    }

Aggiornare un job di streaming del modello personalizzato

Questa richiesta projects.locations.templates.launch di esempio mostra come aggiornare un job di streaming del modello. Se vuoi aggiornare un modello flessibile, utilizza projects.locations.flexTemplates.launch invece.

1. Esegui Esempio 2: Creare un job di streaming del modello personalizzato per avviare un job di streaming del modello.
2. Invia la seguente richiesta HTTP POST con i seguenti valori modificati:
   • Sostituisci YOUR_PROJECT_ID con l'ID progetto.
   • Sostituisci LOCATION con la regione Dataflow del job che stai aggiornando.
   • Sostituisci JOB_NAME con il nome esatto del job che vuoi aggiornare.
   • Sostituisci GCS_PATH con il percorso Cloud Storage del file del modello. Il percorso deve iniziare con gs://
   • Imposta parameters sull'elenco di coppie chiave-valore. I parametri elencati sono specifici per 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 YOUR_SUBSCRIPTION_NAME con il nome dell'abbonamento Pub/Sub.
     • Sostituisci YOUR_DATASET con il set di dati BigQuery e sostituisci YOUR_TABLE_NAME con il nome della tabella BigQuery.
   • Utilizza il parametro environment per modificare le impostazioni dell'ambiente, ad esempio il tipo di macchina. Questo esempio utilizza il tipo di macchina n2-highmem-2, che ha più memoria e CPU per worker rispetto al tipo di macchina predefinito.

       POST https://dataflow.googleapis.com/v1b3/projects/YOUR_PROJECT_ID/locations/LOCATION/templates:launch?gcsPath=GCS_PATH
       {
           "jobName": "JOB_NAME",
           "parameters": {
               "inputSubscription": "projects/YOUR_PROJECT_ID/subscriptions/YOUR_TOPIC_NAME",
               "outputTableSpec": "YOUR_PROJECT_ID:YOUR_DATASET.YOUR_TABLE_NAME"
           },
           "environment": {
               "machineType": "n2-highmem-2"
           },
           "update": true
       }
   

3. Accedi all' interfaccia di monitoraggio di Dataflow e verifica che sia stato creato un nuovo job con lo stesso nome. Questo job ha lo stato Aggiornato.

Utilizzare le librerie client dell'API di Google

Valuta la possibilità di utilizzare le librerie client dell'API di Google per effettuare facilmente chiamate alle API REST di Dataflow. Questo script di esempio utilizza la libreria client dell'API di Google per Python.

In questo esempio, devi impostare le seguenti variabili:

• project: imposta l'ID progetto.
• job: imposta un nome del job univoco a tua scelta.
• template: imposta il percorso Cloud Storage del file del modello.
• parameters: imposta un dizionario con i parametri del modello.

Per impostare la regione, includi il location parametro.

from googleapiclient.discovery import build

# project = 'your-gcp-project'
# job = 'unique-job-name'
# template = 'gs://dataflow-templates/latest/Word_Count'
# parameters = {
#     'inputFile': 'gs://dataflow-samples/shakespeare/kinglear.txt',
#     'output': 'gs://<your-gcs-bucket>/wordcount/outputs',
# }

dataflow = build("dataflow", "v1b3")
request = (
    dataflow.projects()
    .templates()
    .launch(
        projectId=project,
        gcsPath=template,
        body={
            "jobName": job,
            "parameters": parameters,
        },
    )
)

response = request.execute()

Per saperne di più sulle opzioni disponibili, consulta il projects.locations.templates.launch metodo nel riferimento dell'API REST di Dataflow.

Utilizzare gcloud CLI

Nota: per utilizzare gcloud CLI per eseguire i modelli, devi avere Cloud SDK versione 138.0.0 o successive.

gcloud CLI può eseguire un modello personalizzato o fornito da Google utilizzando il gcloud dataflow jobs run comando. Gli esempi di esecuzione dei modelli forniti da Google sono documentati nella pagina Modelli forniti da Google.

Per i seguenti esempi di modelli personalizzati, imposta i seguenti valori:

• Sostituisci JOB_NAME con un nome del job a tua scelta.
• Sostituisci YOUR_BUCKET_NAME con il nome del tuo bucket Cloud Storage.
• Imposta --gcs-location sul percorso Cloud Storage del file del modello.
• Imposta --parameters sull'elenco separato da virgole dei parametri da passare al job. Gli spazi tra virgole e valori non sono consentiti.
• Per impedire alle VM di accettare le chiavi SSH archiviate nei metadati del progetto, utilizza il flag additional-experiments con l'opzione di servizio block_project_ssh_keys: --additional-experiments=block_project_ssh_keys.

Creare un job batch del modello personalizzato

Questo esempio crea un job batch da un modello che legge un file di testo e scrive un file di testo di output.

    gcloud dataflow jobs run JOB_NAME \
        --gcs-location gs://YOUR_BUCKET_NAME/templates/MyTemplate \
        --parameters inputFile=gs://YOUR_BUCKET_NAME/input/my_input.txt,output=gs://YOUR_BUCKET_NAME/output/my_output

La richiesta restituisce una risposta con il seguente formato.

    id: 2016-10-11_17_10_59-1234530157620696789
    projectId: YOUR_PROJECT_ID
    type: JOB_TYPE_BATCH

Creare un job di streaming del modello personalizzato

Questo esempio crea un job di streaming da un modello che legge da un argomento Pub/Sub e scrive in una tabella BigQuery. La tabella BigQuery deve già esistere con lo schema appropriato.

    gcloud dataflow jobs run JOB_NAME \
        --gcs-location gs://YOUR_BUCKET_NAME/templates/MyTemplate \
        --parameters topic=projects/project-identifier/topics/resource-name,table=my_project:my_dataset.my_table_name

La richiesta restituisce una risposta con il seguente formato.

    id: 2016-10-11_17_10_59-1234530157620696789
    projectId: YOUR_PROJECT_ID
    type: JOB_TYPE_STREAMING

Per un elenco completo dei flag per il comando gcloud dataflow jobs run, consulta il riferimento di gcloud CLI.

Monitoraggio e risoluzione dei problemi

L'interfaccia di monitoraggio di Dataflow ti consente di monitorare i job Dataflow. Se un job non viene eseguito, puoi trovare suggerimenti per la risoluzione dei problemi, strategie di debug e un catalogo di errori comuni nella guida Risolvere i problemi della pipeline.