Para adoptar con más firmeza el éxito y la creciente preferencia de los clientes por las soluciones de OSS, Cloud Composer evolucionará para convertirse en Servicio administrado para Apache Airflow. Este cambio de nombre mejora la comprensión de los clientes sobre nuestra cartera y refuerza nuestro compromiso de ser el ecosistema de nube más abierto.

Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Inicia canalizaciones de Dataflow con Airflow administrado

Managed Airflow (Gen 3) | Managed Airflow (Gen 2) | Managed Airflow (Legacy Gen 1)

En esta página, se describe cómo usar DataflowTemplateOperator para iniciar canalizaciones de Dataflow desde Managed Airflow. La canalización de Cloud Storage Text a BigQuery es una canalización por lotes que te permite subir archivos de texto almacenados en Cloud Storage, transformarlos con una función definida por el usuario (UDF) de JavaScript que proporciones y generar los resultados en BigQuery.

Se subirán una función definida por el usuario, un archivo de entrada y un esquema JSON a un bucket de Cloud Storage. Un DAG que hace referencia a estos archivos iniciará una canalización por lotes de Dataflow que aplicará la función definida por el usuario y el archivo de esquema JSON al archivo de entrada. Luego, este contenido se subirá a una tabla de BigQuery.

Descripción general

Antes de iniciar el flujo de trabajo, crearás las siguientes entidades:
- Una tabla de BigQuery vacía con un conjunto de datos vacío que contendrá las siguientes columnas de información: location, average_temperature, month y, de forma opcional, inches_of_rain, is_current y latest_measurement.
- Es un archivo JSON que normalizará los datos del archivo .txt en el formato correcto para el esquema de la tabla de BigQuery. El objeto JSON tendrá un array de BigQuery Schema, en el que cada objeto contendrá un nombre de columna, un tipo de entrada y si es un campo obligatorio.
- Un archivo de entrada .txt que contendrá los datos que se subirán de forma masiva a la tabla de BigQuery.
- Una función definida por el usuario escrita en JavaScript que transformará cada línea del archivo .txt en las variables relevantes para nuestra tabla.
- Un archivo DAG de Airflow que apuntará a la ubicación de estos archivos.
A continuación, subirás el archivo .txt, el archivo de UDF .js y el archivo de esquema .json a un bucket de Cloud Storage. También subirás el DAG a tu entorno de Managed Airflow.
Después de subir el DAG, Airflow ejecutará una tarea desde él. Esta tarea iniciará una canalización de Dataflow que aplicará la función definida por el usuario al archivo .txt y lo formateará según el esquema JSON.
Por último, los datos se subirán a la tabla de BigQuery que creaste antes.

Antes de comenzar

En esta guía, se requiere familiaridad con JavaScript para escribir la función definida por el usuario.
En esta guía, se supone que ya tienes un entorno de Managed Airflow. Consulta Crea un entorno para crear uno. Puedes usar cualquier versión de Managed Airflow con esta guía.
Habilita las APIs de Managed Service para Apache Airflow, Dataflow, Cloud Storage y BigQuery si alguna aún no está habilitada.
Roles necesarios para habilitar las APIs
Para habilitar APIs, necesitas el permiso serviceusage.services.enable. Si creaste el proyecto, es probable que ya tengas este permiso a través del rol de propietario (roles/owner). De lo contrario, puedes obtener este permiso a través del rol de administrador de Service Usage (roles/serviceusage.serviceUsageAdmin). Obtén más información para otorgar roles.
Habilitar las API
Asegúrate de tener los siguientes permisos:
- Roles de Managed Airflow: Crear un entorno (si no tienes uno), administrar objetos en el bucket del entorno, ejecutar DAGs y acceder a la IU de Airflow
- Roles de Cloud Storage: Crea un bucket y administra objetos en él.
- Roles de BigQuery: Crear un conjunto de datos y una tabla, modificar los datos de la tabla, modificar el esquema y los metadatos de la tabla
- Roles de Dataflow: Permiten ver los trabajos de Dataflow.
Asegúrate de que la cuenta de servicio de tu entorno tenga permisos para crear trabajos de Dataflow, acceder al bucket de Cloud Storage y leer y actualizar datos para la tabla en BigQuery.

Crea una tabla de BigQuery vacía con una definición de esquema

Crea una tabla de BigQuery con una definición de esquema. Usarás esta definición de esquema más adelante en esta guía. Esta tabla de BigQuery contendrá los resultados de la carga por lotes.

Para crear una tabla vacía con una definición de esquema, haz lo siguiente:

Console

En la consola de Google Cloud , ve a la página BigQuery:

Ir a BigQuery
En el panel de navegación, en la sección Recursos, expande tu proyecto.
En el panel de detalles, haz clic en Crear conjunto de datos.
En la página Crear conjunto de datos, en la sección ID del conjunto de datos, asigna el nombre average_weather a tu conjunto de datos. Deja todos los demás campos en su estado predeterminado.
Haz clic en Crear conjunto de datos.
Vuelve al panel de navegación y, en la sección Recursos, expande tu proyecto. Luego, haz clic en el conjunto de datos average_weather.
En el panel de detalles, haz clic en Crear tabla.
En la sección Origen de la página Crear tabla, selecciona Tabla vacía.
En la sección Destination (Destino) de la página Create table (Crear tabla), haz lo siguiente:
- En Nombre del conjunto de datos, selecciona el conjunto de datos average_weather.
- En el campo Nombre de la tabla, ingresa el nombre average_weather.
- Verifica que Tipo de tabla esté establecido en Tabla nativa.

En la sección Esquema, ingresa la definición del esquema. Puedes usar uno de los siguientes enfoques:

Ingresa la información del esquema de forma manual. Para ello, habilita Editar como texto y, luego, ingresa el esquema de la tabla como un array JSON. Escribe en los siguientes campos:

[
    {
        "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"
    }
]

Usa Agregar campo para ingresar el esquema de forma manual:

En Configuración de partición y agrupamiento en clústeres, deja el valor predeterminado, No partitioning.
En la sección Opciones avanzadas, en Encriptación, deja el valor predeterminado, Google-owned and managed key.
Haz clic en Crear tabla.

bq

Usa el comando bq mk para crear un conjunto de datos vacío y una tabla en este conjunto.

Ejecuta el siguiente comando para crear un conjunto de datos del clima promedio global:

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

Reemplaza lo siguiente:

LOCATION: Es la región en la que se encuentra el entorno.
PROJECT_ID: Es el ID del proyecto.

Ejecuta el siguiente comando para crear una tabla vacía en este conjunto de datos con la definición del esquema:

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

Una vez que se crea la tabla, puedes actualizar el vencimiento, la descripción y las etiquetas. También puedes modificar la definición de esquema.

Python

Guarda este código como dataflowtemplateoperator_create_dataset_and_table_helper.py y actualiza las variables para que reflejen tu proyecto y ubicación. Luego, ejecútalo con el siguiente comando:

python dataflowtemplateoperator_create_dataset_and_table_helper.py

Python

Para autenticarte en Managed Airflow, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura la autenticación para un entorno de desarrollo local.


# 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}")

Cree un bucket de Cloud Storage

Crea un bucket para almacenar todos los archivos necesarios para el flujo de trabajo. El DAG que crearás más adelante en esta guía hará referencia a los archivos que subas a este bucket de almacenamiento. Para crear un bucket de almacenamiento nuevo, sigue estos pasos:

Console

Abre Cloud Storage en la consola de Google Cloud .

Ir a Cloud Storage
Haz clic en Crear bucket para abrir el formulario de creación de buckets.
1. Ingresa la información de tu bucket y haz clic en Continuar para completar cada paso:
  - Especifica un Nombre único a nivel global para el bucket. En esta guía, se usa bucketName como ejemplo.
  - Selecciona Región para el tipo de ubicación. A continuación, selecciona una Ubicación en la que se almacenarán los datos del bucket.
  - Selecciona Estándar como la clase de almacenamiento predeterminada para tus datos.
  - Selecciona el control de acceso Uniforme para acceder a los objetos.
2. Haz clic en Listo.

gcloud

Usa el comando gcloud storage buckets create:

gcloud storage buckets create gs://bucketName/

Reemplaza lo siguiente:

bucketName: Es el nombre del bucket que creaste anteriormente en esta guía.

Muestras de código

C#


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

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

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

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

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 un esquema de BigQuery con formato JSON para la tabla de salida

Crea un archivo de esquema de BigQuery con formato JSON que coincida con la tabla de salida que creaste antes. Ten en cuenta que los nombres, los tipos y los modos de los campos deben coincidir con los que se definieron anteriormente en el esquema de tu tabla de BigQuery. Este archivo normalizará los datos de tu archivo .txt en un formato compatible con tu esquema de BigQuery. Asígnale el nombre jsonSchema.json a este archivo.

{
    "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"
    }]
}

Crea un archivo JavaScript para formatear tus datos

En este archivo, definirás tu UDF (función definida por el usuario) que proporciona la lógica para transformar las líneas de texto en tu archivo de entrada. Ten en cuenta que esta función toma cada línea de texto de tu archivo de entrada como su propio argumento, por lo que se ejecutará una vez para cada línea de tu archivo de entrada. Asígnale el nombre transformCSVtoJSON.js a este archivo.


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 tu archivo de entrada

Este archivo contendrá la información que deseas subir a tu tabla de BigQuery. Copia este archivo de forma local y asígnale el nombre 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

Sube tus archivos al bucket

Sube los siguientes archivos al bucket de Cloud Storage que creaste antes:

Esquema de BigQuery con formato JSON (.json)
Función definida por el usuario de JavaScript (transformCSVtoJSON.js)
El archivo de entrada del texto que deseas procesar (.txt)

Console

En la consola de Google Cloud , ve a la página Buckets de Cloud Storage.

Ir a Buckets
En la lista de buckets, haz clic en el tuyo.
En la pestaña Objetos del bucket, realiza una de las siguientes acciones:
- Arrastra y suelta los archivos deseados desde tu escritorio o administrador de archivos en el panel principal de la Google Cloud consola.
- Haz clic en el botón Subir archivos, selecciona los archivos que deseas subir en el cuadro de diálogo que aparece y haz clic en Abrir.

gcloud

Ejecuta el comando gcloud storage cp:

gcloud storage cp OBJECT_LOCATION gs://bucketName

Reemplaza lo siguiente:

bucketName: Es el nombre del bucket que creaste anteriormente en esta guía.
OBJECT_LOCATION: Es la ruta de acceso local a tu objeto. Por ejemplo, Desktop/transformCSVtoJSON.js.

Muestras de código

Python

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

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

Antes de ejecutar el DAG, establece las siguientes variables de Airflow.

Variable de Airflow	Valor
`project_id`	El ID del proyecto. Ejemplo: `example-project`.
`gce_zone`	Es la zona de Compute Engine en la que se debe crear el clúster de Dataflow. Ejemplo: `us-central1-a`. Para obtener más información sobre las zonas válidas, consulta Regiones y zonas.
`bucket_path`	Ubicación del bucket de Cloud Storage que creaste anteriormente. Ejemplo: `gs://example-bucket.`

Ahora harás referencia a los archivos que creaste antes para crear un DAG que inicie el flujo de trabajo de Dataflow. Copia este DAG y guárdalo de forma local como 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/",
        },
    )

Sube el DAG a Cloud Storage

Sube tu DAG a la carpeta /dags en el bucket de tu entorno. Una vez que se complete la carga correctamente, puedes verla haciendo clic en el vínculo DAGs Folder en la página Managed Airflow Environments.

La carpeta de DAGs de tu entorno contiene tu DAG

Cómo ver el estado de la tarea

Ve a la interfaz web de Airflow.
En la página de los DAG, haz clic en el nombre del DAG (por ejemplo, composerDataflowDAG).
En la página de detalles de los DAG, haz clic en Graph View.
Verifica el estado:
- Failed: La tarea tiene un cuadro rojo alrededor. También puedes mantener el puntero sobre la tarea y ver si aparece el mensaje Estado: con errores.
- Success: La tarea tiene un cuadro verde alrededor. También puedes mantener el puntero sobre la tarea y ver si aparece el mensaje State: Success.

Después de unos minutos, puedes verificar los resultados en Dataflow y BigQuery.

Visualiza tu trabajo en Dataflow

En la consola de Google Cloud , ve a la página Dataflow.

Ir a Dataflow
El trabajo se llama dataflow_operator_transform_csv_to_bq con un ID único adjunto al final del nombre con un guion, como el siguiente:
Haz clic en el nombre para ver los detalles del trabajo.

Visualiza tus resultados en BigQuery

En la consola de Google Cloud , ve a la página BigQuery.

Ir a BigQuery
Puedes enviar una consulta mediante SQL estándar. Usa la siguiente consulta para ver las filas que se agregaron a su tabla:
```
SELECT * FROM projectId.average_weather.average_weather
```

Inicia canalizaciones de Dataflow con Airflow administrado Organiza tus páginas con colecciones Guarda y categoriza el contenido según tus preferencias.

Descripción general

Antes de comenzar

Crea una tabla de BigQuery vacía con una definición de esquema

Console

bq

Python

Python

Cree un bucket de Cloud Storage

Console

gcloud

Muestras de código

C#

Go

Java

Python

Ruby

Crea un esquema de BigQuery con formato JSON para la tabla de salida

Crea un archivo JavaScript para formatear tus datos

Crea tu archivo de entrada

Sube tus archivos al bucket

Console

gcloud

Muestras de código

Python

Ruby

Configura DataflowTemplateOperator

Airflow 2

Airflow 1

Sube el DAG a Cloud Storage

Cómo ver el estado de la tarea

Visualiza tu trabajo en Dataflow

Visualiza tus resultados en BigQuery

¿Qué sigue?

Inicia canalizaciones de Dataflow con Airflow administrado