1. 金融数据服务项目的整体架构设计思路1.1 为什么选择模块化分层架构做金融数据服务这类项目最怕的就是把数据采集、清洗、计算、存储、接口输出全部揉在一个大泥球里。我见过太多团队一开始图快把行情拉取和指标计算写在一个脚本里前三个月跑得挺欢等到要加一个新的数据源或者换一个计算口径时牵一发动全身改一处崩三处。所以这个项目从第一天起就定下了分层解耦的基调。具体怎么分我把它切成四层数据接入层、数据处理层、数据存储层、服务输出层。数据接入层只负责跟外部数据源打交道不管是公开的行情接口、第三方数据商的推送还是内部业务系统导出的文件统一在这一层做适配和归一化。处理层拿到标准化后的原始数据做清洗、对齐、指标计算。存储层根据数据的冷热程度和查询模式分别落到关系库、时序库和缓存里。服务输出层则对外提供统一的查询接口和订阅通道。这么分的好处在于每一层可以独立演进。比如某天要换掉一个行情源只需要在接入层加一个适配器处理层和存储层完全无感。再比如计算逻辑要调整处理层内部改就行不影响数据采集的稳定性。这种设计在金融场景下尤其重要因为金融数据对时效性和准确性的要求远高于一般业务数据任何一层的抖动都可能传导到最终用户看到的数字上。1.2 技术选型的取舍逻辑选型这块我踩过不少坑这里直接说结论和背后的考量。消息队列用的是Kafka。原因很直接金融数据天然是流式的行情推送、交易流水、风控事件都是持续不断产生的。Kafka的高吞吐和分区顺序性正好匹配这个场景而且它能做一定时间窗口的数据回放这对排查“某笔计算为什么得出这个结果”这类问题非常关键。有人会问为什么不用RabbitMQ答案是RabbitMQ在单分区顺序性和堆积能力上不如Kafka金融场景下消息积压是常态必须扛得住。计算引擎选了Flink做流处理批处理部分用Spark。流批分离听起来不够“现代”但实际落地时流计算和批计算的需求差异很大。流计算要求低延迟、Exactly-Once语义Flink在这块成熟度最高批处理更多是T1的报表和回溯计算Spark的生态和SQL支持更顺手。强行用一套引擎统一流批在金融这种对账要求极高的场景下反而容易出问题。存储是分三类走的。原始明细和需要事务保证的数据放PostgreSQL它的JSONB类型和窗口函数在金融数据处理里很好用。时序特征明显的数据比如行情快照、指标时间序列放TimescaleDB它本质是PostgreSQL的时序扩展学习成本低压缩比也不错。热数据缓存用Redis主要扛高频查询和去重。这里有个细节不要用Redis做唯一数据源它只做加速层所有数据必须能从前面的持久化存储里重建。服务框架用Go写核心接口Python写数据分析和模型部分。Go的并发模型适合处理大量并行的查询请求Python在数据科学工具链上无可替代。两者之间通过gRPC通信接口定义清晰性能也够。注意技术选型没有绝对的对错关键是匹配团队的技术储备和业务的真实需求。如果团队里没人懂Flink硬上Flink就是给自己挖坑这时候用Spark Structured Streaming也能解决大部分问题。1.3 数据流转的核心链路整个系统的数据流转可以概括为“采集-标准化-计算-存储-服务”五个环节。采集环节通过适配器模式对接不同数据源每个适配器实现统一的接口输出标准化的数据对象。标准化环节做字段映射、单位统一、时间对齐这一步是保证后续计算一致性的基础。计算环节分实时和离线两条线实时线走Flink做窗口聚合和指标计算离线线走Spark做全量回溯和复杂模型。存储环节根据数据特征路由到不同的存储引擎。服务环节对外暴露RESTful API和WebSocket订阅两种方式前者用于查询后者用于实时推送。这条链路里最容易出问题的是时间对齐。金融数据的时间戳来源五花八门有交易所时间、有本地时间、有数据商加工后的时间精度也不一样毫秒、微秒、甚至纳秒都有。我们的做法是统一转换成UTC毫秒时间戳并且在标准化环节记录原始时间戳和转换逻辑方便追溯。这个细节看似小但在做跨市场数据关联时时间对不齐会导致结果完全错误。2. 核心细节解析与实操要点2.1 数据接入层的适配器设计数据接入层是整个系统的入口它的稳定性直接决定了后续所有环节能否正常工作。我采用适配器模式每个数据源对应一个适配器实现所有适配器继承同一个基类基类定义了connect、subscribe、parse、normalize、healthCheck五个必须实现的方法。以接入一个行情数据源为例适配器的核心逻辑是这样的首先建立连接并完成认证然后根据配置订阅需要的标的收到原始消息后调用parse方法解析成内部数据结构再通过normalize方法做字段映射和单位转换最后把标准化后的数据推到消息队列。healthCheck方法定期检查连接状态一旦发现异常就触发重连和告警。这里有个实操要点适配器必须做限流和背压。金融数据源通常有频率限制无节制的拉取会被封禁。我们在适配器内部实现了令牌桶限流并且根据下游消费能力动态调整拉取速率。背压机制则是通过监控消息队列的堆积情况当堆积超过阈值时主动降低拉取频率避免雪崩。另一个容易忽略的点是断线重连的数据补偿。网络抖动导致连接断开后重连时如果只从当前时刻开始拉取中间断掉的数据就丢了。我们的做法是在适配器里记录最后一条成功处理的数据标识重连后从这个标识之后开始拉取确保数据不丢。对于不支持按标识回溯的数据源就依赖数据源自带的补数接口在重连后主动调用补数。2.2 数据清洗与标准化的关键规则数据清洗这块金融场景下的坑特别多。我总结了几条必须遵守的规则。第一空值和异常值的处理要区分对待。行情数据里出现价格为0或者成交量为负这明显是异常值必须过滤掉并记录日志。但有些字段的空值是有业务含义的比如某些衍生品没有涨跌停限制涨跌停价格字段就是空的这种空值不能当异常处理。我们的做法是在标准化环节为每个字段定义nullable属性和validRange只有超出有效范围的值才判定为异常。第二重复数据的去重逻辑要基于业务主键。金融数据经常出现重复推送比如行情快照可能因为网络重传收到两次。去重不能简单地按时间戳因为同一时间戳可能对应不同的数据内容。我们的做法是定义一个业务主键通常是“标的代码时间戳数据类型”在Redis里做短期去重同时保留一个滑动窗口用于处理延迟到达的重复数据。第三单位统一要建立映射表。不同数据源的价格单位可能不一样有的用元有的用分有的用基点。成交量有的用股有的用手。这些必须在标准化环节统一否则后续计算全错。我们维护了一张单位映射表每个数据源在配置里声明自己的单位标准化时自动转换。第四时间戳的精度和时区要统一。前面提过统一转UTC毫秒。但这里有个细节有些数据源给的是秒级时间戳转换时要乘以1000有些给的是字符串格式要按指定格式解析。这些都在适配器的normalize方法里处理并且记录原始值以便排查。2.3 实时计算中的窗口与状态管理实时计算部分用Flink做核心是窗口和状态管理。金融数据的实时计算需求主要有三类滑动窗口的统计指标如过去5分钟成交量、事件驱动的触发计算如价格突破阈值、以及会话窗口的聚合如一次交易会话的汇总。窗口选择上滑动窗口用于计算移动平均、移动最大最小值这类指标。窗口长度和滑动步长的选择要根据业务需求来比如计算5分钟移动平均窗口长度5分钟滑动步长可以设为1秒这样每秒都能得到最新的平均值。但滑动步长太小会导致计算量暴增需要权衡。我们的经验是滑动步长不小于1秒对于更高频的需求用Flink的ContinuousProcessingTimeTrigger做连续触发。状态管理是实时计算里最容易出问题的部分。Flink的状态分为Keyed State和Operator State金融场景下大量使用Keyed State按标的代码分组。状态大小会随着时间不断增长必须设置TTL。我们的做法是根据业务需求设置状态过期时间比如行情指标的状态保留最近24小时超期的自动清理。同时开启Flink的增量检查点避免每次检查点都全量快照导致性能抖动。提示Flink的状态后端选择RocksDB因为它支持超大状态并且可以溢写到磁盘。但RocksDB的性能依赖磁盘IO务必用SSD并且调整好writebuffer和blockcache参数。另一个实操要点是水位线的处理。金融数据可能存在乱序水位线决定了窗口何时触发计算。水位线设置太激进会漏掉延迟数据设置太保守会导致计算延迟。我们的做法是结合历史数据的延迟分布设置一个动态水位线允许一定程度的延迟同时对于超过水位线的迟到数据通过侧输出流单独处理不丢弃。2.4 存储层的冷热分离策略存储层的设计核心是冷热分离。热数据是最近几天的高频查询数据冷数据是历史归档数据。热数据放Redis和TimescaleDB冷数据放对象存储或者压缩后的PostgreSQL分区表。Redis的使用有几个要点。第一Key的设计要规范我们采用业务域:数据类型:标的:时间粒度的格式比如market:snapshot:000001.SZ:1m。这样便于批量操作和监控。第二过期时间要合理设置热数据的TTL根据查询频率来定高频查询的设长一点低频的设短一点。第三避免大Key和热Key大Key会导致操作阻塞热Key会导致单节点压力过大。我们的做法是对大Key做拆分对热Key做本地缓存或者多副本。TimescaleDB的使用主要是利用它的超表和连续聚合功能。超表自动按时间分区查询时只扫描相关分区效率很高。连续聚合可以预计算常用的聚合指标比如分钟级的OHLC查询时直接读预计算结果不用实时扫描原始数据。这里要注意连续聚合的刷新策略刷新频率太高会影响写入性能太低会导致数据不够新。我们的经验是分钟级聚合每5分钟刷新一次小时级聚合每30分钟刷新一次。冷数据的归档策略是按时间分区滚动归档。比如保留最近3个月的热数据超过3个月的数据自动归档到对象存储归档格式用Parquet压缩比高且支持列式查询。归档后的数据如果需要查询通过一个统一的查询网关路由对用户透明。3. 实操过程与核心环节实现3.1 环境搭建与基础依赖安装环境搭建这块我推荐用Docker Compose做本地开发环境生产环境用Kubernetes。本地开发环境的核心组件包括Kafka、Flink、PostgreSQL带TimescaleDB扩展、Redis、以及一个用于测试的数据模拟器。先装Docker和Docker Compose然后写docker-compose.yml。Kafka用单节点KRaft模式省去ZooKeeper的依赖。Flink用Session Cluster模式方便提交多个作业。PostgreSQL用TimescaleDB的官方镜像Redis用官方镜像并开启持久化。version: 3.8 services: kafka: image: bitnami/kafka:3.6 ports: - 9092:9092 environment: - KAFKA_CFG_NODE_ID0 - KAFKA_CFG_PROCESS_ROLEScontroller,broker - KAFKA_CFG_LISTENERSPLAINTEXT://:9092,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://localhost:9092 - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS0kafka:9093 - KAFKA_CFG_CONTROLLER_LISTENER_NAMESCONTROLLER timescaledb: image: timescale/timescaledb:latest-pg16 ports: - 5432:5432 environment: - POSTGRES_PASSWORDfinance123 volumes: - ./pgdata:/var/lib/postgresql/data redis: image: redis:7-alpine ports: - 6379:6379 command: redis-server --appendonly yes flink-jobmanager: image: flink:1.18-scala_2.12 ports: - 8081:8081 command: jobmanager environment: - FLINK_PROPERTIESjobmanager.rpc.address: flink-jobmanager flink-taskmanager: image: flink:1.18-scala_2.12 depends_on: - flink-jobmanager command: taskmanager environment: - FLINK_PROPERTIESjobmanager.rpc.address: flink-jobmanager - taskmanager.numberOfTaskSlots: 4启动后用docker-compose up -d拉起所有服务。然后初始化数据库创建超表和连续聚合。CREATE EXTENSION IF NOT EXISTS timescaledb; CREATE TABLE market_snapshot ( time TIMESTAMPTZ NOT NULL, symbol VARCHAR(20) NOT NULL, price NUMERIC(18,6), volume BIGINT, bid_price NUMERIC(18,6), ask_price NUMERIC(18,6) ); SELECT create_hypertable(market_snapshot, time); CREATE MATERIALIZED VIEW market_1m_ohlc WITH (timescaledb.continuous) AS SELECT time_bucket(1 minute, time) AS bucket, symbol, first(price, time) AS open, max(price) AS high, min(price) AS low, last(price, time) AS close, sum(volume) AS volume FROM market_snapshot GROUP BY bucket, symbol; SELECT add_continuous_aggregate_policy(market_1m_ohlc, start_offset INTERVAL 1 hour, end_offset INTERVAL 1 minute, schedule_interval INTERVAL 5 minutes);3.2 数据采集适配器的编码实现适配器用Python写因为大部分数据源的SDK都是Python的。基类定义如下from abc import ABC, abstractmethod from dataclasses import dataclass from typing import List, Optional import time dataclass class NormalizedRecord: symbol: str timestamp_ms: int data_type: str payload: dict source: str class BaseAdapter(ABC): def __init__(self, config: dict): self.config config self.last_processed_id: Optional[str] None self.rate_limiter TokenBucket( rateconfig.get(rate_limit, 100), capacityconfig.get(burst, 200) ) abstractmethod def connect(self) - bool: pass abstractmethod def subscribe(self, symbols: List[str]) - None: pass abstractmethod def parse(self, raw_message) - NormalizedRecord: pass abstractmethod def health_check(self) - bool: pass def normalize(self, record: NormalizedRecord) - NormalizedRecord: record.timestamp_ms self._to_utc_ms(record.timestamp_ms) record.payload self._convert_units(record.payload) return record def _to_utc_ms(self, ts) - int: if isinstance(ts, str): ts int(float(ts)) if ts 1e12: ts ts * 1000 return ts def _convert_units(self, payload: dict) - dict: unit_map self.config.get(unit_map, {}) for field, factor in unit_map.items(): if field in payload: payload[field] payload[field] * factor return payload以接入一个模拟行情源为例实现一个具体的适配器import json import random from kafka import KafkaProducer class MockMarketAdapter(BaseAdapter): def __init__(self, config): super().__init__(config) self.producer None self.symbols [] def connect(self) - bool: self.producer KafkaProducer( bootstrap_serversself.config[kafka_servers], value_serializerlambda v: json.dumps(v).encode(utf-8) ) return True def subscribe(self, symbols): self.symbols symbols def parse(self, raw_message): data json.loads(raw_message) return NormalizedRecord( symboldata[symbol], timestamp_msdata[ts], data_typesnapshot, payload{ price: data[price], volume: data[volume], bid_price: data[bid], ask_price: data[ask] }, sourcemock_market ) def health_check(self): return self.producer is not None def run(self): while True: for symbol in self.symbols: self.rate_limiter.acquire() raw json.dumps({ symbol: symbol, ts: int(time.time() * 1000), price: round(random.uniform(10, 100), 2), volume: random.randint(100, 10000), bid: round(random.uniform(9, 99), 2), ask: round(random.uniform(11, 101), 2) }) record self.parse(raw) record self.normalize(record) self.producer.send(market.raw, { symbol: record.symbol, timestamp_ms: record.timestamp_ms, data_type: record.data_type, payload: record.payload, source: record.source }) time.sleep(1)这个适配器每秒为每个标的生成一条模拟行情经过标准化后推到Kafka的market.raw主题。实际生产环境中run方法里的模拟逻辑替换成真实的数据源订阅即可。3.3 Flink实时计算作业的编写与提交Flink作业用Java写核心是从Kafka消费原始数据做窗口聚合然后写入TimescaleDB和Redis。public class MarketAggregationJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(30000); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000); env.setStateBackend(new RocksDBStateBackend(file:///tmp/flink-checkpoints)); Properties kafkaProps new Properties(); kafkaProps.setProperty(bootstrap.servers, localhost:9092); kafkaProps.setProperty(group.id, market-aggregation); DataStreamMarketEvent stream env .addSource(new FlinkKafkaConsumer(market.raw, new SimpleStringSchema(), kafkaProps)) .map(new MarketEventParser()) .assignTimestampsAndWatermarks( WatermarkStrategy.MarketEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) - event.getTimestampMs()) ); DataStreamOhlcResult ohlcStream stream .keyBy(MarketEvent::getSymbol) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new OhlcAggregator(), new OhlcWindowFunction()); ohlcStream.addSink(new TimescaleSink()); ohlcStream.addSink(new RedisSink()); env.execute(Market Aggregation Job); } }OhlcAggregator实现AggregateFunction接口维护开高低收和成交量的累加状态。OhlcWindowFunction在窗口触发时输出最终结果。TimescaleSink用JDBC批量写入RedisSink用Jedis写入缓存。提交作业的命令flink run -c com.finance.MarketAggregationJob \ -p 4 \ /path/to/market-aggregation.jar-p 4指定并行度为4根据TaskManager的slot数量调整。提交后可以在Flink Web UI的8081端口查看作业状态和指标。3.4 服务接口的开发与性能调优服务接口用Go写核心是查询接口和订阅接口。查询接口从TimescaleDB和Redis读数据订阅接口通过WebSocket推送实时数据。package main import ( database/sql encoding/json net/http github.com/go-redis/redis/v8 github.com/gorilla/websocket ) type Server struct { db *sql.DB rdb *redis.Client upgrader websocket.Upgrader } func (s *Server) GetOHLC(w http.ResponseWriter, r *http.Request) { symbol : r.URL.Query().Get(symbol) interval : r.URL.Query().Get(interval) limit : r.URL.Query().Get(limit) cacheKey : ohlc: symbol : interval : limit cached, err : s.rdb.Get(r.Context(), cacheKey).Result() if err nil { w.Header().Set(Content-Type, application/json) w.Write([]byte(cached)) return } rows, err : s.db.QueryContext(r.Context(), SELECT bucket, open, high, low, close, volume FROM market_1m_ohlc WHERE symbol $1 ORDER BY bucket DESC LIMIT $2, symbol, limit) if err ! nil { http.Error(w, err.Error(), 500) return } defer rows.Close() var results []map[string]interface{} for rows.Next() { var bucket time.Time var open, high, low, close float64 var volume int64 rows.Scan(bucket, open, high, low, close, volume) results append(results, map[string]interface{}{ time: bucket.UnixMilli(), open: open, high: high, low: low, close: close, volume: volume, }) } resp, _ : json.Marshal(results) s.rdb.Set(r.Context(), cacheKey, resp, 10*time.Second) w.Header().Set(Content-Type, application/json) w.Write(resp) }性能调优方面有几个关键点。第一数据库连接池要配好MaxOpenConns设为CPU核数的2到4倍MaxIdleConns设为相同值ConnMaxLifetime设为5分钟。第二Redis缓存要设合理的TTL行情数据变化快TTL设10秒左右比较合适太长了数据不新鲜太短了缓存命中率低。第三WebSocket推送要做批量合并不要每条数据都推一次而是攒够一定数量或者间隔一定时间推一次减少网络开销。注意Go的database/sql连接池在长时间空闲后可能持有失效连接务必设置ConnMaxLifetime并且在使用前做Ping检查。4. 常见问题与排查技巧实录4.1 数据不一致的排查思路数据不一致是金融数据服务里最头疼的问题。表现可能是同一个指标在不同接口返回的值不一样或者实时计算的结果和离线回溯的结果对不上。排查这类问题我一般按以下顺序走。第一步确认时间范围。实时计算和离线计算的时间窗口是否完全一致实时用的是事件时间还是处理时间离线用的是哪个时区这些对不上的话结果必然不一致。我们的做法是统一用事件时间并且所有时间都转UTC。第二步检查数据源。实时链路和离线链路是否用了同一个数据源如果实时从Kafka消费离线从数据库读而数据库的数据是Kafka消费后写入的那就要确认写入过程中有没有数据丢失或重复。我们的做法是在写入时记录消息的唯一标识离线计算前先做一次去重。第三步对比中间结果。在计算链路的每个环节打点记录输入和输出的数量和关键字段的校验和。比如实时链路在窗口聚合前记录输入记录数聚合后记录输出记录数离线链路同样打点两边对比就能定位到是哪一步出了问题。第四步检查状态。Flink的状态是否因为检查点恢复导致重复计算我们的做法是开启Exactly-Once并且在状态里记录已处理的窗口标识恢复时跳过已处理的窗口。4.2 实时计算延迟的优化手段实时计算延迟高用户看到的行情就是滞后的这在金融场景下是致命的。延迟的来源主要有几个数据源推送延迟、消息队列堆积、计算逻辑复杂、下游写入慢。数据源推送延迟通常没法控制但可以监控。我们在适配器里记录每条数据的产生时间和接收时间差值就是推送延迟超过阈值就告警。消息队列堆积是最常见的延迟来源。排查方法是看Kafka的消费者Lag如果Lag持续增长说明消费速度跟不上生产速度。解决办法有两个增加消费者并行度或者优化消费逻辑。增加并行度要注意分区数消费者数不能超过分区数。优化消费逻辑主要是减少不必要的反序列化和网络调用。计算逻辑复杂导致的延迟用Flink的Web UI看各个算子的繁忙程度。如果某个算子持续100%繁忙就是瓶颈。优化手段包括减少状态大小、用增量聚合代替全量聚合、把复杂计算拆成多个算子并行处理。下游写入慢导致的延迟通常是数据库写入瓶颈。优化手段包括批量写入代替单条写入、异步写入、增加数据库连接数、对写入表做分区。我们的经验是TimescaleDB的批量写入用COPY命令比INSERT快一个数量级。4.3 常见问题速查表问题现象可能原因排查方法解决方案数据重复消息重传、检查点恢复对比消息唯一标识开启Exactly-Once、业务主键去重数据丢失消费位点提交过早、异常未捕获对比生产消费数量手动提交位点、异常重试计算延迟高消息堆积、算子瓶颈看Kafka Lag和Flink UI增加并行度、优化算子查询超时数据库慢查询、缓存失效看慢查询日志、缓存命中率加索引、调TTL、预计算内存溢出状态过大、数据倾斜看GC日志、Flink状态大小设TTL、加盐打散、调内存时间对不齐时区不一致、精度不一致对比原始时间戳统一UTC毫秒、记录转换逻辑4.4 实操避坑经验分享坑一Kafka分区数设太少。一开始为了省事market.raw主题只设了3个分区结果数据量上来后消费跟不上。后来改成12个分区并行度提到12才勉强跟上。分区数一旦设定不能减少所以一开始就要预估好数据量宁可多设几个。坑二Flink状态TTL设太长。最初没设TTL状态无限增长跑了三天后TaskManager OOM。后来设了24小时TTL并且开启了增量检查点内存稳定了。TTL要根据业务需求设不是越长越好。坑三Redis大Key导致阻塞。有一次把某个标的的全部历史行情存成一个Hash结果这个Key有几十万字段每次操作都超时。后来拆成按天存储每天一个Key问题解决。Redis的Value不要超过10KBHash的字段数不要超过1000。坑四数据库连接池泄漏。Go服务里有个查询接口忘了defer rows.Close()跑了一段时间后连接池耗尽所有查询都超时。后来加了连接池监控WaitCount持续大于0就告警。写Go的数据库代码defer rows.Close()是肌肉记忆。坑五时区问题导致对账失败。数据源给的是本地时间我们存的是UTC对账时没转换差了8小时。后来在标准化环节强制转UTC并且在数据库里存TIMESTAMPTZ类型让数据库自己处理时区。坑六连续聚合刷新太频繁。TimescaleDB的连续聚合一开始设的1分钟刷新一次结果写入性能下降了一半。后来改成5分钟刷新一次写入恢复正常查询时数据最多滞后5分钟业务上可以接受。坑七WebSocket推送没做背压。行情剧烈波动时推送频率暴增客户端处理不过来连接被拖垮。后来加了推送队列和背压机制队列满了就丢弃旧数据保证连接稳定。这些坑都是我一个个踩过来的写出来是希望后来者能少走弯路。金融数据服务这个领域稳定性比功能丰富更重要宁可功能少一点也要保证核心链路不出问题。
金融数据服务架构实战：模块化分层与实时计算优化
1. 金融数据服务项目的整体架构设计思路1.1 为什么选择模块化分层架构做金融数据服务这类项目最怕的就是把数据采集、清洗、计算、存储、接口输出全部揉在一个大泥球里。我见过太多团队一开始图快把行情拉取和指标计算写在一个脚本里前三个月跑得挺欢等到要加一个新的数据源或者换一个计算口径时牵一发动全身改一处崩三处。所以这个项目从第一天起就定下了分层解耦的基调。具体怎么分我把它切成四层数据接入层、数据处理层、数据存储层、服务输出层。数据接入层只负责跟外部数据源打交道不管是公开的行情接口、第三方数据商的推送还是内部业务系统导出的文件统一在这一层做适配和归一化。处理层拿到标准化后的原始数据做清洗、对齐、指标计算。存储层根据数据的冷热程度和查询模式分别落到关系库、时序库和缓存里。服务输出层则对外提供统一的查询接口和订阅通道。这么分的好处在于每一层可以独立演进。比如某天要换掉一个行情源只需要在接入层加一个适配器处理层和存储层完全无感。再比如计算逻辑要调整处理层内部改就行不影响数据采集的稳定性。这种设计在金融场景下尤其重要因为金融数据对时效性和准确性的要求远高于一般业务数据任何一层的抖动都可能传导到最终用户看到的数字上。1.2 技术选型的取舍逻辑选型这块我踩过不少坑这里直接说结论和背后的考量。消息队列用的是Kafka。原因很直接金融数据天然是流式的行情推送、交易流水、风控事件都是持续不断产生的。Kafka的高吞吐和分区顺序性正好匹配这个场景而且它能做一定时间窗口的数据回放这对排查“某笔计算为什么得出这个结果”这类问题非常关键。有人会问为什么不用RabbitMQ答案是RabbitMQ在单分区顺序性和堆积能力上不如Kafka金融场景下消息积压是常态必须扛得住。计算引擎选了Flink做流处理批处理部分用Spark。流批分离听起来不够“现代”但实际落地时流计算和批计算的需求差异很大。流计算要求低延迟、Exactly-Once语义Flink在这块成熟度最高批处理更多是T1的报表和回溯计算Spark的生态和SQL支持更顺手。强行用一套引擎统一流批在金融这种对账要求极高的场景下反而容易出问题。存储是分三类走的。原始明细和需要事务保证的数据放PostgreSQL它的JSONB类型和窗口函数在金融数据处理里很好用。时序特征明显的数据比如行情快照、指标时间序列放TimescaleDB它本质是PostgreSQL的时序扩展学习成本低压缩比也不错。热数据缓存用Redis主要扛高频查询和去重。这里有个细节不要用Redis做唯一数据源它只做加速层所有数据必须能从前面的持久化存储里重建。服务框架用Go写核心接口Python写数据分析和模型部分。Go的并发模型适合处理大量并行的查询请求Python在数据科学工具链上无可替代。两者之间通过gRPC通信接口定义清晰性能也够。注意技术选型没有绝对的对错关键是匹配团队的技术储备和业务的真实需求。如果团队里没人懂Flink硬上Flink就是给自己挖坑这时候用Spark Structured Streaming也能解决大部分问题。1.3 数据流转的核心链路整个系统的数据流转可以概括为“采集-标准化-计算-存储-服务”五个环节。采集环节通过适配器模式对接不同数据源每个适配器实现统一的接口输出标准化的数据对象。标准化环节做字段映射、单位统一、时间对齐这一步是保证后续计算一致性的基础。计算环节分实时和离线两条线实时线走Flink做窗口聚合和指标计算离线线走Spark做全量回溯和复杂模型。存储环节根据数据特征路由到不同的存储引擎。服务环节对外暴露RESTful API和WebSocket订阅两种方式前者用于查询后者用于实时推送。这条链路里最容易出问题的是时间对齐。金融数据的时间戳来源五花八门有交易所时间、有本地时间、有数据商加工后的时间精度也不一样毫秒、微秒、甚至纳秒都有。我们的做法是统一转换成UTC毫秒时间戳并且在标准化环节记录原始时间戳和转换逻辑方便追溯。这个细节看似小但在做跨市场数据关联时时间对不齐会导致结果完全错误。2. 核心细节解析与实操要点2.1 数据接入层的适配器设计数据接入层是整个系统的入口它的稳定性直接决定了后续所有环节能否正常工作。我采用适配器模式每个数据源对应一个适配器实现所有适配器继承同一个基类基类定义了connect、subscribe、parse、normalize、healthCheck五个必须实现的方法。以接入一个行情数据源为例适配器的核心逻辑是这样的首先建立连接并完成认证然后根据配置订阅需要的标的收到原始消息后调用parse方法解析成内部数据结构再通过normalize方法做字段映射和单位转换最后把标准化后的数据推到消息队列。healthCheck方法定期检查连接状态一旦发现异常就触发重连和告警。这里有个实操要点适配器必须做限流和背压。金融数据源通常有频率限制无节制的拉取会被封禁。我们在适配器内部实现了令牌桶限流并且根据下游消费能力动态调整拉取速率。背压机制则是通过监控消息队列的堆积情况当堆积超过阈值时主动降低拉取频率避免雪崩。另一个容易忽略的点是断线重连的数据补偿。网络抖动导致连接断开后重连时如果只从当前时刻开始拉取中间断掉的数据就丢了。我们的做法是在适配器里记录最后一条成功处理的数据标识重连后从这个标识之后开始拉取确保数据不丢。对于不支持按标识回溯的数据源就依赖数据源自带的补数接口在重连后主动调用补数。2.2 数据清洗与标准化的关键规则数据清洗这块金融场景下的坑特别多。我总结了几条必须遵守的规则。第一空值和异常值的处理要区分对待。行情数据里出现价格为0或者成交量为负这明显是异常值必须过滤掉并记录日志。但有些字段的空值是有业务含义的比如某些衍生品没有涨跌停限制涨跌停价格字段就是空的这种空值不能当异常处理。我们的做法是在标准化环节为每个字段定义nullable属性和validRange只有超出有效范围的值才判定为异常。第二重复数据的去重逻辑要基于业务主键。金融数据经常出现重复推送比如行情快照可能因为网络重传收到两次。去重不能简单地按时间戳因为同一时间戳可能对应不同的数据内容。我们的做法是定义一个业务主键通常是“标的代码时间戳数据类型”在Redis里做短期去重同时保留一个滑动窗口用于处理延迟到达的重复数据。第三单位统一要建立映射表。不同数据源的价格单位可能不一样有的用元有的用分有的用基点。成交量有的用股有的用手。这些必须在标准化环节统一否则后续计算全错。我们维护了一张单位映射表每个数据源在配置里声明自己的单位标准化时自动转换。第四时间戳的精度和时区要统一。前面提过统一转UTC毫秒。但这里有个细节有些数据源给的是秒级时间戳转换时要乘以1000有些给的是字符串格式要按指定格式解析。这些都在适配器的normalize方法里处理并且记录原始值以便排查。2.3 实时计算中的窗口与状态管理实时计算部分用Flink做核心是窗口和状态管理。金融数据的实时计算需求主要有三类滑动窗口的统计指标如过去5分钟成交量、事件驱动的触发计算如价格突破阈值、以及会话窗口的聚合如一次交易会话的汇总。窗口选择上滑动窗口用于计算移动平均、移动最大最小值这类指标。窗口长度和滑动步长的选择要根据业务需求来比如计算5分钟移动平均窗口长度5分钟滑动步长可以设为1秒这样每秒都能得到最新的平均值。但滑动步长太小会导致计算量暴增需要权衡。我们的经验是滑动步长不小于1秒对于更高频的需求用Flink的ContinuousProcessingTimeTrigger做连续触发。状态管理是实时计算里最容易出问题的部分。Flink的状态分为Keyed State和Operator State金融场景下大量使用Keyed State按标的代码分组。状态大小会随着时间不断增长必须设置TTL。我们的做法是根据业务需求设置状态过期时间比如行情指标的状态保留最近24小时超期的自动清理。同时开启Flink的增量检查点避免每次检查点都全量快照导致性能抖动。提示Flink的状态后端选择RocksDB因为它支持超大状态并且可以溢写到磁盘。但RocksDB的性能依赖磁盘IO务必用SSD并且调整好writebuffer和blockcache参数。另一个实操要点是水位线的处理。金融数据可能存在乱序水位线决定了窗口何时触发计算。水位线设置太激进会漏掉延迟数据设置太保守会导致计算延迟。我们的做法是结合历史数据的延迟分布设置一个动态水位线允许一定程度的延迟同时对于超过水位线的迟到数据通过侧输出流单独处理不丢弃。2.4 存储层的冷热分离策略存储层的设计核心是冷热分离。热数据是最近几天的高频查询数据冷数据是历史归档数据。热数据放Redis和TimescaleDB冷数据放对象存储或者压缩后的PostgreSQL分区表。Redis的使用有几个要点。第一Key的设计要规范我们采用业务域:数据类型:标的:时间粒度的格式比如market:snapshot:000001.SZ:1m。这样便于批量操作和监控。第二过期时间要合理设置热数据的TTL根据查询频率来定高频查询的设长一点低频的设短一点。第三避免大Key和热Key大Key会导致操作阻塞热Key会导致单节点压力过大。我们的做法是对大Key做拆分对热Key做本地缓存或者多副本。TimescaleDB的使用主要是利用它的超表和连续聚合功能。超表自动按时间分区查询时只扫描相关分区效率很高。连续聚合可以预计算常用的聚合指标比如分钟级的OHLC查询时直接读预计算结果不用实时扫描原始数据。这里要注意连续聚合的刷新策略刷新频率太高会影响写入性能太低会导致数据不够新。我们的经验是分钟级聚合每5分钟刷新一次小时级聚合每30分钟刷新一次。冷数据的归档策略是按时间分区滚动归档。比如保留最近3个月的热数据超过3个月的数据自动归档到对象存储归档格式用Parquet压缩比高且支持列式查询。归档后的数据如果需要查询通过一个统一的查询网关路由对用户透明。3. 实操过程与核心环节实现3.1 环境搭建与基础依赖安装环境搭建这块我推荐用Docker Compose做本地开发环境生产环境用Kubernetes。本地开发环境的核心组件包括Kafka、Flink、PostgreSQL带TimescaleDB扩展、Redis、以及一个用于测试的数据模拟器。先装Docker和Docker Compose然后写docker-compose.yml。Kafka用单节点KRaft模式省去ZooKeeper的依赖。Flink用Session Cluster模式方便提交多个作业。PostgreSQL用TimescaleDB的官方镜像Redis用官方镜像并开启持久化。version: 3.8 services: kafka: image: bitnami/kafka:3.6 ports: - 9092:9092 environment: - KAFKA_CFG_NODE_ID0 - KAFKA_CFG_PROCESS_ROLEScontroller,broker - KAFKA_CFG_LISTENERSPLAINTEXT://:9092,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERSPLAINTEXT://localhost:9092 - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS0kafka:9093 - KAFKA_CFG_CONTROLLER_LISTENER_NAMESCONTROLLER timescaledb: image: timescale/timescaledb:latest-pg16 ports: - 5432:5432 environment: - POSTGRES_PASSWORDfinance123 volumes: - ./pgdata:/var/lib/postgresql/data redis: image: redis:7-alpine ports: - 6379:6379 command: redis-server --appendonly yes flink-jobmanager: image: flink:1.18-scala_2.12 ports: - 8081:8081 command: jobmanager environment: - FLINK_PROPERTIESjobmanager.rpc.address: flink-jobmanager flink-taskmanager: image: flink:1.18-scala_2.12 depends_on: - flink-jobmanager command: taskmanager environment: - FLINK_PROPERTIESjobmanager.rpc.address: flink-jobmanager - taskmanager.numberOfTaskSlots: 4启动后用docker-compose up -d拉起所有服务。然后初始化数据库创建超表和连续聚合。CREATE EXTENSION IF NOT EXISTS timescaledb; CREATE TABLE market_snapshot ( time TIMESTAMPTZ NOT NULL, symbol VARCHAR(20) NOT NULL, price NUMERIC(18,6), volume BIGINT, bid_price NUMERIC(18,6), ask_price NUMERIC(18,6) ); SELECT create_hypertable(market_snapshot, time); CREATE MATERIALIZED VIEW market_1m_ohlc WITH (timescaledb.continuous) AS SELECT time_bucket(1 minute, time) AS bucket, symbol, first(price, time) AS open, max(price) AS high, min(price) AS low, last(price, time) AS close, sum(volume) AS volume FROM market_snapshot GROUP BY bucket, symbol; SELECT add_continuous_aggregate_policy(market_1m_ohlc, start_offset INTERVAL 1 hour, end_offset INTERVAL 1 minute, schedule_interval INTERVAL 5 minutes);3.2 数据采集适配器的编码实现适配器用Python写因为大部分数据源的SDK都是Python的。基类定义如下from abc import ABC, abstractmethod from dataclasses import dataclass from typing import List, Optional import time dataclass class NormalizedRecord: symbol: str timestamp_ms: int data_type: str payload: dict source: str class BaseAdapter(ABC): def __init__(self, config: dict): self.config config self.last_processed_id: Optional[str] None self.rate_limiter TokenBucket( rateconfig.get(rate_limit, 100), capacityconfig.get(burst, 200) ) abstractmethod def connect(self) - bool: pass abstractmethod def subscribe(self, symbols: List[str]) - None: pass abstractmethod def parse(self, raw_message) - NormalizedRecord: pass abstractmethod def health_check(self) - bool: pass def normalize(self, record: NormalizedRecord) - NormalizedRecord: record.timestamp_ms self._to_utc_ms(record.timestamp_ms) record.payload self._convert_units(record.payload) return record def _to_utc_ms(self, ts) - int: if isinstance(ts, str): ts int(float(ts)) if ts 1e12: ts ts * 1000 return ts def _convert_units(self, payload: dict) - dict: unit_map self.config.get(unit_map, {}) for field, factor in unit_map.items(): if field in payload: payload[field] payload[field] * factor return payload以接入一个模拟行情源为例实现一个具体的适配器import json import random from kafka import KafkaProducer class MockMarketAdapter(BaseAdapter): def __init__(self, config): super().__init__(config) self.producer None self.symbols [] def connect(self) - bool: self.producer KafkaProducer( bootstrap_serversself.config[kafka_servers], value_serializerlambda v: json.dumps(v).encode(utf-8) ) return True def subscribe(self, symbols): self.symbols symbols def parse(self, raw_message): data json.loads(raw_message) return NormalizedRecord( symboldata[symbol], timestamp_msdata[ts], data_typesnapshot, payload{ price: data[price], volume: data[volume], bid_price: data[bid], ask_price: data[ask] }, sourcemock_market ) def health_check(self): return self.producer is not None def run(self): while True: for symbol in self.symbols: self.rate_limiter.acquire() raw json.dumps({ symbol: symbol, ts: int(time.time() * 1000), price: round(random.uniform(10, 100), 2), volume: random.randint(100, 10000), bid: round(random.uniform(9, 99), 2), ask: round(random.uniform(11, 101), 2) }) record self.parse(raw) record self.normalize(record) self.producer.send(market.raw, { symbol: record.symbol, timestamp_ms: record.timestamp_ms, data_type: record.data_type, payload: record.payload, source: record.source }) time.sleep(1)这个适配器每秒为每个标的生成一条模拟行情经过标准化后推到Kafka的market.raw主题。实际生产环境中run方法里的模拟逻辑替换成真实的数据源订阅即可。3.3 Flink实时计算作业的编写与提交Flink作业用Java写核心是从Kafka消费原始数据做窗口聚合然后写入TimescaleDB和Redis。public class MarketAggregationJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(30000); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000); env.setStateBackend(new RocksDBStateBackend(file:///tmp/flink-checkpoints)); Properties kafkaProps new Properties(); kafkaProps.setProperty(bootstrap.servers, localhost:9092); kafkaProps.setProperty(group.id, market-aggregation); DataStreamMarketEvent stream env .addSource(new FlinkKafkaConsumer(market.raw, new SimpleStringSchema(), kafkaProps)) .map(new MarketEventParser()) .assignTimestampsAndWatermarks( WatermarkStrategy.MarketEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) - event.getTimestampMs()) ); DataStreamOhlcResult ohlcStream stream .keyBy(MarketEvent::getSymbol) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new OhlcAggregator(), new OhlcWindowFunction()); ohlcStream.addSink(new TimescaleSink()); ohlcStream.addSink(new RedisSink()); env.execute(Market Aggregation Job); } }OhlcAggregator实现AggregateFunction接口维护开高低收和成交量的累加状态。OhlcWindowFunction在窗口触发时输出最终结果。TimescaleSink用JDBC批量写入RedisSink用Jedis写入缓存。提交作业的命令flink run -c com.finance.MarketAggregationJob \ -p 4 \ /path/to/market-aggregation.jar-p 4指定并行度为4根据TaskManager的slot数量调整。提交后可以在Flink Web UI的8081端口查看作业状态和指标。3.4 服务接口的开发与性能调优服务接口用Go写核心是查询接口和订阅接口。查询接口从TimescaleDB和Redis读数据订阅接口通过WebSocket推送实时数据。package main import ( database/sql encoding/json net/http github.com/go-redis/redis/v8 github.com/gorilla/websocket ) type Server struct { db *sql.DB rdb *redis.Client upgrader websocket.Upgrader } func (s *Server) GetOHLC(w http.ResponseWriter, r *http.Request) { symbol : r.URL.Query().Get(symbol) interval : r.URL.Query().Get(interval) limit : r.URL.Query().Get(limit) cacheKey : ohlc: symbol : interval : limit cached, err : s.rdb.Get(r.Context(), cacheKey).Result() if err nil { w.Header().Set(Content-Type, application/json) w.Write([]byte(cached)) return } rows, err : s.db.QueryContext(r.Context(), SELECT bucket, open, high, low, close, volume FROM market_1m_ohlc WHERE symbol $1 ORDER BY bucket DESC LIMIT $2, symbol, limit) if err ! nil { http.Error(w, err.Error(), 500) return } defer rows.Close() var results []map[string]interface{} for rows.Next() { var bucket time.Time var open, high, low, close float64 var volume int64 rows.Scan(bucket, open, high, low, close, volume) results append(results, map[string]interface{}{ time: bucket.UnixMilli(), open: open, high: high, low: low, close: close, volume: volume, }) } resp, _ : json.Marshal(results) s.rdb.Set(r.Context(), cacheKey, resp, 10*time.Second) w.Header().Set(Content-Type, application/json) w.Write(resp) }性能调优方面有几个关键点。第一数据库连接池要配好MaxOpenConns设为CPU核数的2到4倍MaxIdleConns设为相同值ConnMaxLifetime设为5分钟。第二Redis缓存要设合理的TTL行情数据变化快TTL设10秒左右比较合适太长了数据不新鲜太短了缓存命中率低。第三WebSocket推送要做批量合并不要每条数据都推一次而是攒够一定数量或者间隔一定时间推一次减少网络开销。注意Go的database/sql连接池在长时间空闲后可能持有失效连接务必设置ConnMaxLifetime并且在使用前做Ping检查。4. 常见问题与排查技巧实录4.1 数据不一致的排查思路数据不一致是金融数据服务里最头疼的问题。表现可能是同一个指标在不同接口返回的值不一样或者实时计算的结果和离线回溯的结果对不上。排查这类问题我一般按以下顺序走。第一步确认时间范围。实时计算和离线计算的时间窗口是否完全一致实时用的是事件时间还是处理时间离线用的是哪个时区这些对不上的话结果必然不一致。我们的做法是统一用事件时间并且所有时间都转UTC。第二步检查数据源。实时链路和离线链路是否用了同一个数据源如果实时从Kafka消费离线从数据库读而数据库的数据是Kafka消费后写入的那就要确认写入过程中有没有数据丢失或重复。我们的做法是在写入时记录消息的唯一标识离线计算前先做一次去重。第三步对比中间结果。在计算链路的每个环节打点记录输入和输出的数量和关键字段的校验和。比如实时链路在窗口聚合前记录输入记录数聚合后记录输出记录数离线链路同样打点两边对比就能定位到是哪一步出了问题。第四步检查状态。Flink的状态是否因为检查点恢复导致重复计算我们的做法是开启Exactly-Once并且在状态里记录已处理的窗口标识恢复时跳过已处理的窗口。4.2 实时计算延迟的优化手段实时计算延迟高用户看到的行情就是滞后的这在金融场景下是致命的。延迟的来源主要有几个数据源推送延迟、消息队列堆积、计算逻辑复杂、下游写入慢。数据源推送延迟通常没法控制但可以监控。我们在适配器里记录每条数据的产生时间和接收时间差值就是推送延迟超过阈值就告警。消息队列堆积是最常见的延迟来源。排查方法是看Kafka的消费者Lag如果Lag持续增长说明消费速度跟不上生产速度。解决办法有两个增加消费者并行度或者优化消费逻辑。增加并行度要注意分区数消费者数不能超过分区数。优化消费逻辑主要是减少不必要的反序列化和网络调用。计算逻辑复杂导致的延迟用Flink的Web UI看各个算子的繁忙程度。如果某个算子持续100%繁忙就是瓶颈。优化手段包括减少状态大小、用增量聚合代替全量聚合、把复杂计算拆成多个算子并行处理。下游写入慢导致的延迟通常是数据库写入瓶颈。优化手段包括批量写入代替单条写入、异步写入、增加数据库连接数、对写入表做分区。我们的经验是TimescaleDB的批量写入用COPY命令比INSERT快一个数量级。4.3 常见问题速查表问题现象可能原因排查方法解决方案数据重复消息重传、检查点恢复对比消息唯一标识开启Exactly-Once、业务主键去重数据丢失消费位点提交过早、异常未捕获对比生产消费数量手动提交位点、异常重试计算延迟高消息堆积、算子瓶颈看Kafka Lag和Flink UI增加并行度、优化算子查询超时数据库慢查询、缓存失效看慢查询日志、缓存命中率加索引、调TTL、预计算内存溢出状态过大、数据倾斜看GC日志、Flink状态大小设TTL、加盐打散、调内存时间对不齐时区不一致、精度不一致对比原始时间戳统一UTC毫秒、记录转换逻辑4.4 实操避坑经验分享坑一Kafka分区数设太少。一开始为了省事market.raw主题只设了3个分区结果数据量上来后消费跟不上。后来改成12个分区并行度提到12才勉强跟上。分区数一旦设定不能减少所以一开始就要预估好数据量宁可多设几个。坑二Flink状态TTL设太长。最初没设TTL状态无限增长跑了三天后TaskManager OOM。后来设了24小时TTL并且开启了增量检查点内存稳定了。TTL要根据业务需求设不是越长越好。坑三Redis大Key导致阻塞。有一次把某个标的的全部历史行情存成一个Hash结果这个Key有几十万字段每次操作都超时。后来拆成按天存储每天一个Key问题解决。Redis的Value不要超过10KBHash的字段数不要超过1000。坑四数据库连接池泄漏。Go服务里有个查询接口忘了defer rows.Close()跑了一段时间后连接池耗尽所有查询都超时。后来加了连接池监控WaitCount持续大于0就告警。写Go的数据库代码defer rows.Close()是肌肉记忆。坑五时区问题导致对账失败。数据源给的是本地时间我们存的是UTC对账时没转换差了8小时。后来在标准化环节强制转UTC并且在数据库里存TIMESTAMPTZ类型让数据库自己处理时区。坑六连续聚合刷新太频繁。TimescaleDB的连续聚合一开始设的1分钟刷新一次结果写入性能下降了一半。后来改成5分钟刷新一次写入恢复正常查询时数据最多滞后5分钟业务上可以接受。坑七WebSocket推送没做背压。行情剧烈波动时推送频率暴增客户端处理不过来连接被拖垮。后来加了推送队列和背压机制队列满了就丢弃旧数据保证连接稳定。这些坑都是我一个个踩过来的写出来是希望后来者能少走弯路。金融数据服务这个领域稳定性比功能丰富更重要宁可功能少一点也要保证核心链路不出问题。