GCP Data Engineer: Arquitectura Enterprise, Lakehouse en BigQuery y Streaming Resiliente

Matriz de Decisión: Motores de Cómputo e Ingesta en GCP

Uno de los desafíos más críticos para un GCP Data Engineer es seleccionar el servicio adecuado según la latencia requerida, el volumen de datos y el modelo de coste operativo (TCO).

Antipatrones Comunes y FinOps en Producción

En proyectos de escala petabyte, un error de diseño puede disparar la factura mensual de Google Cloud en miles de dólares o provocar cuellos de botella en la concurrencia de slots.

Implementación Práctica: Pipeline de Streaming Resiliente en Dataflow

El siguiente fragmento de código implementa un pipeline productivo en Python con Apache Beam para ejecutarse en Cloud Dataflow. Lee eventos desde Cloud Pub/Sub, maneja ventanas temporales fijas, enruta anomalías a una Dead Letter Queue (DLQ) mediante Side Outputs y escribe en BigQuery mediante la moderna Storage Write API.

# streaming_pipeline.py
# Pipeline de Streaming para GCP Dataflow con Apache Beam, DLQ y Storage Write API

import json
import logging
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions, SetupOptions
from apache_beam.transforms.window import FixedWindows
from apache_beam.io.gcp.bigquery import BigQueryDisposition, WriteToBigQuery

# Tag para enrutamiento de errores (Dead Letter Queue)
TAG_DEAD_LETTER = "dead_letter_output"

class ParseAndValidatePayload(beam.DoFn):
    """Transformación que valida el esquema JSON y separa datos corruptos hacia DLQ."""
    
    def process(self, element: bytes):
        try:
            payload = json.loads(element.decode("utf-8"))
            
            # Validación estricta de campos obligatorios
            if "event_id" not in payload or "timestamp" not in payload:
                raise ValueError("Faltan claves primarias obligatorias en el payload.")
                
            # Mutación/Enriquecimiento básico
            payload["processed_by"] = "dataflow-beam-worker"
            yield payload
            
        except Exception as err:
            # Enviar el registro corrupto junto con el mensaje de error a la DLQ
            error_record = {
                "raw_payload": element.decode("utf-8", errors="replace"),
                "error_message": str(err),
                "ingest_timestamp": beam.transforms.window.Timestamp.now().to_rfc3339()
            }
            yield beam.pvalue.TaggedOutput(TAG_DEAD_LETTER, error_record)


def run():
    # Configuración de Opciones de Pipeline para Dataflow
    pipeline_options = PipelineOptions()
    pipeline_options.view_as(StandardOptions).streaming = True
    
    google_cloud_options = pipeline_options.view_as(GoogleCloudOptions)
    google_cloud_options.project = "mi-proyecto-prod-data"
    google_cloud_options.region = "europe-west1"
    google_cloud_options.temp_location = "gs://mi-proyecto-prod-dataflow/temp"
    google_cloud_options.staging_location = "gs://mi-proyecto-prod-dataflow/staging"
    
    input_subscription = "projects/mi-proyecto-prod-data/subscriptions/telemetria-iot-sub"
    target_bq_table = "mi-proyecto-prod-data:lakehouse_bronze.events_telemetry"
    dlq_bq_table = "mi-proyecto-prod-data:lakehouse_bronze.events_dlq"

    with beam.Pipeline(options=pipeline_options) as p:
        
        # 1. Lectura desde Pub/Sub
        raw_events = (
            p 
            | "ReadFromPubSub" >> beam.io.ReadFromPubSub(subscription=input_subscription)
            | "FixedWindow" >> beam.WindowInto(FixedWindows(60))
        )

        # 2. Parseo y división con Side Outputs
        processed_data = (
            raw_events
            | "ParseAndValidate" >> beam.ParDo(ParseAndValidatePayload()).with_outputs(
                TAG_DEAD_LETTER, 
                main="valid_records"
            )
        )

        # 3. Escritura de Registros Válidos mediante BigQuery Storage Write API
        _ = (
            processed_data.valid_records
            | "WriteValidToBigQuery" >> WriteToBigQuery(
                table=target_bq_table,
                schema="event_id:STRING, timestamp:TIMESTAMP, metrics:JSON, processed_by:STRING",
                write_disposition=BigQueryDisposition.WRITE_APPEND,
                create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
                method=WriteToBigQuery.Method.STORAGE_WRITE_API  # Storage Write API de alto rendimiento
            )
        )

        # 4. Escritura de Errores en la tabla de Dead Letter Queue
        _ = (
            processed_data[TAG_DEAD_LETTER]
            | "WriteDLQToBigQuery" >> WriteToBigQuery(
                table=dlq_bq_table,
                schema="raw_payload:STRING, error_message:STRING, ingest_timestamp:TIMESTAMP",
                write_disposition=BigQueryDisposition.WRITE_APPEND,
                create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
                method=WriteToBigQuery.Method.STORAGE_WRITE_API
            )
        )

if __name__ == "__main__":
    logging.getLogger().setLevel(logging.INFO)
    run()

Patrones de Diseño: Arquitectura Lakehouse y Medallion en GCP

El modelo arquitectónico estándar para ingeniería de datos en Google Cloud se basa en la combinación de almacenamiento de objetos desacoplado (Cloud Storage / BigLake) con el motor de ejecución analítico de BigQuery.

Framework de Implementación Paso a Paso en un Entorno Enterprise

1. 1

   Infraestructura como Código con Terraform

   Define datasets, buckets con retention policies, VPC Service Controls (VPC-SC) para evitar la exfiltración de datos y cuentas de servicio con mínimos privilegios (IAM Least Privilege).

2. 2

   Diseño del Pipeline de Ingestión Resiliente

   Implementa topics de Pub/Sub con Dead Letter Topics habilitados nativamente y subscripciones de tipo BigQuery Subscription para casos directos o Cloud Dataflow para transformaciones en vuelo.

3. 3

   Orquestación y Linaje de Datos

   Configura Cloud Composer 2 o Cloud Workflows para coordinar las ejecuciones de dbt, disparar chequeos de validación con Great Expectations / Dataplex y registrar el linaje de datos en Dataplex Catalog.

4. 4

   Monitoreo, Alertas y FinOps Guardrails

   Establece presupuestos en Google Cloud Billing con alertas en Pub/Sub para pausar slots no esenciales, monitorea el slot_allocation y el retardo del Watermark en Dataflow con Cloud Monitoring dashboards.