Procesamiento de datos de alta frecuencia en servidores dedicados

El procesamiento de datos de alta frecuencia tiene una definición precisa: sistemas que deben recopilar, procesar y actuar sobre flujos de datos a velocidades que superan la capacidad de la infraestructura típica de alojamiento web. Fuentes de datos de los mercados financieros que llegan a 100 000 actualizaciones por segundo, redes de sensores industriales que transmiten telemetría desde miles de dispositivos a la vez, canalizaciones de agregación en tiempo real que deben reducir millones de eventos por minuto a resúmenes consultables: estas cargas de trabajo requieren hardware físico dedicado por razones que van más allá de la simple capacidad de la CPU.

Por qué el «bare metal» es importante para las cargas de trabajo de alta frecuencia

La infraestructura virtualizada introduce una latencia no determinista en los peores puntos posibles de un proceso de alta frecuencia. El programador del hipervisor determina cuándo se ejecutan las CPU virtuales de la máquina virtual. Bajo carga, una máquina virtual que compite con otros usuarios por el tiempo de CPU físico sufre retrasos de programación de entre 1 y 10 ms. Para las aplicaciones web, una fluctuación de programación de 5 ms pasa desapercibida. Para un procesador de flujos de datos financieros que debe reaccionar a los eventos del mercado en menos de 1 ms, una fluctuación de 5 ms en la programación es un problema que lo descarta por completo.

Los servidores dedicados «bare metal» eliminan por completo la capa del hipervisor. Tus procesos se ejecutan directamente en la CPU física, sin ningún intermediario de programación. En combinación con las opciones del núcleo en tiempo real de Linux, la fijación de afinidad de CPU y la asignación de memoria compatible con NUMA, los servidores dedicados pueden alcanzar una latencia de procesamiento inferior al milisegundo para cargas de trabajo de alta frecuencia, algo que la infraestructura virtualizada no puede igualar de forma fiable.

La documentación sobre la arquitectura del procesador EPYC de AMD señala que el diseño de chiplets del 4545P ofrece una latencia de acceso a la memoria uniforme en todos los núcleos, lo cual es importante para cargas de trabajo de alta frecuencia sensibles a NUMA, en las que los patrones de acceso a la memoria pueden determinar en gran medida el tiempo de procesamiento.

Caso de uso 1: Fuentes de datos de los mercados financieros

Los proveedores de datos financieros (Bloomberg, Refinitiv, CME Group) publican datos de mercado a velocidades que requieren una infraestructura de procesamiento específica. Un feed de acciones durante las horas de negociación puede generar entre 50 000 y 500 000 actualizaciones por segundo para miles de instrumentos.

Requisitos de procesamiento:

• Pila de red de baja latencia: las redes con omisión del núcleo (DPDK, RDMA) eliminan la sobrecarga de la pila TCP en las implementaciones más sensibles a la latencia; las redes estándar del núcleo son suficientes para la mayoría de los casos de uso por debajo del millón de mensajes por segundo
• Estructuras de datos sin bloqueos: las colas tradicionales basadas en mutex provocan conflictos cuando el volumen de mensajes es elevado; los búferes circulares sin bloqueos permiten que los subprocesos productores y consumidores funcionen sin bloquearse
• Afiliación de CPU: asigna el hilo de recepción de red y los hilos de procesamiento a núcleos de CPU específicos para eliminar la variabilidad en la programación

Implementación básica en Python de una cola de mensajes de alto rendimiento utilizando multiprocesamiento:

import multiprocessing as mp
from collections import deque
import time
 
class HighFrequencyProcessor:
    def __init__(self, num_workers=8):
        self.queue = mp.Queue(maxsize=100000)
        self.results = mp.Queue()
        self.workers = []
        
        # Pin workers to specific cores for consistent latency
        for i in range(num_workers):
            p = mp.Process(
                target=self._worker,
                args=(self.queue, self.results, i),
                daemon=True
            )
            p.start()
            self.workers.append(p)
    
    def _worker(self, queue, results, worker_id):
        # Set CPU affinity if psutil available
        try:
            import psutil
            psutil.Process().cpu_affinity([worker_id % mp.cpu_count()])
        except ImportError:
            pass
        
        while True:
            try:
                message = queue.get(timeout=0.001)
                result = self._process_message(message)
                results.put(result)
            except Exception:
                continue
    
    def _process_message(self, message):
        # Application-specific processing logic
        return {
            'timestamp': time.time_ns(),
            'symbol': message.get('symbol'),
            'price': message.get('price'),
            'processed': True
        }
    
    def ingest(self, message):
        try:
            self.queue.put_nowait(message)
            return True
        except mp.queues.Full:
            # Queue full - implement backpressure or drop strategy
            return False

import multiprocessing as mp
from collections import deque
import time
 
class HighFrequencyProcessor:
    def __init__(self, num_workers=8):
        self.queue = mp.Queue(maxsize=100000)
        self.results = mp.Queue()
        self.workers = []
        
        # Pin workers to specific cores for consistent latency
        for i in range(num_workers):
            p = mp.Process(
                target=self._worker,
                args=(self.queue, self.results, i),
                daemon=True
            )
            p.start()
            self.workers.append(p)
    
    def _worker(self, queue, results, worker_id):
        # Set CPU affinity if psutil available
        try:
            import psutil
            psutil.Process().cpu_affinity([worker_id % mp.cpu_count()])
        except ImportError:
            pass
        
        while True:
            try:
                message = queue.get(timeout=0.001)
                result = self._process_message(message)
                results.put(result)
            except Exception:
                continue
    
    def _process_message(self, message):
        # Application-specific processing logic
        return {
            'timestamp': time.time_ns(),
            'symbol': message.get('symbol'),
            'price': message.get('price'),
            'processed': True
        }
    
    def ingest(self, message):
        try:
            self.queue.put_nowait(message)
            return True
        except mp.queues.Full:
            # Queue full - implement backpressure or drop strategy
            return False

Para aplicaciones en las que la latencia de microsegundos es importante, Rust es el lenguaje preferido en Linux. Su modelo de gestión de memoria elimina las pausas de la recolección de basura, que de otro modo provocarían picos de latencia impredecibles en los peores momentos. El patrón de búfer circular de LMAX Disruptor ofrece una arquitectura de colas sin bloqueos de eficacia probada, con implementaciones de código abierto disponibles en Java (la implementación de referencia) y Rust. Vamos es una alternativa práctica para equipos que necesitan un rendimiento casi en tiempo real con primitivas de concurrencia más sencillas; su goroutine El programador gestiona miles de controladores de mensajes simultáneos sin necesidad de la gestión manual de subprocesos que requiere Python.

Caso de uso 2: Redes de sensores industriales

Las redes de sensores del IoT procedentes de equipos de fabricación, infraestructuras de redes inteligentes o sistemas de monitorización medioambiental generan grandes volúmenes de datos de telemetría que deben recopilarse, validarse y agregarse en tiempo real.

Una implementación típica del IoT industrial puede incluir 10 000 sensores que transmiten lecturas cada segundo: 10 000 mensajes por segundo de forma constante, con picos durante los eventos de detección de anomalías. El procesamiento de cada mensaje implica la normalización de la marca de tiempo, la conversión de unidades, la validación del rango y la agregación en un almacenamiento de series temporales.

InfluxDB es la base de datos de series temporales estándar para datos de sensores de alta frecuencia. Su formato de protocolo de línea está optimizado para escrituras de alto rendimiento:

# Escribir varios puntos en una sola solicitud HTTP (escrituras por lotes)
curl -i -XPOST 'http://localhost:8086/write?db=sensors&precision=ns' \
  --data-binary '
sensor_data,facility=plant1,device=temp_sensor_001 temperature=72.4,humidity=45.2 1675000000000000000
sensor_data,facility=plant1,device=temp_sensor_002 temperature=71.8,humidity=44.9 1675000000000000001
sensor_data,instalación=planta1,dispositivo=presión_001 presión=14,7,caudal=125,3 1675000000000000002'

# Write multiple points in a single HTTP request (batch writes)
curl -i -XPOST 'http://localhost:8086/write?db=sensors&precision=ns' \
  --data-binary '
sensor_data,facility=plant1,device=temp_sensor_001 temperature=72.4,humidity=45.2 1675000000000000000
sensor_data,facility=plant1,device=temp_sensor_002 temperature=71.8,humidity=44.9 1675000000000000001
sensor_data,facility=plant1,device=pressure_001 pressure=14.7,flow_rate=125.3 1675000000000000002'

Las escrituras por lotes superan con creces a las escrituras individuales cuando el volumen de mensajes es elevado. La documentación de InfluxDB sobre el rendimiento de las escrituras recomienda lotes de entre 5.000 y 10.000 puntos por solicitud de escritura para obtener el máximo rendimiento.

Kafka se sitúa antes de InfluxDB en la mayoría de los flujos de sensores en producción, actuando como un búfer de mensajes duradero que absorbe los picos de ingesta y permite que varios consumidores procesen el mismo flujo de datos con fines distintos:

# Crea un tema de Kafka para los datos de los sensores con la partición adecuada
kafka-topics.sh --create \
  --topic sensor-readings \
  --partitions 32 \          # Una partición por hilo de procesamiento
  --replication-factor 1 \   # Implementación en un solo servidor
  --bootstrap-server localhost:9092

# Create a Kafka topic for sensor data with appropriate partitioning
kafka-topics.sh --create \
  --topic sensor-readings \
  --partitions 32 \          # One partition per processing thread
  --replication-factor 1 \   # Single-server deployment
  --bootstrap-server localhost:9092

Las 32 particiones permiten que 32 subprocesos de consumo en paralelo procesen los datos de los sensores al mismo tiempo. En el servidor Extreme con un procesador EPYC de 16 núcleos (32 subprocesos), esto se traduce directamente en el máximo paralelismo sin sobresuscripción.

Caso de uso 3: Canalizaciones de agregación en tiempo real

Los canales de agregación convierten los flujos de eventos de alta velocidad en resúmenes que se pueden consultar: recuentos de visitas por página por minuto, totales de transacciones por hora, sesiones de usuarios activos por región. El reto consiste en calcular estas agregaciones en tiempo real mientras se procesan millones de eventos sin procesar cada hora.

Apache Flink y Apache Kafka Streams son las herramientas de código abierto estándar para la agregación de datos en tiempo real a gran escala. Para implementaciones de un solo servidor en hardware dedicado, Kafka Streams es más fácil de manejar (no requiere un clúster independiente) y ofrece prácticamente las mismas capacidades de agregación.

Un canal de agregación de Kafka Streams en Java:

StreamsBuilder builder = new StreamsBuilder();
 
// Read from input topic
KStream<String, PageViewEvent> pageViews = builder.stream("page-views");
 
// Aggregate into 1-minute tumbling windows
KTable<Windowed<String>, Long> viewCounts = pageViews
    .groupBy((key, value) -> value.getPageId())
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count(Materialized.as("page-view-counts"));
 
// Write aggregated results to output topic
viewCounts.toStream()
    .map((windowedKey, count) -> KeyValue.pair(
        windowedKey.key(),
        new AggregatedCount(windowedKey.window().startTime(), count)
    ))
    .to("page-view-aggregates");

StreamsBuilder builder = new StreamsBuilder();
 
// Read from input topic
KStream<String, PageViewEvent> pageViews = builder.stream("page-views");
 
// Aggregate into 1-minute tumbling windows
KTable<Windowed<String>, Long> viewCounts = pageViews
    .groupBy((key, value) -> value.getPageId())
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count(Materialized.as("page-view-counts"));
 
// Write aggregated results to output topic
viewCounts.toStream()
    .map((windowedKey, count) -> KeyValue.pair(
        windowedKey.key(),
        new AggregatedCount(windowedKey.window().startTime(), count)
    ))
    .to("page-view-aggregates");

Los estados de las agregaciones con ventanas consumen mucha memoria. Un proceso que mantenga ventanas móviles de una hora con 100 000 ID de página únicos requiere aproximadamente entre 1 y 2 GB de estado por etapa del proceso. Los 192 GB de RAM DDR5 del servidor Extreme ofrecen margen suficiente para ejecutar múltiples etapas de agregación con una generosa asignación de estado sin que se sature la memoria.

Optimización del hardware para cargas de trabajo de alta frecuencia en Linux

Hay varias opciones de configuración del núcleo de Linux y del hardware que resultan especialmente útiles para las cargas de trabajo de procesamiento de alta frecuencia.

Ajuste de la frecuencia de la CPU: el procesamiento a alta frecuencia se beneficia de velocidades de reloj constantes de la CPU. Desactiva el ajuste de frecuencia para evitar que los núcleos funcionen a una frecuencia reducida entre ráfagas:

# Configurar el regulador de rendimiento (funcionar siempre a la frecuencia máxima)
for cpu in /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor; do
    echo performance > $cpu
done
 
# Hacerlo permanente mediante cpupower
cpupower frequency-set -g performance

# Set performance governor (run at maximum frequency always)
for cpu in /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor; do
    echo performance > $cpu
done
 
# Make persistent via cpupower
cpupower frequency-set -g performance

Ten en cuenta el NUMA: El AMD EPYC 4545P utiliza una arquitectura de chiplets, en la que la latencia de acceso a la memoria varía en función del nodo NUMA al que esté asignada la memoria, en relación con el núcleo que accede a ella. Para cargas de trabajo sensibles a la latencia, asigna los subprocesos a núcleos que se encuentren dentro del mismo nodo NUMA que la memoria a la que acceden:

# Comprueba la topología NUMA
numactl --hardware
 
# Ejecuta un proceso con afinidad NUMA (vinculado a las CPU y la memoria del nodo 0)
numactl --cpunodebind=0 --membind=0 ./tu_procesador

# Check NUMA topology
numactl --hardware
 
# Run a process with NUMA affinity (bind to node 0 CPUs and memory)
numactl --cpunodebind=0 --membind=0 ./your_processor

Páginas gigantes: Las páginas de memoria predeterminadas del núcleo de Linux, de 4 KB, requieren muchas entradas en la TLB para conjuntos de trabajo grandes. Al habilitar las páginas gigantes de 2 MB, se reducen las faltas en la TLB durante el procesamiento que consume mucha memoria:

# Asigna 512 páginas enormes (512 x 2 MB = 1 GB)
echo 512 > /proc/sys/vm/nr_hugepages
 
# Para que se mantenga tras los reinicios
echo "vm.nr_hugepages = 512" >> /etc/sysctl.conf

# Allocate 512 huge pages (512 x 2MB = 1GB)
echo 512 > /proc/sys/vm/nr_hugepages
 
# Persistent across reboots
echo "vm.nr_hugepages = 512" >> /etc/sysctl.conf

Afiliación de IRQ: Para un procesamiento de red de alto rendimiento, asigna el manejo de las interrupciones de red a núcleos específicos de la CPU para evitar el «cache thrashing» cuando las interrupciones se gestionan en diferentes núcleos:

# Asignar las interrupciones de la tarjeta de red a los núcleos 0-3
# Primero, identifica los números de interrupción de la tarjeta de red
cat /proc/interrupts | grep eth0
 
# Establece la afinidad (ejemplo: la interrupción 23 al núcleo 0)
echo 1 > /proc/irq/23/smp_affinity

# Pin NIC interrupts to cores 0-3
# First identify NIC interrupt numbers
cat /proc/interrupts | grep eth0
 
# Set affinity (example for interrupt 23 to core 0)
echo 1 > /proc/irq/23/smp_affinity

Almacenamiento de datos de alta frecuencia

Las cargas de trabajo de alta frecuencia suelen generar grandes volúmenes de datos. Un flujo de datos financieros que procesa 100 000 actualizaciones por segundo, almacenando cada evento en 200 bytes, genera 20 MB por segundo, lo que equivale a 1,7 TB al día.

El servidor Extreme InMotion Hostingincluye 2 NVMe de 3,84 TB cada uno, lo que ofrece aproximadamente 4 días de almacenamiento bruto a este ritmo antes de que sea necesario archivar los datos. Para una retención más prolongada, configura una estrategia de almacenamiento por niveles:

• Almacenamiento en caliente (NVMe): últimas 48-72 horas de datos sin procesar, totalmente consultables
• Almacenamiento «warm» (almacenamiento de objetos): 30-90 días, comprimido, consultable con cierta latencia
• Almacenamiento en frío (archivo): más de 90 días, comprimido, recuperación lenta

El formato Apache Parquet ofrece compresión en columnas que reduce los datos financieros y de series temporales de sensores hasta un 10-20 % de su tamaño original, sin dejar de ser consultables mediante herramientas analíticas como Apache Spark, DuckDB o ClickHouse.

La infraestructura dedicada InMotion Hostingpara cargas de trabajo de alta frecuencia

La combinación del servidor Extreme, con un procesador AMD EPYC 4545P (16 núcleos, 32 subprocesos), 192 GB de RAM DDR5 ECC, 2SSD NVMe de 3,84 TB y una velocidad de puerto base de 3 Gbps (ampliable a 10 Gbps), da respuesta a las limitaciones específicas del procesamiento de datos a alta frecuencia: el paralelismo de la CPU para el procesamiento simultáneo de mensajes, el ancho de banda de la memoria para grandes almacenes de estado, NVMe para escrituras de alta velocidad y la capacidad de red para la ingesta de datos desde fuentes externas.

El puerto básico de 3 Gbps resulta especialmente útil para implementaciones de redes de sensores y agregadores de datos financieros, donde el volumen de datos entrantes es constante y no intermitente. Los equipos que necesiten un ancho de banda garantizado en lugar de capacidad para picos de tráfico pueden aumentar la velocidad del puerto en incrementos de 1 Gbps.

El hecho de que se trate de un servidor «bare metal» elimina las fluctuaciones en la programación del hipervisor, una característica que hace que los servidores dedicados sean especialmente adecuados para cargas de trabajo de procesamiento sensibles a la latencia que las máquinas virtuales en la nube no pueden gestionar de forma fiable. Para aplicaciones en las que la latencia de procesamiento se mide en microsegundos en lugar de milisegundos, la gama de servidores dedicados de InMotion ofrece la base de hardware que requieren las cargas de trabajo de alta frecuencia.