Building Real-time Financial Data Pipelines: A Practical Guide to Kafka and Flink Streaming Architecture

Exchange/Data Vendor --> Kafka Producer --> Kafka Cluster --> Flink Job --> Analytics/Alerts
                                                |                              |
                                           CDC Connector                  TimescaleDB
                                                |                              |
                                           Legacy DB                     Iceberg Lake

from confluent_kafka import Producer, KafkaError
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
import json
import time

# Avro schema definition
MARKET_DATA_SCHEMA = """
{
  "type": "record",
  "name": "MarketData",
  "namespace": "com.finance.marketdata",
  "fields": [
    {"name": "symbol", "type": "string"},
    {"name": "price", "type": "double"},
    {"name": "volume", "type": "long"},
    {"name": "bid", "type": "double"},
    {"name": "ask", "type": "double"},
    {"name": "timestamp", "type": "long", "logicalType": "timestamp-millis"},
    {"name": "exchange", "type": "string"},
    {"name": "event_type", "type": {"type": "enum", "name": "EventType",
      "symbols": ["TRADE", "QUOTE", "BBO"]}}
  ]
}
"""

class MarketDataProducer:
    def __init__(self, bootstrap_servers: str, schema_registry_url: str):
        self.schema_registry = SchemaRegistryClient(
            {"url": schema_registry_url}
        )
        self.avro_serializer = AvroSerializer(
            self.schema_registry, MARKET_DATA_SCHEMA
        )

        self.producer = Producer({
            "bootstrap.servers": bootstrap_servers,
            "acks": "all",                # Wait for all ISR replicas (durability)
            "linger.ms": 1,               # 1ms batch wait (latency vs throughput)
            "batch.size": 16384,          # 16KB batch size
            "compression.type": "lz4",    # LZ4 compression (speed priority)
            "max.in.flight.requests.per.connection": 5,
            "enable.idempotence": True,   # Idempotent producer
            "retries": 3,
            "retry.backoff.ms": 100,
            "partitioner": "murmur2",     # Symbol-based partitioning
        })

    def send_market_data(self, symbol: str, data: dict):
        """Send market data to Kafka"""
        try:
            self.producer.produce(
                topic="market-data-raw",
                key=symbol.encode("utf-8"),
                value=self.avro_serializer(
                    data,
                    SerializationContext("market-data-raw", MessageField.VALUE)
                ),
                callback=self._delivery_callback,
                timestamp=int(time.time() * 1000),
            )
            self.producer.poll(0)  # Async callback processing
        except KafkaError as e:
            print(f"Kafka produce error: {e}")

    def _delivery_callback(self, err, msg):
        if err:
            print(f"Delivery failed for {msg.key()}: {err}")
        # In production: collect metrics (Prometheus counters, etc.)

    def flush(self):
        self.producer.flush(timeout=5)

-- Kafka source table definition
CREATE TABLE market_data_raw (
    symbol        STRING,
    price         DOUBLE,
    volume        BIGINT,
    bid           DOUBLE,
    ask           DOUBLE,
    event_time    TIMESTAMP(3),
    exchange      STRING,
    event_type    STRING,
    -- Watermark: allow up to 5 seconds of late data
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'market-data-raw',
    'properties.bootstrap.servers' = 'kafka-broker:9092',
    'properties.group.id' = 'flink-market-processor',
    'format' = 'avro-confluent',
    'avro-confluent.url' = 'http://schema-registry:8081',
    'scan.startup.mode' = 'latest-offset'
);

-- 1-minute VWAP (Volume Weighted Average Price) aggregation
CREATE TABLE vwap_1m (
    symbol        STRING,
    window_start  TIMESTAMP(3),
    window_end    TIMESTAMP(3),
    vwap          DOUBLE,
    total_volume  BIGINT,
    trade_count   BIGINT,
    high_price    DOUBLE,
    low_price     DOUBLE,
    PRIMARY KEY (symbol, window_start) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:postgresql://timescaledb:5432/market',
    'table-name' = 'vwap_1m'
);

INSERT INTO vwap_1m
SELECT
    symbol,
    window_start,
    window_end,
    SUM(price * volume) / SUM(volume) AS vwap,
    SUM(volume) AS total_volume,
    COUNT(*) AS trade_count,
    MAX(price) AS high_price,
    MIN(price) AS low_price
FROM TABLE(
    TUMBLE(TABLE market_data_raw, DESCRIPTOR(event_time), INTERVAL '1' MINUTE)
)
WHERE event_type = 'TRADE'
GROUP BY symbol, window_start, window_end;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class MovingAverageJob {

    public static DataStream<PriceAlert> computeMovingAverage(
            DataStream<MarketData> marketStream) {

        return marketStream
            .keyBy(MarketData::getSymbol)
            .window(SlidingEventTimeWindows.of(
                Time.minutes(5),   // Window size: 5 minutes
                Time.seconds(30)   // Slide interval: 30 seconds
            ))
            .process(new ProcessWindowFunction<MarketData, PriceAlert,
                     String, TimeWindow>() {
                @Override
                public void process(String symbol,
                        Context context,
                        Iterable<MarketData> elements,
                        Collector<PriceAlert> out) {

                    double sum = 0;
                    long count = 0;
                    double maxPrice = Double.MIN_VALUE;
                    double minPrice = Double.MAX_VALUE;

                    for (MarketData data : elements) {
                        sum += data.getPrice();
                        count++;
                        maxPrice = Math.max(maxPrice, data.getPrice());
                        minPrice = Math.min(minPrice, data.getPrice());
                    }

                    double ma = sum / count;
                    double volatility = (maxPrice - minPrice) / ma;

                    // Generate alert when volatility exceeds threshold
                    if (volatility > 0.02) {
                        out.collect(new PriceAlert(
                            symbol,
                            ma,
                            volatility,
                            context.window().getEnd(),
                            "HIGH_VOLATILITY"
                        ));
                    }
                }
            });
    }
}

# debezium-postgres-connector.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: finance-cdc-connector
  labels:
    strimzi.io/cluster: kafka-connect-cluster
spec:
  class: io.debezium.connector.postgresql.PostgresConnector
  tasksMax: 3
  config:
    # Database connection
    database.hostname: postgres-finance.internal
    database.port: '5432'
    database.user: cdc_reader
    database.password: 'VAULT_SECRET_REF'
    database.dbname: finance_core
    database.server.name: finance-cdc

    # CDC configuration
    plugin.name: pgoutput
    slot.name: flink_cdc_slot
    publication.name: finance_pub

    # Target tables
    table.include.list: >
      public.orders,
      public.transactions,
      public.positions,
      public.account_balances

    # Schema and transforms
    key.converter: io.confluent.connect.avro.AvroConverter
    value.converter: io.confluent.connect.avro.AvroConverter
    key.converter.schema.registry.url: http://schema-registry:8081
    value.converter.schema.registry.url: http://schema-registry:8081

    # Snapshot mode
    snapshot.mode: initial
    snapshot.locking.mode: none

    # Heartbeat (prevent WAL retention)
    heartbeat.interval.ms: '10000'

    # Topic routing
    transforms: route
    transforms.route.type: org.apache.kafka.connect.transforms.RegexRouter
    transforms.route.regex: 'finance-cdc.public.(.*)'
    transforms.route.replacement: 'cdc.$1'

    # Error handling
    errors.tolerance: all
    errors.deadletterqueue.topic.name: cdc-dlq
    errors.deadletterqueue.context.headers.enable: true

# Anomaly detection using PyFlink
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings
from pyflink.table.expressions import col, lit, call

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)
env.enable_checkpointing(10000)  # 10-second checkpoint interval

t_env = StreamTableEnvironment.create(env)

# Anomaly detection SQL
anomaly_detection_sql = """
WITH price_stats AS (
    SELECT
        symbol,
        price,
        volume,
        event_time,
        -- 5-minute moving average and standard deviation
        AVG(price) OVER w AS price_ma,
        STDDEV_POP(price) OVER w AS price_std,
        AVG(volume) OVER w AS volume_ma
    FROM market_data_raw
    WINDOW w AS (
        PARTITION BY symbol
        ORDER BY event_time
        RANGE BETWEEN INTERVAL '5' MINUTE PRECEDING AND CURRENT ROW
    )
),
anomalies AS (
    SELECT
        symbol,
        price,
        volume,
        event_time,
        price_ma,
        price_std,
        volume_ma,
        -- Z-Score based anomaly scoring
        ABS(price - price_ma) / NULLIF(price_std, 0) AS price_zscore,
        CAST(volume AS DOUBLE) / NULLIF(volume_ma, 0) AS volume_ratio,
        CASE
            WHEN ABS(price - price_ma) / NULLIF(price_std, 0) > 3.0
                THEN 'PRICE_SPIKE'
            WHEN CAST(volume AS DOUBLE) / NULLIF(volume_ma, 0) > 5.0
                THEN 'VOLUME_SURGE'
            WHEN ABS(price - price_ma) / NULLIF(price_std, 0) > 2.0
                AND CAST(volume AS DOUBLE) / NULLIF(volume_ma, 0) > 3.0
                THEN 'COMBINED_ANOMALY'
            ELSE 'NORMAL'
        END AS anomaly_type
    FROM price_stats
)
SELECT * FROM anomalies
WHERE anomaly_type <> 'NORMAL'
"""

# Alert sink table (published to Kafka topic)
t_env.execute_sql("""
CREATE TABLE anomaly_alerts (
    symbol        STRING,
    price         DOUBLE,
    volume        BIGINT,
    event_time    TIMESTAMP(3),
    anomaly_type  STRING,
    price_zscore  DOUBLE,
    volume_ratio  DOUBLE
) WITH (
    'connector' = 'kafka',
    'topic' = 'anomaly-alerts',
    'properties.bootstrap.servers' = 'kafka-broker:9092',
    'format' = 'json'
)
""")

# flink-conf.yaml - Financial workload optimization
taskmanager.memory.process.size: 8192m
taskmanager.memory.managed.fraction: 0.4
taskmanager.numberOfTaskSlots: 4

# State backend (RocksDB for large state)
state.backend: rocksdb
state.backend.rocksdb.memory.managed: true
state.backend.incremental: true

# Checkpoint configuration
execution.checkpointing.interval: 10s
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.min-pause: 5s
execution.checkpointing.timeout: 60s
execution.checkpointing.max-concurrent-checkpoints: 1

# Network buffers
taskmanager.network.memory.fraction: 0.1
taskmanager.network.memory.min: 128mb
taskmanager.network.memory.max: 1gb

# Watermark configuration
pipeline.auto-watermark-interval: 200ms

# Restart strategy
restart-strategy: exponential-delay
restart-strategy.exponential-delay.initial-backoff: 1s
restart-strategy.exponential-delay.max-backoff: 60s
restart-strategy.exponential-delay.backoff-multiplier: 2.0

[Producer]                [Kafka]               [Flink]
    |  -- produce(msg) -->   |                      |
    |  <-- ack (success) --  |                      |
    |                        |  -- consume(msg) -->  |
    |                        |                  [process]
    |                        |                  [checkpoint]
    |                        |  <-- commit offset -- |
    |                        |                      |
    |   === Broker Failure ===                      |
    |  -- produce(msg) -->   X (fail)               |
    |  -- retry -->          | (new leader)         |
    |  <-- ack (success) --  |                      |