Analítica en Tiempo Real

Descripción General

La analítica en tiempo real procesa datos a medida que llegan, entregando insights en segundos en lugar de horas o días. A diferencia de la analítica por batches que corre durante la noche, los sistemas en tiempo real ingieren eventos, computan agregaciones sobre la marcha, y actualizan dashboards continuamente. Esto habilita decisiones operacionales inmediatas: detección de fraude, análisis de comportamiento de usuario en vivo, monitoreo de IoT, y personalización en tiempo real.

A continuación: recolección de eventos, procesamiento de streams, bases de datos OLAP y diseño de dashboards para analítica sub-segundo.

Cuándo Usar

• For alternatives, see Complete Guide to LLM Application Architecture.

• Necesitas detectar anomalías o fraude en segundos de que ocurren eventos

• Operaciones de negocio dependen de visibilidad minuto a minuto (trading, logística, gaming)

• Los usuarios esperan dashboards en vivo que actualizan sin refrescar manualmente

• Dispositivos IoT transmiten telemetría que requiere respuesta inmediata

• Los motores de personalización necesitan comportamiento de usuario actual, no datos de ayer

• La latencia de batch (horas) causa oportunidades perdidas o reacciones tardías

Cuándo NO Usar

• Análisis de tendencias históricas donde minutos de delay son aceptables. ETL por batch es más simple
• Joins complejos multi-tabla a través de petabytes. Pre-agregación puede ser necesaria
• Reportes regulatorios requiriendo trazas completas de auditoría y reconciliación. Batch es más confiable
• Tu volumen de datos es lo suficientemente pequeño para que consultas de PostgreSQL completen en segundos sobre datos crudos

Conceptos Clave

Concepto	Descripción
Stream de Eventos	Flujo continuo de eventos desde productores a consumidores
OLAP	Online Analytical Processing — bases de datos optimizadas para agregaciones intensivas en lectura
Vista Materializada	Resultado de consulta precomputado que actualiza incrementalmente
Windowing	Agrupar eventos de stream en buckets basados en tiempo para agregación
Semántica Exactly-Once	Garantía de que cada evento se procesa una vez a pesar de fallos
Backpressure	Manejar casos donde los consumidores no pueden mantenerse al día con los productores

Arquitectura de Analítica en Tiempo Real

┌─────────────┐
│   Eventos   │  (Click, compra, lectura de sensor, llamada a API)
└──────┬──────┘
       │
┌──────▼──────┐
│   Kafka /   │  (Streaming de eventos, buffering, replay)
│   Kinesis   │
└──────┬──────┘
       │
   ┌───┴───┐
   │       │
┌──▼──┐ ┌──▼──┐
│Flink│ │Spark│  (Procesamiento de streams, agregaciones)
│/Kafka│ │Stream│
│Streams│ │     │
└──┬──┘ └──┬──┘
   │       │
   └───┬───┘
       │
┌──────▼──────┐
│ClickHouse / │  (Almacenamiento OLAP, consultas sub-segundo)
│Druid /      │
│Apache Pinot│
└──────┬──────┘
       │
┌──────▼──────┐
│ Dashboards  │  (Grafana, Superset, UI custom)
│   & APIs    │
└─────────────┘

Implementación de Analítica en Tiempo Real Paso a Paso

1. Colecta Eventos

Instrumenta tus aplicaciones para emitir eventos estructurados:

# Ejemplo: Productor de eventos Python con Kafka
from kafka import KafkaProducer
import json
import time

producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',           # Esperar todas las réplicas
    retries=3,
    max_block_ms=1000    # Fail fast si Kafka no disponible
)

def track_event(event_type, user_id, properties, timestamp=None):
    """Emitir un evento analítico estructurado."""
    event = {
        'event_type': event_type,
        'user_id': user_id,
        'timestamp': timestamp or time.time(),
        'properties': properties,
        'session_id': properties.get('session_id'),
        'device': properties.get('device'),
        'country': properties.get('country')
    }
    
    # Enviar a Kafka (no bloqueante con callback)
    future = producer.send('events', key=str(user_id).encode(), value=event)
    
    # Opcional: Agregar callback para confirmación de entrega
    future.add_callback(
        lambda metadata: print(f"Enviado a {metadata.topic} partición {metadata.partition}"
    ))
    future.add_errback(
        lambda exc: print(f"Fallo al enviar: {exc}"
    ))

# Uso
track_event('product_viewed', user_id=12345, properties={
    'product_id': 'sku-789',
    'category': 'electronics',
    'price': 299.99,
    'session_id': 'sess-abc',
    'device': 'mobile'
})

// Ejemplo: Tracking de eventos en navegador (liviano)
function trackEvent(eventType, properties) {
    const event = {
        event_type: eventType,
        timestamp: Date.now(),
        url: window.location.href,
        user_id: getUserId(),
        session_id: getSessionId(),
        properties: properties
    };
    
    // Enviar vía Beacon API (sobrevive unload de página)
    navigator.sendBeacon('/analytics/collect', JSON.stringify(event));
}

// Uso
trackEvent('button_clicked', { button_id: 'checkout', page: 'cart' });

Guía de Esquema de Eventos

Campo	Tipo	Requerido	Descripción
event_type	String	Sí	Nombre de evento categórico (product_viewed, purchase)
timestamp	Number	Sí	Timestamp Unix con precisión de milisegundo
user_id	String	Sí	Identificador de usuario único (hasheado para privacidad)
session_id	String	No	Agrupa eventos en sesiones de usuario
properties	Object	No	Datos específicos de evento (product_id, amount, category)
device	String	No	mobile, desktop, tablet
country	String	No	Código de país ISO para analítica geo

2. Procesa Streams con Windowing

Computa agregaciones sobre ventanas deslizantes o tumbling:

// Ejemplo: Kafka Streams para agregaciones en tiempo real
StreamsBuilder builder = new StreamsBuilder();

KStream<String, Event> events = builder.stream("events",
    Consumed.with(Serdes.String(), eventSerde));

// Ventana tumbling: buckets de 1 minuto
KTable<Windowed<String>, Long> pageViewsPerMinute = events
    .filter((key, event) -> "page_viewed".equals(event.getEventType()))
    .groupBy((key, event) -> event.getProperties().get("page_id"))
    .windowedBy(TimeWindows.of(Duration.ofMinutes(1)))
    .count(Materialized.as("page-view-counts"));

// Ventana deslizante: últimos 5 minutos, actualizada cada 10 segundos
KTable<Windowed<String>, Double> avgResponseTime = events
    .filter((key, event) -> "api_call".equals(event.getEventType()))
    .groupBy((key, event) -> event.getProperties().get("endpoint"))
    .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)))
    .aggregate(
        () -> new ResponseTimeStats(),
        (key, event, stats) -> stats.add(event.getProperties().get("response_time")),
        Materialized.as("response-time-stats")
    )
    .mapValues(ResponseTimeStats::getAverage);

// Escribir resultados a tópico de salida
pageViewsPerMinute.toStream()
    .to("analytics.page_views_per_minute", Produced.with(windowedSerde, Serdes.Long()));

// Ejemplo: Flink SQL para procesamiento de streams
from pyflink.table import StreamTableEnvironment
from pyflink.datastream import StreamExecutionEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

# Definir fuente Kafka
t_env.execute_sql("""
    CREATE TABLE events (
        event_type STRING,
        user_id STRING,
        timestamp TIMESTAMP(3),
        properties MAP<STRING, STRING>,
        WATERMARK FOR timestamp AS timestamp - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'events',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'json'
    )
""")

# Agregación con ventana tumbling
t_env.execute_sql("""
    CREATE TABLE page_views_per_minute (
        page_id STRING,
        view_count BIGINT,
        window_start TIMESTAMP(3),
        window_end TIMESTAMP(3),
        PRIMARY KEY (page_id, window_start, window_end) NOT ENFORCED
    ) WITH (
        'connector' = 'jdbc',
        'url' = 'jdbc:clickhouse://clickhouse:8123/analytics',
        'table-name' = 'page_views_per_minute',
        'driver' = 'ru.yandex.clickhouse.ClickHouseDriver'
    )
""")

t_env.execute_sql("""
    INSERT INTO page_views_per_minute
    SELECT 
        properties['page_id'] as page_id,
        COUNT(*) as view_count,
        TUMBLE_START(timestamp, INTERVAL '1' MINUTE) as window_start,
        TUMBLE_END(timestamp, INTERVAL '1' MINUTE) as window_end
    FROM events
    WHERE event_type = 'page_viewed'
    GROUP BY 
        properties['page_id'],
        TUMBLE(timestamp, INTERVAL '1' MINUTE)
""")

Tipos de Ventana

Tipo de Ventana	Comportamiento	Caso de Uso
Tumbling	Tamaño fijo, no superpuesta	Métricas por hora, conteos diarios
Sliding	Tamaño fijo, superpuesta	Promedios móviles, detección de tendencias
Session	Variable, gaps de inactividad	Análisis de sesión de usuario, tracking de funnel
Global	Todos los eventos, disparada manualmente	Contadores acumulativos, máquinas de estado
Watermark	Maneja eventos que llegan tarde	Streams de eventos fuera de orden

3. Almacena en Base de Datos OLAP

Elige una base de datos columnar optimizada para consultas analíticas:

// Ejemplo: Tabla ClickHouse para analítica de eventos
CREATE TABLE events (
    event_type LowCardinality(String),
    user_id UInt64,
    timestamp DateTime64(3),
    session_id UUID,
    properties String,  -- JSON como String, parsear con JSONExtract
    device LowCardinality(String),
    country LowCardinality(String),
    page_id LowCardinality(String),
    product_id LowCardinality(String),
    amount Decimal(10, 2)
) ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (event_type, timestamp, user_id)
TTL timestamp + INTERVAL 90 DAY;  -- Auto-eliminar datos viejos

// Vista materializada para vistas de página pre-agregadas
CREATE MATERIALIZED VIEW page_views_hourly
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMMDD(hour)
ORDER BY (page_id, hour)
AS SELECT
    toStartOfHour(timestamp) as hour,
    page_id,
    count() as views,
    uniqExact(user_id) as unique_users
FROM events
WHERE event_type = 'page_viewed'
GROUP BY hour, page_id;

// Consultar datos pre-agregados (sub-segundo)
SELECT 
    hour,
    page_id,
    views,
    unique_users
FROM page_views_hourly
WHERE hour >= now() - INTERVAL 24 HOUR
ORDER BY hour DESC, views DESC
LIMIT 100;

// Ejemplo: Especificación de ingestión de Druid
{
  "type": "kafka",
  "spec": {
    "dataSchema": {
      "dataSource": "events",
      "timestampSpec": {
        "column": "timestamp",
        "format": "iso"
      },
      "dimensionsSpec": {
        "dimensions": [
          "event_type",
          "user_id",
          "session_id",
          "device",
          "country",
          "page_id",
          "product_id",
          "category"
        ]
      },
      "metricsSpec": [
        { "type": "count", "name": "count" },
        { "type": "doubleSum", "name": "amount", "fieldName": "amount" },
        { "type": "thetaSketch", "name": "unique_users", "fieldName": "user_id" }
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "HOUR",
        "queryGranularity": "MINUTE"
      }
    }
  }
}

Comparación de Bases de Datos OLAP

Característica	ClickHouse	Apache Druid	Apache Pinot	BigQuery	Snowflake
Latencia	Sub-segundo	Sub-segundo	Sub-segundo	1-5 segundos	1-10 segundos
Auto-hospedado	Sí	Sí	Sí	No	No
Ingestión streaming	Nativa	Nativa	Nativa	Streaming API	Snowpipe
Soporte SQL	Completo	Druid SQL	Pinot SQL	Completo	Completo
Updates/deletes	Limitado	Limitado	Limitado	Completo	Completo
Mejor para	Series temporales	Multi-tenant	Orientado a usuarios	Ad-hoc	Empresarial
Modelo de costo	Hardware	Hardware	Hardware	Basado en consulta	Almacenamiento + computo

4. Construye Dashboards en Tiempo Real

Consulta bases de datos OLAP para visualizaciones en vivo:

// Queries de ClickHouse compatibles con Grafana

// Usuarios activos en tiempo real (últimos 5 minutos)
SELECT 
    toStartOfMinute(timestamp) as minute,
    uniqExact(user_id) as active_users
FROM events
WHERE timestamp >= now() - INTERVAL 5 MINUTE
GROUP BY minute
ORDER BY minute;

// Top productos por revenue (última hora)
SELECT 
    product_id,
    sum(amount) as revenue,
    count() as orders
FROM events
WHERE event_type = 'purchase'
  AND timestamp >= now() - INTERVAL 1 HOUR
GROUP BY product_id
ORDER BY revenue DESC
LIMIT 10;

// Funnel de conversión (últimos 30 minutos)
SELECT 
    event_type,
    count() as events,
    uniqExact(user_id) as unique_users
FROM events
WHERE event_type IN ('product_viewed', 'added_to_cart', 'checkout_started', 'purchase')
  AND timestamp >= now() - INTERVAL 30 MINUTE
GROUP BY event_type
ORDER BY 
    multiIf(
        event_type = 'product_viewed', 1,
        event_type = 'added_to_cart', 2,
        event_type = 'checkout_started', 3,
        event_type = 'purchase', 4,
        5
    );

// Distribución geo de tráfico actual
SELECT 
    country,
    count() as requests,
    uniqExact(user_id) as unique_users
FROM events
WHERE timestamp >= now() - INTERVAL 5 MINUTE
GROUP BY country
ORDER BY requests DESC
LIMIT 20;

Diseño de Dashboards para Tiempo Real

Patrón	Estrategia de Consulta	Frecuencia de Refresh
Contadores en vivo	SELECT count() FROM events WHERE timestamp > now() - 5m	5-10 segundos
Series de tiempo	Vista materializada pre-agregada	10-30 segundos
Listas Top-N	ORDER BY metric DESC LIMIT 10	30-60 segundos
Análisis de funnel	Filtrado multi-etapa con window functions	1-5 minutos
Alertas de anomalía	Detección estadística de anomalías en agregados	1 minuto

Lo que funciona

• Usa event-time, no processing-time. Skew de reloj y llegadas tardías hacen processing-time no confiable. Watermarks manejan datos tardíos elegantemente.
• Pre-agrega donde sea posible. Vistas materializadas en ClickHouse o agregaciones de Druid reducen costo de consulta 1000×.
• Elige el tamaño de ventana correcto. Demasiado pequeño = ruidoso; demasiado grande = insights retrasados. Empieza con ventanas tumbling de 1 minuto.
• Maneja backpressure. Si los consumidores se quedan atrás, escala horizontalmente o usa sampling (procesa 10% de eventos) en lugar de descartar datos.
• Evoluciona esquemas con cuidado. Agregar campos es fácil; remover o cambiar tipos requiere reprocesamiento o dual schemas.
• Monitorea latencia end-to-end. Desde generación de evento hasta visualización en dashboard. Alerta si la latencia excede tu SLA.

Errores Comunes

• Usar bases de datos transaccionales para analítica. PostgreSQL/MySQL no pueden manejar agregaciones de alta cardinalidad a escala.
• Sin validación de esquema de eventos. Eventos inválidos rompen silenciosamente agregaciones downstream.
• Processing-time en lugar de event-time. Dashboards muestran “ahora” pero los eventos son de hace 5 minutos debido a delays de red.
• Sobre-ingeniería para escala pequeña. Si tienes <100 eventos/segundo, PostgreSQL con índices apropiados puede ser suficiente.
• Ignorar datos tardíos. Sin watermarks, eventos tardíos corrompen agregaciones con ventana o son descartados.
• No configurar TTL. Crecimiento de datos sin límites destruye rendimiento de consulta y presupuestos de almacenamiento.

Variantes

• Arquitectura Lambda: Capa batch (Hadoop/Spark) + capa speed (Storm/Flink). Compleja, en gran parte reemplazada
• Arquitectura Kappa: Streaming puro con capacidad de reprocesamiento. Más simple, preferida hoy
• Batch + streaming híbrido: Flink/Spark para agregaciones complejas, vistas materializadas para conteos simples
• Cloud-native: Kinesis + Athena, Pub/Sub + BigQuery, Event Hubs + Synapse. Completamente gestionado

Conclusión

La analítica en tiempo real convierte streams de eventos en inteligencia accionable en segundos. Al instrumentar aplicaciones con eventos estructurados, procesarlos a través de agregaciones con ventana, y almacenar resultados en bases de datos OLAP, construyes sistemas que reaccionan al presente en lugar de reportar sobre el pasado.