Procesamiento en Tiempo Real

El batch es suficiente para la mayoria de casos en LATAM. Pero ciertos escenarios requieren datos en tiempo real: deteccion de fraude, personalizacion instantanea, monitoreo y alertas operacionales.

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['kafka-broker:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all', retries=3
)

event = {
    'transaction_id': 'txn_abc123',
    'user_id': 'usr_456',
    'amount': 15000.50,
    'currency': 'ARS',
    'merchant': 'MercadoLibre',
    'country': 'AR',
    'timestamp': '2025-06-15T14:30:00Z'
}
producer.send('transactions', value=event)
producer.flush()

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName('FraudDetection').getOrCreate()

# Leer stream desde Kafka
df_stream = spark.readStream \
    .format('kafka') \
    .option('kafka.bootstrap.servers', 'kafka:9092') \
    .option('subscribe', 'transactions') \
    .load()

# Detectar anomalias: +5 transacciones en 1 minuto
anomalias = df_stream \
    .withWatermark('timestamp', '2 minutes') \
    .groupBy(F.window('timestamp', '1 minute'), 'user_id') \
    .agg(F.count('*').alias('num_txns'), F.sum('amount').alias('total')) \
    .filter(F.col('num_txns') > 5)

query = anomalias.writeStream.format('kafka') \
    .option('topic', 'fraud_alerts') \
    .option('checkpointLocation', '/checkpoints/fraud') \
    .start()

Empieza con batch, migra a micro-batch cuando necesites, y solo streaming real cuando el negocio lo justifique.