リアルタイム金融データパイプライン構築：Kafka、Flinkストリーミングアーキテクチャ実践ガイド

取引所/データベンダー --> Kafka Producer --> Kafka Cluster --> Flink Job --> 分析/アラート
                                               |                             |
                                          CDC Connector                 TimescaleDB
                                               |                             |
                                          レガシー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スキーマ定義
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",                # 全ISRレプリカ確認（耐久性）
            "linger.ms": 1,               # 1msバッチ待機（遅延vsスループット）
            "batch.size": 16384,          # 16KBバッチサイズ
            "compression.type": "lz4",    # LZ4圧縮（速度優先）
            "max.in.flight.requests.per.connection": 5,
            "enable.idempotence": True,   # 冪等性保証
            "retries": 3,
            "retry.backoff.ms": 100,
            "partitioner": "murmur2",     # シンボルベースパーティショニング
        })

    def send_market_data(self, symbol: str, data: dict):
        """市場データを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)  # 非同期コールバック処理
        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}")
        # 本番ではメトリクス収集（Prometheusカウンター等）

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

-- Kafkaソーステーブル定義
CREATE TABLE market_data_raw (
    symbol        STRING,
    price         DOUBLE,
    volume        BIGINT,
    bid           DOUBLE,
    ask           DOUBLE,
    event_time    TIMESTAMP(3),
    exchange      STRING,
    event_type    STRING,
    -- ウォーターマーク：最大5秒の遅延を許容
    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分間VWAP（出来高加重平均価格）集計
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),   // ウィンドウサイズ：5分
                Time.seconds(30)   // スライド間隔：30秒
            ))
            .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;

                    // ボラティリティが閾値を超えた場合アラート生成
                    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.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設定
    plugin.name: pgoutput
    slot.name: flink_cdc_slot
    publication.name: finance_pub

    # キャプチャ対象テーブル
    table.include.list: >
      public.orders,
      public.transactions,
      public.positions,
      public.account_balances

    # スキーマと変換
    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: initial
    snapshot.locking.mode: none

    # ハートビート（WAL保持防止）
    heartbeat.interval.ms: '10000'

    # トピックルーティング
    transforms: route
    transforms.route.type: org.apache.kafka.connect.transforms.RegexRouter
    transforms.route.regex: 'finance-cdc.public.(.*)'
    transforms.route.replacement: 'cdc.$1'

    # 障害復旧
    errors.tolerance: all
    errors.deadletterqueue.topic.name: cdc-dlq
    errors.deadletterqueue.context.headers.enable: true

# 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秒チェックポイント

t_env = StreamTableEnvironment.create(env)

# 異常取引検知SQL
anomaly_detection_sql = """
WITH price_stats AS (
    SELECT
        symbol,
        price,
        volume,
        event_time,
        -- 過去5分間の移動平均/標準偏差
        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スコアベースの異常スコア
        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'
"""

# アラートシンクテーブル（Kafkaトピックへ送信）
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 - 金融ワークロード最適化
taskmanager.memory.process.size: 8192m
taskmanager.memory.managed.fraction: 0.4
taskmanager.numberOfTaskSlots: 4

# 状態バックエンド（大規模状態向けRocksDB）
state.backend: rocksdb
state.backend.rocksdb.memory.managed: true
state.backend.incremental: true

# チェックポイント設定
execution.checkpointing.interval: 10s
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.min-pause: 5s
execution.checkpointing.timeout: 60s
execution.checkpointing.max-concurrent-checkpoints: 1

# ネットワークバッファ
taskmanager.network.memory.fraction: 0.1
taskmanager.network.memory.min: 128mb
taskmanager.network.memory.max: 1gb

# ウォーターマーク設定
pipeline.auto-watermark-interval: 200ms

# 再起動戦略
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 -- |
    |                        |                      |
    |   === ブローカー障害 ===  |                      |
    |  -- produce(msg) -->   X (失敗)               |
    |  -- retry -->          | (新リーダー)          |
    |  <-- ack (success) --  |                      |