El Proyecto | Sobre nosotros | Contribuir | Donaciones | Licencia
HOME Mis cursos Finalizados
◄ Anterior
◄

Automatización de Flujos de Trabajo con Cloud Functions

Siguiente ►
►

Introducción

En este tema, aprenderemos cómo automatizar flujos de trabajo en BigQuery utilizando Google Cloud Functions. Las Cloud Functions son funciones ligeras y de propósito general que se ejecutan en la nube en respuesta a eventos. Son ideales para tareas de automatización debido a su capacidad para integrarse con otros servicios de Google Cloud y su facilidad de uso.

Objetivos

  • Comprender qué son las Cloud Functions y cómo funcionan.
  • Aprender a crear y desplegar una Cloud Function.
  • Integrar Cloud Functions con BigQuery para automatizar tareas.
  • Ejemplos prácticos de automatización de flujos de trabajo.

¿Qué son las Cloud Functions?

Google Cloud Functions es un servicio de computación sin servidor que permite ejecutar código en respuesta a eventos sin necesidad de gestionar servidores. Algunas características clave incluyen:

  • Ejecución basada en eventos: Las funciones se activan en respuesta a eventos de otros servicios de Google Cloud, como cambios en Cloud Storage, mensajes en Pub/Sub, o eventos HTTP.
  • Escalabilidad automática: Las funciones escalan automáticamente según la demanda.
  • Pago por uso: Solo pagas por el tiempo de ejecución de la función.

Creación y Despliegue de una Cloud Function

Paso 1: Configuración del Entorno

Antes de crear una Cloud Function, asegúrate de tener configurado tu entorno de Google Cloud:

  1. Instala la CLI de Google Cloud: Si no la tienes instalada, sigue las instrucciones en Google Cloud SDK.
  2. Autentícate en Google Cloud: Ejecuta gcloud auth login para autenticarte.
  3. Selecciona tu proyecto: Usa gcloud config set project [PROJECT_ID] para seleccionar tu proyecto.

Paso 2: Escribir la Función

Vamos a escribir una función simple que ejecuta una consulta en BigQuery y guarda los resultados en una tabla.

import google.cloud.bigquery as bigquery

def run_query(event, context):
    client = bigquery.Client()
    
    query = """
    SELECT name, COUNT(*) as name_count
    FROM `my_dataset.my_table`
    GROUP BY name
    """
    
    job = client.query(query)
    results = job.result()
    
    destination_table = client.dataset('my_dataset').table('results_table')
    job_config = bigquery.QueryJobConfig(destination=destination_table)
    
    job = client.query(query, job_config=job_config)
    job.result()  # Espera a que el trabajo termine.
    
    print(f"Query results loaded to table {destination_table}")

Paso 3: Desplegar la Función

Para desplegar la función, usa el siguiente comando:

gcloud functions deploy run_query \
--runtime python39 \
--trigger-http \
--allow-unauthenticated

Este comando despliega la función run_query con Python 3.9, activada por solicitudes HTTP y permite el acceso no autenticado.

Integración con BigQuery

Ejemplo Práctico: Automatización de una Tarea de Limpieza de Datos

Supongamos que queremos automatizar una tarea de limpieza de datos que se ejecuta cada vez que se carga un nuevo archivo en Cloud Storage.

  1. Escribir la Función:
import google.cloud.bigquery as bigquery
import google.cloud.storage as storage

def clean_data(event, context):
    client = bigquery.Client()
    storage_client = storage.Client()
    
    bucket_name = event['bucket']
    file_name = event['name']
    
    # Leer datos del archivo
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    data = blob.download_as_string()
    
    # Procesar y limpiar datos (ejemplo simple)
    cleaned_data = data.decode('utf-8').replace('old_value', 'new_value')
    
    # Guardar datos limpios en BigQuery
    table_id = 'my_dataset.cleaned_table'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,
        autodetect=True,
    )
    
    job = client.load_table_from_file(
        cleaned_data, table_id, job_config=job_config
    )
    
    job.result()  # Espera a que el trabajo termine.
    
    print(f"Data cleaned and loaded to table {table_id}")
  1. Desplegar la Función:
gcloud functions deploy clean_data \
--runtime python39 \
--trigger-resource [BUCKET_NAME] \
--trigger-event google.storage.object.finalize

Este comando despliega la función clean_data, que se activa cada vez que se carga un nuevo archivo en el bucket especificado.

Ejercicios Prácticos

Ejercicio 1: Crear una Cloud Function para Ejecutar una Consulta

Objetivo: Crear una Cloud Function que ejecute una consulta en BigQuery y guarde los resultados en una tabla.

  1. Escribe una función que ejecute una consulta simple en BigQuery.
  2. Despliega la función usando la CLI de Google Cloud.
  3. Prueba la función enviando una solicitud HTTP.

Ejercicio 2: Automatizar la Limpieza de Datos

Objetivo: Crear una Cloud Function que se active al cargar un archivo en Cloud Storage, limpie los datos y los guarde en BigQuery.

  1. Escribe una función que lea datos de un archivo en Cloud Storage.
  2. Limpia los datos (puedes usar cualquier lógica de limpieza que prefieras).
  3. Guarda los datos limpios en una tabla de BigQuery.
  4. Despliega la función y prueba cargando un archivo en el bucket.

Soluciones

Solución al Ejercicio 1

import google.cloud.bigquery as bigquery

def run_query(event, context):
    client = bigquery.Client()
    
    query = """
    SELECT name, COUNT(*) as name_count
    FROM `my_dataset.my_table`
    GROUP BY name
    """
    
    job = client.query(query)
    results = job.result()
    
    destination_table = client.dataset('my_dataset').table('results_table')
    job_config = bigquery.QueryJobConfig(destination=destination_table)
    
    job = client.query(query, job_config=job_config)
    job.result()  # Espera a que el trabajo termine.
    
    print(f"Query results loaded to table {destination_table}")

Solución al Ejercicio 2

import google.cloud.bigquery as bigquery
import google.cloud.storage as storage

def clean_data(event, context):
    client = bigquery.Client()
    storage_client = storage.Client()
    
    bucket_name = event['bucket']
    file_name = event['name']
    
    # Leer datos del archivo
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(file_name)
    data = blob.download_as_string()
    
    # Procesar y limpiar datos (ejemplo simple)
    cleaned_data = data.decode('utf-8').replace('old_value', 'new_value')
    
    # Guardar datos limpios en BigQuery
    table_id = 'my_dataset.cleaned_table'
    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,
        autodetect=True,
    )
    
    job = client.load_table_from_file(
        cleaned_data, table_id, job_config=job_config
    )
    
    job.result()  # Espera a que el trabajo termine.
    
    print(f"Data cleaned and loaded to table {table_id}")

Conclusión

En este tema, hemos aprendido cómo utilizar Google Cloud Functions para automatizar flujos de trabajo en BigQuery. Hemos cubierto desde la creación y despliegue de una función simple hasta la integración con BigQuery para tareas más complejas como la limpieza de datos. Con estas habilidades, puedes automatizar una variedad de tareas en tu entorno de BigQuery, mejorando la eficiencia y reduciendo la intervención manual.

◄ Anterior
◄
Siguiente ►
►

Curso de BigQuery

Módulo 1: Introducción a BigQuery

Módulo 2: SQL Básico en BigQuery

Módulo 3: SQL Intermedio en BigQuery

Módulo 4: SQL Avanzado en BigQuery

Módulo 5: Gestión de Datos en BigQuery

Módulo 6: Optimización del Rendimiento de BigQuery

Módulo 7: Seguridad y Cumplimiento en BigQuery

Módulo 8: Integración y Automatización de BigQuery

Módulo 9: Machine Learning en BigQuery (BQML)

Módulo 10: Casos de Uso de BigQuery en el Mundo Real

El Proyecto | Sobre nosotros | Contribuir | Donaciones | Licencia
© Copyright 2024. Todos los derechos reservados