Ingesta desde Kafka / Azure Event Hubs

Objetivo: aprender a consumir eventos desde Azure Event Hubs con Lakeflow, almacenarlos en una tabla Bronze y entender cómo se gestionan los offsets.

Esta vez utilizaremos un ejemplo muy parecido al de vuestra plataforma de datos: eventos de tickets que llegan desde Event Hubs y se almacenan en Delta Lake.

1. Arquitectura de nuestra ingesta

Databricks permite utilizar directamente el conector Kafka integrado para consumir Azure Event Hubs, sin instalar un conector específico de Event Hubs. Microsoft Learn

2. Primero: ¿qué devuelve Kafka?

Cuando Spark lee un evento, no devuelve directamente un ticket con campos como ticket_id o amount.

Devuelve una estructura similar a esta:

key y value son binarios. Si utilizamos Avro, tendremos que deserializarlos posteriormente. En esta lección conservaremos el mensaje original.

Esto es especialmente útil en Bronze: si mañana cambia el esquema Avro, seguimos conservando los bytes recibidos.

3. El código del pipeline

Creamos un archivo bronze_tickets.py en nuestro pipeline Lakeflow.

Para la conexión necesitamos un namespace de Event Hubs, el nombre del Event Hub y una política SAS con permiso Listen.

El ejemplo utiliza un Databricks Secret Scope llamado eventhubs que contiene la clave SAS bajo el nombre reader-key. Debes configurar previamente ambos valores; no escribas la clave real en el código.

from pyspark import pipelines as dp
from pyspark.sql import functions as F

# Configuración de Event Hubs
namespace = spark.conf.get("app.eh.namespace")
topic = spark.conf.get("app.eh.topic")
policy_name = spark.conf.get("app.eh.policy")

# Recuperamos la clave desde Secrets
policy_key = dbutils.secrets.get(
    scope="eventhubs",
    key="reader-key"
)

connection_string = (
    f"Endpoint=sb://{namespace}.servicebus.windows.net/;"
    f"SharedAccessKeyName={policy_name};"
    f"SharedAccessKey={policy_key}"
)

jaas = (
    "kafkashaded.org.apache.kafka.common.security."
    "plain.PlainLoginModule required "
    'username="$ConnectionString" '
    f'password="{connection_string}";'
)


@dp.table(name="bronze_tickets")
def bronze_tickets():

    df = (
        spark.readStream
        .format("kafka")
        .option(
            "kafka.bootstrap.servers",
            f"{namespace}.servicebus.windows.net:9093"
        )
        .option("subscribe", topic)
        .option("kafka.security.protocol", "SASL_SSL")
        .option("kafka.sasl.mechanism", "PLAIN")
        .option("kafka.sasl.jaas.config", jaas)
        .option("startingOffsets", "earliest")
        .load()
    )

    return df.select(
        "key",
        "value",
        "topic",
        "partition",
        "offset",
        F.col("timestamp").alias("kafka_timestamp"),
        F.current_timestamp().alias("ingest_time")
    )

La sintaxis de autenticación, incluido kafkashaded.org.apache.kafka..., sigue el ejemplo oficial de Azure Databricks. Microsoft Learn

En el apartado Configuration del pipeline, define los parámetros de ejemplo:

{
  "app.eh.namespace": "mi-namespace-dev",
  "app.eh.topic": "analytical_ticket",
  "app.eh.policy": "reader"
}

Son nombres ilustrativos: sustitúyelos por los recursos autorizados de tu entorno de desarrollo.

Para ejecutar el pipeline necesitas también conectividad de red al endpoint de Event Hubs, permisos para leer el secreto y permisos de escritura en el catálogo/esquema de destino.

4. ¿Qué ocurre con los offsets?

Aquí está una de las ventajas más importantes de Lakeflow.

Supongamos que Kafka contiene los offsets 100, 101, 102, 103 y 104.

En Lakeflow no tienes que escribir manualmente:

df.writeStream \
    .option("checkpointLocation", "...") \
    .start()

El pipeline administra el checkpoint del flujo.

Pero esto no significa que los offsets sean irrelevantes. Debes comprender cómo funcionan la retención de Event Hubs, el progreso del consumidor y la recuperación de fallos.

Dos detalles importantes:

• startingOffsets="earliest" se aplica al inicio de una consulta nueva, no reinicia el consumo desde cero en cada ejecución.

• Si Event Hubs elimina mensajes por retención antes de que el consumidor los lea, un checkpoint no permite recuperarlos.

5. Ejecutar y verificar Bronze

Configura el pipeline en modo Triggered, así procesaremos los eventos disponibles y terminaremos la ejecución.

Al finalizar, consulta la tabla de destino (sustituyendo el catálogo y esquema por los tuyos):

SELECT
    topic,
    partition,
    offset,
    kafka_timestamp,
    ingest_time,
    length(value) AS payload_bytes
FROM dev_catalog.demo_lakeflow.bronze_tickets
ORDER BY kafka_timestamp DESC
LIMIT 10;

Si hay eventos disponibles y la conexión funciona, verás sus metadatos.

Para inspeccionar los mensajes, puedes utilizar CAST(value AS STRING) únicamente si conoces que el payload es texto, como JSON. En vuestro caso, si es Avro binario, ese cast no lo deserializa correctamente.

6. Tres decisiones que tomaría para producción

En vuestro proyecto, donde tenéis múltiples Event Hubs y contratos Avro en Apicurio, mantener Bronze raw separado de la deserialización tiene bastante sentido.

De hecho, Databricks recomienda desacoplar los pipelines de ingesta de los de transformación para facilitar el mantenimiento y la recuperación de errores. Microsoft Learn

También conviene saber que existe un conector Kafka gestionado de Lakeflow Connect, actualmente documentado como Beta, además del enfoque programático que hemos utilizado. Microsoft Learn

7. Comprueba lo aprendido

Qué debes recordar: Lakeflow simplifica considerablemente la ingesta desde Event Hubs, pero no elimina la necesidad de comprender Kafka, offsets, retención y formatos de serialización.

En la lección 6 convertiremos esos bytes Avro en datos Silver utilizables, introduciendo los esquemas y los contratos de datos.