Esecuzione di modelli classici

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

Nota: oltre al file modello, le pipeline basate su modelli si basano su file a cui viene fatto riferimento al momento della creazione del modello. Se sposti o rimuovi i file sottoposti a staging, il job della pipeline non andrà a buon fine.

Utilizzare la console Google Cloud

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

Modelli forniti da Google

Per eseguire un modello fornito da Google:

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



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



4. Inserisci un nome per il job nel campo Nome job.
5. Inserisci i valori dei parametri nei campi dei parametri forniti. La sezione Parametri aggiuntivi non è necessaria se utilizzi un messaggio fornito da Google.
6. Fai clic su Esegui job.

Modelli personalizzati

Per eseguire un modello personalizzato:

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



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



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

Utilizzo dell'API REST

Per eseguire un modello con una richiesta all'API REST, invia una richiesta POST HTTP con l'ID progetto. Questa richiesta richiede 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 valori dell'ambiente di runtime nella proprietà environment, non nella proprietà parameters. La La proprietà environment definisce l'ambiente di runtime per il job.

Creare un job batch di modelli personalizzati

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 va a buon fine, 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 di Dataflow che preferisci.
• Sostituisci JOB_NAME con il nome di un job a tua scelta.
• Sostituisci YOUR_BUCKET_NAME con il nome del tuo account Cloud Storage di sincronizzare la directory di una VM con un bucket.
• Imposta gcsPath sulla posizione del file di modello in Cloud Storage.
• Imposta parameters sull'elenco di coppie chiave-valore.
• Imposta tempLocation su una posizione in cui hai scrittura autorizzazione. 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"
        }
    }

Crea un job di flussi di dati con modello personalizzato

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

Modifica i seguenti valori:

• Sostituisci YOUR_PROJECT_ID con l'ID progetto.
• Sostituisci LOCATION con la regione di Dataflow che preferisci.
• Sostituisci JOB_NAME con il nome di un job a tua scelta.
• Sostituisci YOUR_BUCKET_NAME con il nome del tuo cloud Cloud Storage.
• Sostituisci GCS_PATH con la posizione di Cloud Storage del file del modello. La località 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 tue esigenze. Se utilizzi il modello di esempio, sostituisci le seguenti variabili.
  • Sostituisci YOUR_SUBSCRIPTION_NAME con il nome della sottoscrizione Pub/Sub.
  • Sostituisci YOUR_DATASET con il tuo set di dati BigQuery e sostituisci YOUR_TABLE_NAME con il nome della tua tabella BigQuery.
• Imposta tempLocation su una posizione in cui hai scrittura autorizzazione. 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"
        }
    }

Aggiorna un job di flussi di dati personalizzato

Questa richiesta projects.locations.templates.launch di esempio mostra come aggiornare un job di inserimento di flussi di modelli. Se vuoi aggiornare un modello flessibile, utilizza projects.locations.flexTemplates.launch.

1. Esegui Esempio 2: creazione di un job di inserimento di flussi da modello personalizzato per avviare un job di inserimento di flussi da modello.
2. Invia la seguente richiesta POST HTTP 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 da aggiornare.
   • Sostituisci GCS_PATH con la posizione di Cloud Storage del file del modello. La posizione 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 tue esigenze. Se utilizzi il modello di esempio, sostituisci le seguenti variabili.
     • Sostituisci YOUR_SUBSCRIPTION_NAME con il nome della sottoscrizione Pub/Sub.
     • Sostituisci YOUR_DATASET con il tuo set di dati BigQuery e 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 delle API di Google

Valuta l'uso della classe Librerie client delle API di Google per effettuare facilmente chiamate alle API REST di Dataflow. Questo script di esempio utilizza la libreria client API di Google per Python.

In questo esempio, devi impostare le seguenti variabili:

• project: imposta l'ID progetto.
• job: imposta un nome job univoco di tua scelta.
• template: impostato sulla posizione di Cloud Storage del file modello.
• parameters: impostato su un dizionario con i parametri del modello.

Per impostare il region, includi il location .

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 ulteriori informazioni sulle opzioni disponibili, consulta il metodo projects.locations.templates.launch nel riferimento all'API REST di Dataflow.

Utilizza gcloud CLI

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

gcloud CLI può eseguire un deployment Fornita da Google utilizzando il comando gcloud dataflow jobs run. Esempi di che eseguono modelli forniti da Google sono documentati Pagina dei modelli forniti da Google.

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

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

Creare un job batch di modelli personalizzati

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 nel formato seguente.

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

Creare un job di streaming di modelli personalizzati

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à essere esistente 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 nel formato seguente.

    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 la documentazione di riferimento di gcloud CLI.

Monitoraggio e risoluzione dei problemi

L'interfaccia di monitoraggio di Dataflow consente di monitorare i job Dataflow. Se un job non va a buon fine, puoi trovare suggerimenti per la risoluzione dei problemi, strategie di debug e un catalogo di errori comuni nella guida Risoluzione dei problemi della pipeline.