Apache Spark con Scala en GCP

Matriz de Decisión: Opciones de Spark en GCP

Seleccionar el modelo de computación correcto es una decisión crítica para equilibrar el throughput de procesamiento y los costes operativos en la nube:

Antipatrones Comunes y Fugas de Costes (FinOps)

Al migrar o construir pipelines de Spark con Scala sobre GCP, ignorar la naturaleza del almacenamiento distribuido y la capa de abstracción JVM puede desencadenar sobrecostes catastróficos y degradación severa de rendimiento.

Implementación Práctica: Pipeline Scala de Alto Rendimiento en GCP

A continuación se muestra un pipeline en Scala 2.13 que implementa lectura optimizada desde Google Cloud Storage (formato Parquet), transformaciones analíticas fuertemente tipadas y persistencia directa sobre BigQuery utilizando la API Storage Read/Write:

package com.datainsights.gcp.spark

import org.apache.spark.sql.{Dataset, Encoder, Encoders, SaveMode, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel

// Definición de dominio para tipado seguro en tiempo de compilación
case class TransactionEvent(
    transaction_id: String,
    customer_id: String,
    amount: Double,
    timestamp: Long,
    category: String
)

case class AggregatedMetric(
    customer_id: String,
    total_spend: Double,
    transaction_count: Long,
    avg_ticket: Double,
    dominant_category: String
)

object SparkGcpProductionJob {

  def main(args: Array[String]): Unit = {
    // 1. Inicialización de SparkSession optimizada para Dataproc / GCP
    val spark = SparkSession.builder()
      .appName("Spark-Scala-GCP-BigQuery-Optimization")
      // Habilitar Adaptive Query Execution (AQE)
      .config("spark.sql.adaptive.enabled", "true")
      .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
      .config("spark.sql.adaptive.skewJoin.enabled", "true")
      // Configuración de conector BigQuery indirecto/directo
      .config("viewsEnabled", "true")
      .config("materializationDataset", "dataproc_temp_staging")
      .getOrCreate()

    import spark.implicits._

    implicit val transactionEncoder: Encoder[TransactionEvent] = Encoders.product[TransactionEvent]
    implicit val metricEncoder: Encoder[AggregatedMetric] = Encoders.product[AggregatedMetric]

    val gcsInputPath = "gs://production-lakehouse-raw/events/transactions/year=2026/*"
    val bqTargetTable = "enterprise_dw.customer_financial_summary"
    val temporaryGcsBucket = "dataproc-staging-ephemeral-bucket"

    try {
      // 2. Lectura y deserialización a Dataset fuertemente tipado
      val rawDataset: Dataset[TransactionEvent] = spark.read
        .parquet(gcsInputPath)
        .as[TransactionEvent]

      // 3. Transformación analítica distribuida
      val aggregatedMetrics: Dataset[AggregatedMetric] = rawDataset
        .filter($"amount" > 0.0)
        .groupByKey(_.customer_id)
        .mapGroups { case (customerId, iterator) =>
          val events = iterator.toList
          val total = events.map(_.amount).sum
          val count = events.size
          val avg = if (count > 0) total / count else 0.0
          
          val topCategory = events
            .groupBy(_.category)
            .maxBy(_._2.size)
            ._1

          AggregatedMetric(
            customer_id = customerId,
            total_spend = BigDecimal(total).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble,
            transaction_count = count.toLong,
            avg_ticket = BigDecimal(avg).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble,
            dominant_category = topCategory
          )
        }

      // 4. Escritura optimizada hacia Google BigQuery
      aggregatedMetrics.write
        .format("bigquery")
        .option("table", bqTargetTable)
        .option("temporaryGcsBucket", temporaryGcsBucket)
        .option("writeMethod", "direct") // Usa BigQuery Storage Write API para ultra baja latencia
        .mode(SaveMode.Append)
        .save()

      println(s"Job completado exitosamente. Métricas persistidas en ${bqTargetTable}.")

    } catch {
      case ex: Exception =>
        System.err.println(s"Error crítico en ejecución distribuida: ${ex.getMessage}")
        ex.printStackTrace()
        System.exit(1)
    } finally {
      spark.stop()
    }
  }
}

Patrones de Diseño y Tuning de Memoria en GCP

Al ejecutar Scala sobre Spark en clusters Dataproc, la configuración de la JVM debe calibrarse meticulosamente para evitar cuellos de botella por Garbage Collection (GC pauses) y OOM (Out Of Memory) errors:

Checklist: Despliegue de Spark con Scala en Dataproc Serverless

Sigue este framework de validación antes de promover cualquier job de Scala a entornos productivos de Google Cloud:

• Artifact Packaging: Empaquetar la aplicación con sbt assembly creando un Fat-JAR con dependencias marcadas como provided para aquellas librerías nativas de Dataproc (spark-core, spark-sql).
• Service Account con Menor Privilegio (IAM): Asignar a la identidad del cluster únicamente los roles roles/dataproc.worker, roles/storage.objectAdmin (sobre buckets específicos) y roles/bigquery.dataEditor.
• Cloud Monitoring & Spark History Server: Conectar los logs de eventos hacia un bucket dedicado de GCS (spark.eventLog.dir) y habilitar la exportación de métricas de Cloud Monitoring para observabilidad de executors.
• VPC Service Controls: Asegurar que las subredes de Dataproc tengan habilitado Private Google Access para evitar tráfico por internet pública hacia BigQuery y GCS.

Preguntas Frecuentes (FAQ)