本文字数6400估计阅读时间17分钟作者Lionel Palacin编者按本文译自 ClickHouse 原博客链接点文末-查看原文。 原文围绕「用 ClickHouse 承载高频 tick 数据的实时写入、存储与查询。」展开。文章从报价与成交的基本概念切入逐步搭建数据链路对实时分析场景的选型有参考价值。技术栈组合清晰适合作为同类应用的落地范例。实时 tick 数据应用是实时分析的典型场景。追踪 Web 应用中的用户行为、监控物联网设备的指标都属于这一类面对高频事件流需要以低延迟完成写入、存储和查询。放到金融市场区别就在于紧迫性。哪怕只延迟几秒一笔本来盈利的交易也可能变成亏损。每一笔成交、每一次报价更新都会产生一个新的 tick多个标的加在一起每秒可能多达数千个。ClickHouse 很适合这类负载。它能处理高频写入、基于时间的查询和低延迟查询。内置压缩能降低存储开销哪怕每个标的有数十亿行也不成问题。物化视图可以在数据写入时就完成预聚合或重新组织从而优化查询性能无需额外增加处理层。下面一步步演示如何构建一个实时 tick 数据应用用 Massive原 Polygon.io获取市场数据用 ClickHouse 实时存储和查询 tick后端用 Node.js实时可视化用 React。下面正式开始。什么是 tick开始之前有必要先弄清楚报价quote和成交trade分别是什么。报价代表市场参与者当前愿意买入或卖出某只证券的价格具体包含最优买价best bid即有人愿意支付的最高价和最优卖价best ask即有人愿意接受的最低卖出价。随着新订单进入或退出市场报价会不断更新。symbxbpbsaxapascitqzinserted_atSPY12602.73211602.7461[1,93]174958347820663152225NYSE1749583479396成交则是买卖双方之间真实发生的交易。当有人接受当前的卖价或买价、订单被撮合成交时就产生了一笔成交。成交记录包含成交价、成交量和时间戳。symixpsctqztrfitrftinserted_atSPY5298352503482511607.261[12,37]175084225597222126NYSE001750842257842逐笔数据tick data通常来自两类数据流。一类是报价更新另一类是成交记录。两者对理解市场行为都不可或缺但在分析和策略开发中各有各的用处。获取实时市场数据清楚了要采集的数据类型接下来看怎么获取。首先要找到并订阅一个股票市场 API。可选的服务商很多本次演示选用的是 Massive。所选套餐必须明确包含通过 WebSocket 提供的股票 成交与报价。只有 REST API 密钥或者只含聚合数据的套餐都不够用。订阅前先确认当前的频道权限。WebSockets 对行情数据的流式传输至关重要因为它免去了轮询 REST API 的延迟与开销。每次请求数据都新建连接还可能漏掉两次调用之间的逐笔更新WebSocket 则维持一条长连接数据一就绪就推送过来。对于毫秒必争的高频行情数据这一点尤为关键。用 Massive API 开始采集数据并不复杂与 /stocks 端点建立连接用你的 Massive API 密钥完成认证然后开始处理消息。下面这个独立的 Node.js 诊断脚本会先完成认证认证成功后再订阅并记录收到的事件。装好依赖、配置好.env之后在示例目录下运行它。启动完整应用之前要先把它停掉账户的连接数上限会导致多个客户端无法共用同一路数据。require(dotenv).config(); const WebSocket require(ws); if (!process.env.MASSIVE_API_KEY) throw new Error(Set MASSIVE_API_KEY in .env); const symbols (process.env.MASSIVE_SYMBOLS || AAPL,MSFT,NVDA) .split(,).map((symbol) symbol.trim()).filter(Boolean); const ws new WebSocket(wss://socket.massive.com/stocks); ws.on(open, () { ws.send(JSON.stringify({ action: auth, params: process.env.MASSIVE_API_KEY })); }); ws.on(message, (data) { const payload JSON.parse(data.toString()); for (const row of payload) { if (row.ev status) { console.log(row.status, row.message); if (row.status auth_success) { ws.send(JSON.stringify({ action: subscribe, params: symbols.flatMap((symbol) [T.${symbol}, Q.${symbol}]).join(,), })); } else if (row.status auth_failed) { ws.close(); } } else if (row.ev T || row.ev Q) { console.log(row); } } }); ws.on(error, (error) console.error(error.message)); process.on(SIGINT, () ws.close());将数据写入 ClickHouse在 ClickHouse 中为逐笔数据建模Tick 数据的建模相对直接它只有两类事件一类是 trade一类是 quote每类事件的字段都不多而且大多是数值型。下面是创建两张独立表的 DDL。t是 SIP 时间戳单位为毫秒q是每个标的的序列号。condition 和 indicator 数组用UInt32因为其取值可能超过 255。trade 表结构还保留了可选的ds小数数量文本形式和pt参与者时间戳字段。可选字段缺失时使用默认值。当前 quote 的 size 以股数表示。示例中价格用Float64并对整数 trade 数量s求和如果你的应用有需要应改用合适的 decimal 类型并对小数成交量单独处理。CREATE TABLE IF NOT EXISTS quotes ( sym LowCardinality(String), bx UInt8, bp Float64, bs UInt64, ax UInt8, ap Float64, as UInt64, c UInt8, i Array(UInt32), t UInt64, q UInt64, z Enum8(NYSE 1, AMEX 2, Nasdaq 3), inserted_at UInt64 DEFAULT toUnixTimestamp64Milli(now64()) ) ENGINE MergeTree ORDER BY (sym, t - (t % 60000)); CREATE TABLE IF NOT EXISTS trades ( sym LowCardinality(String), i String, x UInt8, p Float64, s UInt64, ds String DEFAULT , pt UInt64 DEFAULT 0, c Array(UInt32), t UInt64, q UInt64, z Enum8(NYSE 1, AMEX 2, Nasdaq 3), trfi UInt64, trft UInt64, inserted_at UInt64 DEFAULT toUnixTimestamp64Milli(now64()) ) ENGINE MergeTree ORDER BY (sym, t - (t % 60000));数据量会增长得很快。开发阶段先订阅少量标的测好吞吐量之后再订阅整个市场。选择合适的 order key 对性能至关重要。在本例中行首先按 sym股票代码排序把同一标的的所有事件归到一起。在每个标的分组内行按t - (t % 60000)排序从而形成 1 分钟的时间桶。这种做法在这里很合适因为我们是按标的聚合数据来生成可视化的。分钟桶把时间相近的事件聚在一起而查询仍然用原始时间戳和序列号来确定事件顺序。排序键和过滤条件要结合实际负载做基准测试单靠时间桶并不能保证每个查询都能高效裁剪。数据写入策略为这类应用设计数据写入管道有多种方式比如借助 Kafka 这样的消息队列。不过为了尽量降低延迟通常更好的做法是保持系统简单只要条件允许就直接把数据从 WebSocket 连接推入 ClickHouse。这套方案就绪之后下一步是选择合适的写入方式。ClickHouse 同时支持同步插入和异步插入。对于同步写入数据先在客户端攒批再发送出去。批大小batch size要在内存占用、延迟和系统开销之间取得平衡。批次越大插入请求越少吞吐越高但内存占用会增加单条记录的延迟也会变大。批次越小内存压力越小但会产生过多小数据分片反而给 ClickHouse 带来更多负载。异步写入时数据持续发往 ClickHouse攒批由内部完成。到达的记录先写入内存缓冲区再按可配置的阈值刷写到存储。当客户端不方便攒批时——比如数据来自大量小客户端——这种方式很合适。在这个场景里同步写入更合适。数据只由一个客户端从 WebSocket API 推送过来批处理可以放在客户端做从而更好地控制性能和资源占用。下面这个精简的攒批类展示了写入路径。创建一个TickBatcher把收到的 WebSocket 帧连同上面的认证处理程序一起传给它的handleMessage方法。在 awaitclose()之前先停止接收消息。完整示例另外加上了有界并发插入、重连处理、关闭超时和指标。两者都用同步插入因此成功就意味着 ClickHouse 已确认写入。const { createClient } require(clickhouse/client); class TickBatcher { constructor() { this.client createClient({ url: process.env.CLICKHOUSE_URL || http://localhost:8123, username: process.env.CLICKHOUSE_USERNAME || default, password: process.env.CLICKHOUSE_PASSWORD || , database: process.env.CLICKHOUSE_DATABASE || default, compression: { request: true, response: true }, clickhouse_settings: { async_insert: 0 }, }); this.batches { trades: [], quotes: [] }; this.flushing { trades: false, quotes: false }; this.failedRecords 0; this.droppedRecords 0; this.timer setInterval(() { void this.flushBatch(trades); void this.flushBatch(quotes); }, 2000); } handleMessage(data) { const payload JSON.parse(data.toString()); for (const { ev, ...fields } of payload) { const table ev T ? trades : ev Q ? quotes : null; if (!table) continue; if (this.batches[table].length 10000) { this.droppedRecords; continue; } this.batches[table].push(fields); if (this.batches[table].length 1000) void this.flushBatch(table); } } async flushBatch(table) { if (this.flushing[table]) return; this.flushing[table] true; try { while (this.batches[table].length) { const values this.batches[table].splice(0, 1000); try { await this.client.insert({ table, values, format: JSONEachRow }); } catch (error) { this.failedRecords values.length; console.error(Insert failed for ${table}:, error.message); } } } finally { this.flushing[table] false; } } async close() { clearInterval(this.timer); while (Object.values(this.flushing).some(Boolean)) { await new Promise((resolve) setTimeout(resolve, 10)); } await Promise.all([this.flushBatch(trades), this.flushBatch(quotes)]); await this.client.close(); } }这个精简类把每张待写入表的缓冲区限制在 10,000 条记录并统计被丢弃或失败的记录失败的写入不会重试。要做到不丢数据还需要持久化缓冲、重放和去重。完整示例在 1,000 行或每两秒触发一次刷写最多同时执行四个插入。可视化实时市场数据数据存进 ClickHouse 之后搭建可视化层就很直接了。真正的难点在于写出正确的 SQL 查询。下面看看具体怎么做。我们聚焦支撑两个关键可视化所需的查询。第一个是实时表格持续更新展示某只股票的最新交易数据。要做这个可视化一个查询就够了数据可以用 ClickHouse 强大的 SQL 查询语言和自定义函数来格式化。with {syms: Array(String)} as symbols, toDate(now(America/New_York)) as curr_day, trades_info as ( select sym, argMax(p, tuple(t, q)) as last_price, round(((last_price - (argMin(p, tuple(t, q)))) / nullIf(argMin(p, tuple(t, q)), 0)) * 100, 2) as change_pct, sum(s) as total_volume, max(t) as latest_t from trades where toDate(fromUnixTimestamp64Milli(toInt64(t), America/New_York)) curr_day and sym in symbols group by sym order by sym asc ), quotes_info as ( select sym, argMax(bp, tuple(t, q)) as bid, argMax(ap, tuple(t, q)) as ask, max(t) as latest_t from quotes where toDate(fromUnixTimestamp64Milli(toInt64(t), America/New_York)) curr_day and sym in symbols group by sym order by sym asc ) select t.sym as ticker, t.last_price as last, q.bid as bid, q.ask as ask, t.change_pct as change, t.total_volume as volume, toUnixTimestamp64Milli(now64()) - toInt64(greatest(t.latest_t, q.latest_t)) as latency from trades_info as t left join quotes_info as q on t.sym q.sym;下面拆解一下这个查询做了什么。首先定义两个变量symbols保存待分析的股票代码列表curr_day记录纽约时区的当前日期。查询随后获取交易数据包括• last_price 最新成交价使用 argMax(p, tuple(t, q)) 在毫秒时间戳相同时按序列号打破平局• change_pct 相对于当前纽约日历日第一笔已摄取交易的百分比变化带有零分母保护。这不是官方开盘价或前收盘价。• total_volume 该日历日已摄取的总整数股成交量它还获取报价数据• bid 最新买价使用 argMax(bp, tuple(t, q))• ask 最新卖价使用 argMax(ap, tuple(t, q))最后将交易和报价结果合并。额外的latency列报告最新收到事件距今的毫秒数而不是只测量数据库插入时间。下列行是示例历史值。该演示不筛选交易条件不处理取消/更正也不重建官方交易所的 OHLCV 序列。tickerlastbidaskchangevolumeNVDA151.2099151.2151.212.1765269276我们要分析的第二个可视化是蜡烛图展示某只股票的价格走势和成交量。我们来看看驱动这一可视化的 SQL 查询。select toUnixTimestamp64Milli(toDateTime64(toStartOfInterval(fromUnixTimestamp64Milli(toInt64(t)), interval 2 minute), 3)) as x, argMin(p, tuple(t, q)) as o, max(p) as h, min(p) as l, argMax(p, tuple(t, q)) as c, sum(s) as v from trades where x toUnixTimestamp64Milli(now64() - interval 1 hour) and sym {sym: String} group by x order by x asc;该查询将最近一小时按两分钟分桶并计算开盘价、最高价、最低价、收盘价和成交量。开盘价和收盘价使用(t, q)来处理时间戳相同的交易。桶过滤器排除了窗口起始处部分重叠的桶。浏览器调用一个命名查询的 Express 端点只接受服务器定义、参数化的 SQL。ClickHouse 凭据保留在服务器上绝不放入NEXT_PUBLIC_变量。为了可视化查询结果我们使用 click-ui 组件显示表格使用 Chart.js 绘制蜡烛图。扩展与实用技巧在生产环境中处理高频市场数据不仅需要快速数据库。以下技巧和方法有助于确保系统在数据量增长时保持高性能和可靠。扩展数据摄取处理多个交易标的的 tick 级数据时持续吞吐量很容易超过每秒数万条记录。要应对这种情况可以从以下几个方面入手• 使用 客户端批量写入 批量大小按系统的内存和延迟限制调到最优。• 使用 压缩 压缩插入数据能缩小网络传输的载荷体积降低带宽占用加快传输速度。• 监控 ClickHouse 中创建的 part 数量避免合并过于频繁。这篇 博客 讲的是异步插入但其中关于 part 创建的部分同样适用于同步写入。你也可以借助 高级仪表盘 来 监控 数据 part 的数量。• Massive 也提供了处理高吞吐数据消费的 性能建议 。监控写入延迟前面提到金融应用对数据的实时性要求很高所以监控延迟很有必要。你可以很方便地计算并跟踪事件时间戳tick 发生的时刻与写入时间戳数据被存储的时刻之间的差值。这个「事件到写入」的延迟涵盖行情源传输、客户端批量处理、网络耗时和数据库写入并非单纯的数据库延迟。下面的查询统计最近一小时内每个交易标的最新成交的该延迟。使用有符号运算可以避免时间戳出现意外差异时发生无符号下溢。SELECT sym, count() AS trade_count, argMax(toInt64(inserted_at) - toInt64(t), tuple(t, q)) AS ingest_latency_ms FROM trades WHERE t toUnixTimestamp64Milli(now64() - INTERVAL 1 HOUR) GROUP BY sym ORDER BY trade_count DESC LIMIT 100;把这个指标可视化到仪表盘上有助于及早发现变慢的迹象。善用物化视图Materialized views 适合在数据到达时就预先聚合。对于依赖按时间汇总的特定查询模式这能带来明显的优化。典型例子是按固定间隔比如 1 分钟窗口计算金融数据的 OHLCV开盘价、最高价、最低价、收盘价、成交量指标。在写入时生成这些聚合结果查询就能快速返回不必每次重新计算。先创建一张目标表用来存放 1 分钟的 OHLCV 聚合结果。用AggregatingMergeTree合并聚合状态。排序键要包含所有分组维度这里是 symbol、tape 和 minute。开盘价和收盘价状态沿用与原始查询相同的(t, q)排序。CREATE TABLE trades_1min_ohlcv ( sym LowCardinality(String), z Enum8(NYSE 1, AMEX 2, Nasdaq 3), minute_bucket_ms UInt64, open_price_state AggregateFunction(argMin, Float64, Tuple(UInt64, UInt64)), high_price_state AggregateFunction(max, Float64), low_price_state AggregateFunction(min, Float64), close_price_state AggregateFunction(argMax, Float64, Tuple(UInt64, UInt64)), volume_state AggregateFunction(sum, UInt64), trade_count_state AggregateFunction(count) ) ENGINE AggregatingMergeTree ORDER BY (sym, z, minute_bucket_ms);下一步是创建物化视图。CREATE MATERIALIZED VIEW trades_1min_ohlcv_mv TO trades_1min_ohlcv AS SELECT sym, z, intDiv(t, 60000) * 60000 AS minute_bucket_ms, argMinState(p, tuple(t, q)) AS open_price_state, maxState(p) AS high_price_state, minState(p) AS low_price_state, argMaxState(p, tuple(t, q)) AS close_price_state, sumState(s) AS volume_state, countState() AS trade_count_state FROM trades GROUP BY sym, z, minute_bucket_ms;目标表和视图都要在开始写入前建好。视图只处理新插入的数据块不会自动回填已有数据行。如果需要可以用一条安排好时机的INSERT ... SELECT回填已有数据注意别把重叠的实时数据重复计入。查询里用了-Merge系列函数即使后台合并尚未完成也能合并状态。要查看数据执行下面这个查询。-- Query the table SELECT sym, z, minute_bucket_ms, fromUnixTimestamp64Milli(toInt64(minute_bucket_ms)) as minute_timestamp, argMinMerge(open_price_state) AS open_price, maxMerge(high_price_state) AS high_price, minMerge(low_price_state) AS low_price, argMaxMerge(close_price_state) AS close_price, sumMerge(volume_state) AS volume, countMerge(trade_count_state) AS trade_count FROM trades_1min_ohlcv GROUP BY sym, z, minute_bucket_ms ORDER BY sym, z, minute_bucket_ms;运行完整示例示例源码 现在位于blog-examples/stock-data-demo。需要 Node.js 22 或更高版本以及一个已有的本地或 Cloud ClickHouse 数据库。git clone https://github.com/ClickHouse/examples.git cd examples/blog-examples/stock-data-demo npm ci cp env.example .env # Set CLICKHOUSE_URL, CLICKHOUSE_USERNAME, CLICKHOUSE_PASSWORD, # CLICKHOUSE_DATABASE, and MASSIVE_API_KEY in .env. npm run setup npm run build npm start打开http://localhost:34567/stocks/查看图表或打开http://localhost:34567/admin控制数据写入。默认订阅 AAPL、MSFT 和 NVDA。想换其他代码修改MASSIVE_SYMBOLS即可仪表盘上的自选列表不会改变上游订阅。服务只绑定回环地址API 也没有鉴权公开暴露之前请先加上访问控制。如果你的密钥缺少所需的数据通道管理页面会提示认证失败。周末和节假日市场也可能冷清或休市。想在没有行情权限的情况下试用仪表盘可以另用一个 ClickHouse 数据库把 API key 留空并在npm start之前先运行npm run seed。这会插入明确标注为合成的价格数据时间戳落在最近三分钟内不会获取真实行情。当这些时间戳滑出图表窗口后重新运行 seed 即可。README 涵盖了本地 ClickHouse 的搭建、开发模式、配置和验证方法。结论这篇文章里我们用 Massive 提供市场数据、用 ClickHouse 做高速写入和查询搭建了一个实时逐笔数据tick data应用。内容涉及逐笔数据的流式传输与结构设计、写入性能的管理以及高效查询和可视化的实现。在这个 GitHub 仓库 里有一个可运行的示例可视化层用 React 实现。你可以借助它了解写入和查询的常见模式再针对生产应用补上可靠投递、回放、身份认证以及按具体负载做的调优。关于我们ClickHouseclickhouse.com) 是面向 AI 时代打造的高性能实时分析数据库能够以极致性能处理海量数据分析任务。凭借高并发、低延迟和云原生架构ClickHouse 广泛应用于可观测性、数据仓库、实时分析及 AI 数据基础设施等场景。我们致力于帮助企业在公有云平台上构建安全、弹性且高性价比的实时分析与 AI 数据平台加速释放数据价值推动智能化创新与数字化转型。目前Trip.com、DiDi、Meta、Sony、Netflix、Deutsche Bank、Sierra、Cloudflare 等全球领先企业均在使用 ClickHouse 支撑其关键业务和数据分析平台。