Avvio di pipeline Dataflow con Cloud Composer

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Questa pagina descrive come utilizzare DataflowTemplateOperator per avviare pipeline Dataflow da Cloud Composer. La pipeline da testo di Cloud Storage a BigQuery è una pipeline batch che ti consente di caricare i file di testo archiviati in Cloud Storage, trasformarli utilizzando una funzione JavaScript definita dall'utente (UDF) fornita e restituire i risultati in BigQuery.

Panoramica

• Prima di avviare il flusso di lavoro, crea le seguenti entità:

  • Una tabella BigQuery vuota da un set di dati vuoto che conterrà le seguenti colonne di informazioni: location, average_temperature,month e, facoltativamente, inches_of_rain, is_current e latest_measurement.

  • Un file JSON che normalizzerà i dati del file .txt nel formato corretto per lo schema della tabella BigQuery. L'oggetto JSON avrà un array di BigQuery Schema, dove ogni oggetto conterrà un nome di colonna, il tipo di input e se è un campo obbligatorio.

  • Un file di input .txt che conterrà i dati che verranno caricati in batch nella tabella BigQuery.

  • Una funzione definita dall'utente scritta in JavaScript che trasformerà ogni riga del file .txt nelle variabili pertinenti per la nostra tabella.

  • Un file DAG di Airflow che indirizzerà alla posizione di questi file.

• Successivamente, caricherai il file .txt, il file UDF .js e il file dello schema .json in un bucket Cloud Storage. Caricherai anche il DAG nell'ambiente Cloud Composer.

• Dopo il caricamento del DAG, Airflow eseguirà un'attività. Questa attività avvierà una pipeline Dataflow che applicherà la funzione definita dall'utente al file .txt e lo formatterà in base allo schema JSON.

• Infine, i dati verranno caricati nella tabella BigQuery che hai creato in precedenza.

Prima di iniziare

• Questa guida richiede familiarità con JavaScript per scrivere la funzione definita dall'utente.
• Questa guida presuppone che tu disponga già di un ambiente Cloud Composer. Consulta Creare un ambiente per crearne uno. Puoi utilizzare qualsiasi versione di Cloud Composer con questa guida.

• Enable the Cloud Composer, Dataflow, Cloud Storage, BigQuery APIs.

  Enable the APIs

• Assicurati di disporre delle seguenti autorizzazioni:

  • Ruoli Cloud Composer: crea un ambiente (se non ne hai uno), gestisci gli oggetti nel bucket dell'ambiente, esegui i DAG e accedi alla UI di Airflow.
  • Ruoli Cloud Storage: crea un bucket e gestisci gli oggetti al suo interno.
  • Ruoli BigQuery: crea un set di dati e una tabella, modifica i dati nella tabella, modifica lo schema e i metadati della tabella.
  • Ruoli Dataflow: visualizza i job Dataflow.

• Assicurati che il account di servizio del tuo ambiente disponga delle autorizzazioni per creare job Dataflow, accedere al bucket Cloud Storage e leggere e aggiornare i dati per la tabella in BigQuery.

Crea una tabella BigQuery vuota con una definizione dello schema

Crea una tabella BigQuery con una definizione dello schema. Utilizzerai questa definizione di schema più avanti in questa guida. Questa tabella BigQuery conterrà i risultati del caricamento batch.

Importante: le risorse BigQuery e il bucket Cloud Storage devono trovarsi nella stessa regione Compute Engine.

Per creare una tabella vuota con una definizione dello schema:

Console

1. Nella console Google Cloud , vai alla pagina BigQuery:

   Vai a BigQuery

2. Nel pannello di navigazione, nella sezione Risorse, espandi il progetto.

3. Nel riquadro dei dettagli, fai clic su Crea set di dati.

   

4. Nella pagina Crea set di dati, nella sezione ID set di dati, assegna un nome al set di dati average_weather. Lascia invariati tutti gli altri campi.

   

5. Fai clic su Crea set di dati.

6. Torna al pannello di navigazione, espandi il progetto nella sezione Risorse. Quindi, fai clic sul set di dati average_weather.

7. Nel riquadro dei dettagli, fai clic su Crea tabella.

   

8. Nella sezione Origine della pagina Crea tabella, seleziona Tabella vuota.

9. Nella sezione Destinazione della pagina Crea tabella:

   • Per Nome set di dati, scegli il set di dati average_weather.

     

   • Nel campo Nome tabella, inserisci il nome average_weather.

   • Verifica che Tipo di tabella sia impostato su Tabella nativa.

10. Nella sezione Schema, inserisci la definizione dello schema. Puoi utilizzare uno dei seguenti approcci:

    • Inserisci manualmente le informazioni sullo schema attivando l'opzione Modifica come testo e inserendo lo schema della tabella come array JSON. Digita i seguenti campi:

      [
          {
              "name": "location",
              "type": "GEOGRAPHY",
              "mode": "REQUIRED"
          },
          {
              "name": "average_temperature",
              "type": "INTEGER",
              "mode": "REQUIRED"
          },
          {
              "name": "month",
              "type": "STRING",
              "mode": "REQUIRED"
          },
          {
              "name": "inches_of_rain",
              "type": "NUMERIC"
          },
          {
              "name": "is_current",
              "type": "BOOLEAN"
          },
          {
              "name": "latest_measurement",
              "type": "DATE"
          }
      ]
      

    • Utilizza Aggiungi campo per inserire manualmente lo schema:

      

11. Per Impostazioni di partizionamento e clustering, lascia il valore predefinito, No partitioning.

12. Nella sezione Opzioni avanzate, per Crittografia lascia il valore predefinito, Google-owned and managed key.

13. Fai clic su Crea tabella.

Nota: quando crei una tabella vuota utilizzando la console Google Cloud , non puoi aggiungere un'etichetta, una descrizione o un tempo di scadenza. Puoi aggiungere queste proprietà facoltative quando crei una tabella utilizzando la CLI o l'API. Dopo aver creato una tabella nella console Google Cloud , puoi aggiungere ora di scadenza, descrizione ed etichette.

bq

Utilizza il comando bq mk per creare un set di dati vuoto e una tabella in questo set di dati.

Esegui questo comando per creare un set di dati delle condizioni meteo medie globali:

bq --location=LOCATION mk \
    --dataset PROJECT_ID:average_weather

Sostituisci quanto segue:

• LOCATION: la regione in cui si trova l'ambiente.
• PROJECT_ID: l'ID progetto.

Esegui questo comando per creare una tabella vuota in questo set di dati con la definizione dello schema:

bq mk --table \
PROJECT_ID:average_weather.average_weather \
location:GEOGRAPHY,average_temperature:INTEGER,month:STRING,inches_of_rain:NUMERIC,is_current:BOOLEAN,latest_measurement:DATE

Dopo aver creato la tabella, puoi aggiornare la scadenza, la descrizione e le etichette della tabella. Puoi anche modificare la definizione dello schema.

Python

Salva questo codice come dataflowtemplateoperator_create_dataset_and_table_helper.py e aggiorna le variabili al suo interno in modo che riflettano il tuo progetto e la tua posizione, poi esegui il codice con il seguente comando:

python dataflowtemplateoperator_create_dataset_and_table_helper.py

Python

Per eseguire l'autenticazione in Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

# Make sure to follow the quickstart setup instructions beforehand.
# See instructions here:
# https://cloud.google.com/bigquery/docs/quickstarts/quickstart-client-libraries

# Before running the sample, be sure to install the bigquery library
# in your local environment by running pip install google.cloud.bigquery

from google.cloud import bigquery

# TODO(developer): Replace with your values
project = "your-project"  # Your GCP Project
location = "US"  # the location where you want your BigQuery data to reside. For more info on possible locations see https://cloud.google.com/bigquery/docs/locations
dataset_name = "average_weather"


def create_dataset_and_table(project, location, dataset_name):
    # Construct a BigQuery client object.
    client = bigquery.Client(project)

    dataset_id = f"{project}.{dataset_name}"

    # Construct a full Dataset object to send to the API.
    dataset = bigquery.Dataset(dataset_id)

    # Set the location to your desired location for the dataset.
    # For more information, see this link:
    # https://cloud.google.com/bigquery/docs/locations
    dataset.location = location

    # Send the dataset to the API for creation.
    # Raises google.api_core.exceptions.Conflict if the Dataset already
    # exists within the project.
    dataset = client.create_dataset(dataset)  # Make an API request.

    print(f"Created dataset {client.project}.{dataset.dataset_id}")

    # Create a table from this dataset.

    table_id = f"{client.project}.{dataset_name}.average_weather"

    schema = [
        bigquery.SchemaField("location", "GEOGRAPHY", mode="REQUIRED"),
        bigquery.SchemaField("average_temperature", "INTEGER", mode="REQUIRED"),
        bigquery.SchemaField("month", "STRING", mode="REQUIRED"),
        bigquery.SchemaField("inches_of_rain", "NUMERIC", mode="NULLABLE"),
        bigquery.SchemaField("is_current", "BOOLEAN", mode="NULLABLE"),
        bigquery.SchemaField("latest_measurement", "DATE", mode="NULLABLE"),
    ]

    table = bigquery.Table(table_id, schema=schema)
    table = client.create_table(table)  # Make an API request.
    print(f"Created table {table.project}.{table.dataset_id}.{table.table_id}")

Crea un bucket Cloud Storage

Importante: le risorse BigQuery e il bucket Cloud Storage devono trovarsi nella stessa regione Compute Engine.

Crea un bucket per contenere tutti i file necessari per il flusso di lavoro. Il DAG che creerai più avanti in questa guida farà riferimento ai file che carichi in questo bucket di archiviazione. Per creare un nuovo bucket di archiviazione:

Console

1. Apri Cloud Storage nella console Google Cloud .

   Vai a Cloud Storage

2. Fai clic su Crea bucket per aprire il modulo di creazione del bucket.

   1. Inserisci le informazioni sul bucket e fai clic su Continua per completare ogni passaggio:

      • Specifica un nome univoco a livello globale per il bucket. Questa guida utilizza bucketName come esempio.

      • Seleziona Regione per il tipo di località. Successivamente, seleziona una posizione in cui verranno archiviati i dati del bucket.

      • Seleziona Standard come classe di archiviazione predefinita per i tuoi dati.

      • Seleziona controllo dell'accesso Uniforme per accedere ai tuoi oggetti.

   2. Fai clic su Fine.

gcloud

Utilizza il comando gcloud storage buckets create:

gcloud storage buckets create gs://bucketName/

Sostituisci quanto segue:

• bucketName: il nome del bucket che hai creato in precedenza in questa guida.

Esempi di codice

C#

Per eseguire l'autenticazione in Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

using Google.Apis.Storage.v1.Data;
using Google.Cloud.Storage.V1;
using System;

public class CreateBucketSample
{
    public Bucket CreateBucket(
        string projectId = "your-project-id",
        string bucketName = "your-unique-bucket-name")
    {
        var storage = StorageClient.Create();
        var bucket = storage.CreateBucket(projectId, bucketName);
        Console.WriteLine($"Created {bucketName}.");
        return bucket;
    }
}

Go

Per eseguire l'autenticazione in Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

import (
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/storage"
)

// createBucket creates a new bucket in the project.
func createBucket(w io.Writer, projectID, bucketName string) error {
	// projectID := "my-project-id"
	// bucketName := "bucket-name"
	ctx := context.Background()
	client, err := storage.NewClient(ctx)
	if err != nil {
		return fmt.Errorf("storage.NewClient: %w", err)
	}
	defer client.Close()

	ctx, cancel := context.WithTimeout(ctx, time.Second*30)
	defer cancel()

	bucket := client.Bucket(bucketName)
	if err := bucket.Create(ctx, projectID, nil); err != nil {
		return fmt.Errorf("Bucket(%q).Create: %w", bucketName, err)
	}
	fmt.Fprintf(w, "Bucket %v created\n", bucketName)
	return nil
}

Java

Per eseguire l'autenticazione in Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

import com.google.cloud.storage.Bucket;
import com.google.cloud.storage.BucketInfo;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;

public class CreateBucket {
  public static void createBucket(String projectId, String bucketName) {
    // The ID of your GCP project
    // String projectId = "your-project-id";

    // The ID to give your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();

    Bucket bucket = storage.create(BucketInfo.newBuilder(bucketName).build());

    System.out.println("Created bucket " + bucket.getName());
  }
}

Python

Per eseguire l'autenticazione in Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

from google.cloud import storage


def create_bucket(bucket_name):
    """Creates a new bucket."""
    # bucket_name = "your-new-bucket-name"

    storage_client = storage.Client()

    bucket = storage_client.create_bucket(bucket_name)

    print(f"Bucket {bucket.name} created")

Ruby

Per eseguire l'autenticazione in Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

def create_bucket bucket_name:
  # The ID to give your GCS bucket
  # bucket_name = "your-unique-bucket-name"

  require "google/cloud/storage"

  storage = Google::Cloud::Storage.new
  bucket  = storage.create_bucket bucket_name

  puts "Created bucket: #{bucket.name}"
end

Crea uno schema BigQuery in formato JSON per la tabella di output

Crea un file di schema BigQuery in formato JSON che corrisponda alla tabella di output creata in precedenza. Tieni presente che i nomi, i tipi e le modalità dei campi devono corrispondere a quelli definiti in precedenza nello schema della tabella BigQuery. Questo file normalizzerà i dati del file .txt in un formato compatibile con lo schema BigQuery. Assegna un nome a questo file jsonSchema.json.

{
    "BigQuery Schema": [
    {
        "name": "location",
        "type": "GEOGRAPHY",
        "mode": "REQUIRED"
    },
    {
        "name": "average_temperature",
        "type": "INTEGER",
        "mode": "REQUIRED"
    },
    {
        "name": "month",
        "type": "STRING",
        "mode": "REQUIRED"
    },
    {
        "name": "inches_of_rain",
        "type": "NUMERIC"
    },
    {
        "name": "is_current",
        "type": "BOOLEAN"
    },
    {
        "name": "latest_measurement",
        "type": "DATE"
    }]
}

Creare un file JavaScript per formattare i dati

In questo file definirai la tua UDF (User Defined Function) che fornisce la logica per trasformare le righe di testo nel file di input. Tieni presente che questa funzione considera ogni riga di testo del file di input come un argomento a sé stante, quindi viene eseguita una volta per ogni riga del file di input. Assegna un nome a questo file transformCSVtoJSON.js.

function transformCSVtoJSON(line) {
  var values = line.split(',');
  var properties = [
    'location',
    'average_temperature',
    'month',
    'inches_of_rain',
    'is_current',
    'latest_measurement',
  ];
  var weatherInCity = {};

  for (var count = 0; count < values.length; count++) {
    if (values[count] !== 'null') {
      weatherInCity[properties[count]] = values[count];
    }
  }

  return JSON.stringify(weatherInCity);
}

Crea il file di input

Questo file conterrà le informazioni che vuoi caricare nella tua tabella BigQuery. Copia questo file localmente e assegnagli il nome inputFile.txt.

POINT(40.7128 74.006),45,'July',null,true,2020-02-16
POINT(41.8781 87.6298),23,'October',13,false,2015-02-13
POINT(48.8566 2.3522),80,'December',null,true,null
POINT(6.5244 3.3792),15,'March',14,true,null

Attenzione: assicurati che non ci siano righe vuote alla fine del file inputFile.txt per evitare un errore di Dataflow.

Carica i file nel bucket

Carica i seguenti file nel bucket Cloud Storage che hai creato in precedenza:

• Schema BigQuery in formato JSON (.json)
• Funzione definita dall'utente JavaScript (transformCSVtoJSON.js)

• Il file di input del testo che vuoi elaborare (.txt)

Console

1. Nella console Google Cloud , vai alla pagina Bucket in Cloud Storage.
   

   Vai a Bucket

2. Nell'elenco dei bucket, fai clic sul tuo bucket.

3. Nella scheda Oggetti del bucket, esegui una delle seguenti operazioni:

   • Trascina i file che ti interessano dal desktop o dal file manager nel riquadro principale della console Google Cloud .

   • Fai clic sul pulsante Carica file, seleziona i file da caricare nella finestra di dialogo visualizzata e fai clic su Apri.

gcloud

Esegui il comando gcloud storage cp:

gcloud storage cp OBJECT_LOCATION gs://bucketName

Sostituisci quanto segue:

• bucketName: il nome del bucket che hai creato in precedenza in questa guida.
• OBJECT_LOCATION: il percorso locale dell'oggetto. Ad esempio, Desktop/transformCSVtoJSON.js.

Esempi di codice

Python

Per eseguire l'autenticazione in Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

from google.cloud import storage


def upload_blob(bucket_name, source_file_name, destination_blob_name):
    """Uploads a file to the bucket."""
    # The ID of your GCS bucket
    # bucket_name = "your-bucket-name"
    # The path to your file to upload
    # source_file_name = "local/path/to/file"
    # The ID of your GCS object
    # destination_blob_name = "storage-object-name"

    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    # Optional: set a generation-match precondition to avoid potential race conditions
    # and data corruptions. The request to upload is aborted if the object's
    # generation number does not match your precondition. For a destination
    # object that does not yet exist, set the if_generation_match precondition to 0.
    # If the destination object already exists in your bucket, set instead a
    # generation-match precondition using its generation number.
    generation_match_precondition = 0

    blob.upload_from_filename(source_file_name, if_generation_match=generation_match_precondition)

    print(
        f"File {source_file_name} uploaded to {destination_blob_name}."
    )

Ruby

Per eseguire l'autenticazione in Cloud Composer, configura le Credenziali predefinite dell'applicazione. Per ulteriori informazioni, consulta Configura l'autenticazione per un ambiente di sviluppo locale.

def upload_file bucket_name:, local_file_path:, file_name: nil
  # The ID of your GCS bucket
  # bucket_name = "your-unique-bucket-name"

  # The path to your file to upload
  # local_file_path = "/local/path/to/file.txt"

  # The ID of your GCS object
  # file_name = "your-file-name"

  require "google/cloud/storage"

  storage = Google::Cloud::Storage.new
  bucket  = storage.bucket bucket_name, skip_lookup: true

  file = bucket.create_file local_file_path, file_name

  puts "Uploaded #{local_file_path} as #{file.name} in bucket #{bucket_name}"
end

Configura DataflowTemplateOperator

Prima di eseguire il DAG, imposta le seguenti variabili Airflow.

Variabile Airflow	Valore
project_id	L'ID progetto. Esempio: example-project.
gce_zone	Zona Compute Engine in cui deve essere creato il cluster Dataflow. Esempio: us-central1-a. Per saperne di più sulle zone valide, consulta Regioni e zone.
bucket_path	La posizione del bucket Cloud Storage che hai creato in precedenza. Esempio: gs://example-bucket.

Ora farai riferimento ai file che hai creato in precedenza per creare un DAG che avvii il flusso di lavoro Dataflow. Copia questo DAG e salvalo localmente come composer-dataflow-dag.py.

Airflow 2

"""Example Airflow DAG that creates a Cloud Dataflow workflow which takes a
text file and adds the rows to a BigQuery table.

This DAG relies on four Airflow variables
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id - Google Cloud Project ID to use for the Cloud Dataflow cluster.
* gce_zone - Google Compute Engine zone where Cloud Dataflow cluster should be
  created.
For more info on zones where Dataflow is available see:
https://cloud.google.com/dataflow/docs/resources/locations
* bucket_path - Google Cloud Storage bucket where you've stored the User Defined
Function (.js), the input file (.txt), and the JSON schema (.json).
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataflow import (
    DataflowTemplatedJobStartOperator,
)
from airflow.utils.dates import days_ago

bucket_path = "{{var.value.bucket_path}}"
project_id = "{{var.value.project_id}}"
gce_zone = "{{var.value.gce_zone}}"


default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "dataflow_default_options": {
        "project": project_id,
        # Set to your zone
        "zone": gce_zone,
        # This is a subfolder for storing temporary files, like the staged pipeline job.
        "tempLocation": bucket_path + "/tmp/",
    },
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    # The id you will see in the DAG airflow page
    "composer_dataflow_dag",
    default_args=default_args,
    # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    start_template_job = DataflowTemplatedJobStartOperator(
        # The task id of your job
        task_id="dataflow_operator_transform_csv_to_bq",
        # The name of the template that you're using.
        # Below is a list of all the templates you can use.
        # For versions in non-production environments, use the subfolder 'latest'
        # https://cloud.google.com/dataflow/docs/guides/templates/provided-batch#gcstexttobigquery
        template="gs://dataflow-templates/latest/GCS_Text_to_BigQuery",
        # Use the link above to specify the correct parameters for your template.
        parameters={
            "javascriptTextTransformFunctionName": "transformCSVtoJSON",
            "JSONPath": bucket_path + "/jsonSchema.json",
            "javascriptTextTransformGcsPath": bucket_path + "/transformCSVtoJSON.js",
            "inputFilePattern": bucket_path + "/inputFile.txt",
            "outputTable": project_id + ":average_weather.average_weather",
            "bigQueryLoadingTemporaryDirectory": bucket_path + "/tmp/",
        },
    )

Airflow 1

"""Example Airflow DAG that creates a Cloud Dataflow workflow which takes a
text file and adds the rows to a BigQuery table.

This DAG relies on four Airflow variables
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id - Google Cloud Project ID to use for the Cloud Dataflow cluster.
* gce_zone - Google Compute Engine zone where Cloud Dataflow cluster should be
  created.
  created.
Learn more about the difference between the two here:
https://cloud.google.com/compute/docs/regions-zones
* bucket_path - Google Cloud Storage bucket where you've stored the User Defined
Function (.js), the input file (.txt), and the JSON schema (.json).
"""

import datetime

from airflow import models
from airflow.contrib.operators.dataflow_operator import DataflowTemplateOperator
from airflow.utils.dates import days_ago

bucket_path = "{{var.value.bucket_path}}"
project_id = "{{var.value.project_id}}"
gce_zone = "{{var.value.gce_zone}}"


default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "dataflow_default_options": {
        "project": project_id,
        # Set to your zone
        "zone": gce_zone,
        # This is a subfolder for storing temporary files, like the staged pipeline job.
        "tempLocation": bucket_path + "/tmp/",
    },
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    # The id you will see in the DAG airflow page
    "composer_dataflow_dag",
    default_args=default_args,
    # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    start_template_job = DataflowTemplateOperator(
        # The task id of your job
        task_id="dataflow_operator_transform_csv_to_bq",
        # The name of the template that you're using.
        # Below is a list of all the templates you can use.
        # For versions in non-production environments, use the subfolder 'latest'
        # https://cloud.google.com/dataflow/docs/guides/templates/provided-batch#gcstexttobigquery
        template="gs://dataflow-templates/latest/GCS_Text_to_BigQuery",
        # Use the link above to specify the correct parameters for your template.
        parameters={
            "javascriptTextTransformFunctionName": "transformCSVtoJSON",
            "JSONPath": bucket_path + "/jsonSchema.json",
            "javascriptTextTransformGcsPath": bucket_path + "/transformCSVtoJSON.js",
            "inputFilePattern": bucket_path + "/inputFile.txt",
            "outputTable": project_id + ":average_weather.average_weather",
            "bigQueryLoadingTemporaryDirectory": bucket_path + "/tmp/",
        },
    )

Carica il DAG in Cloud Storage

Carica il DAG nella cartella /dags del bucket dell'ambiente. Una volta completato il caricamento, puoi visualizzarlo facendo clic sul link Cartella DAG nella pagina Ambienti Cloud Composer.

Visualizzare lo stato dell'attività

1. Vai all'interfaccia web di Airflow.
2. Nella pagina DAG, fai clic sul nome del DAG (ad esempio composerDataflowDAG).
3. Nella pagina Dettagli DAG, fai clic su Visualizzazione grafico.

4. Controlla lo stato:

   • Failed: La casella intorno all'attività è rossa. Puoi anche tenere il puntatore sopra l'attività e cercare Stato: non riuscito.

   • Success: La casella intorno all'attività è verde. Puoi anche passare il puntatore sopra l'attività e verificare lo Stato: riuscito.

Dopo alcuni minuti, puoi controllare i risultati in Dataflow e BigQuery.

Visualizzare il job in Dataflow

1. Nella console Google Cloud , vai alla pagina Dataflow.

   Vai a Dataflow

2. Il tuo job si chiama dataflow_operator_transform_csv_to_bq con un ID univoco allegato alla fine del nome con un trattino, ad esempio:

   

3. Fai clic sul nome per visualizzare i dettagli del job.

   

Visualizza i risultati in BigQuery

1. Nella console Google Cloud , vai alla pagina BigQuery.

   Vai a BigQuery

2. Puoi inviare query utilizzando SQL standard. Utilizza la seguente query per visualizzare le righe aggiunte alla tabella:

   SELECT * FROM projectId.average_weather.average_weather
   

Passaggi successivi

• Gestisci i DAG
• Accedere alla UI di Airflow