Langfuse 实时聚合实战ClickHouse 增量物化视图Incremental MV最佳实践【免费下载链接】langfuse Open source AI engineering platform: LLM evals, observability, metrics, prompt management, playground, datasets. Integrates with OpenTelemetry, LangChain, OpenAI SDK, LiteLLM, and more. YC W23项目地址: https://gitcode.com/GitHub_Trending/la/langfuse本篇指南围绕 Langfuse 仓库内置的 ClickHouse 最佳实践规则query-mv-incremental位于 .agents/skills/clickhouse-best-practices/rules/query-mv-incremental.md展开讲解在 Langfuse 这类高吞吐可观测性平台中如何用增量物化视图把每次查询全表聚合的反模式改造成插入时预聚合、查询时只读千行级结果的正解。读完本文你将掌握AggregatingMergeTree Materialized View State/Merge 函数三件套的完整落地姿势并能在 Langfuse 真实迁移脚本packages/shared/clickhouse/migrations/canonical/中找到对应生产级范例作为参照。背景为什么 Langfuse 需要增量聚合Langfuse 将 traces、observations、scores、events 等核心遥测数据存储在 ClickHouse 中其中事件表events是典型的超大明细表累积数据可达数十亿行。而 UI 侧的仪表盘、trace 列表页需要按小时、按事件类型、按项目做实时统计若沿用传统数据库思维每次页面加载都现算一条GROUP BY就要扫描最近 7 天的全部明细代价是从数十亿行中读数据延迟与集群压力都不可接受。规则文件给出的核心主张非常明确Impact: HIGH— Incremental MVs 在插入时自动把视图的查询应用到新数据块结果写入目标表部分结果随时间合并。读数千行而不是数十亿行对集群开销极小。这条规则属于该技能包 28 条规则中的query-mv-*物化视图类别评级为 HIGH。物化视图的选型总览可见 .agents/skills/clickhouse-best-practices/SKILL.md增量 MVquery-mv-incremental用于实时聚合Refreshable MVquery-mv-refreshable用于复杂 JOIN 与批处理工作流。反模式每次查询全量聚合规则文件首先给出了需要避免的写法。以事件流分析为例若在每次仪表盘加载时直接对明细表做聚合-- Full aggregation on every dashboard load SELECT event_type, toStartOfHour(timestamp) as hour, count() as events, uniq(user_id) as unique_users FROM events WHERE timestamp now() - INTERVAL 7 DAY GROUP BY event_type, hour; -- Scans 7 days of data every time (billions of rows)这条 SQL 的问题在于聚合结果没有被保存每次请求都要重新扫描 7 天明细数十亿行。在 Langfuse 的查询路径中这类明细表如 events列式存储虽然能压缩数据但高频、重复的全表聚合依然会持续消耗 CPU 与 IO拖慢仪表盘首屏。值得强调的是Langfuse 技能包中有一条相关的 Langfuse 专属约束events表被设计为无需FINAL查询events时严禁使用FINAL关键字因为该关键字会显著拖慢性能。因此对事件明细的查询优化思路应转向预聚合落表而非靠 FINAL 现场合并。正解增量 MV 三段式预聚合方案规则文件给出了标准的增量物化视图三段式这也是 ClickHouse 社区广泛采用的最佳实践。请完整复制以下结构-- 1. 创建聚合结果目标表 CREATE TABLE events_hourly ( event_type LowCardinality(String), hour DateTime, events AggregateFunction(count), unique_users AggregateFunction(uniq, UInt64) ) ENGINE AggregatingMergeTree() ORDER BY (event_type, hour); -- 2. 创建物化视图增量填充目标表 CREATE MATERIALIZED VIEW events_hourly_mv TO events_hourly AS SELECT event_type, toStartOfHour(timestamp) as hour, countState() as events, uniqState(user_id) as unique_users FROM events GROUP BY event_type, hour; -- 3. 查询预聚合数据 SELECT event_type, hour, countMerge(events) as events, uniqMerge(unique_users) as unique_users FROM events_hourly WHERE hour now() - INTERVAL 7 DAY GROUP BY event_type, hour; -- Reads thousands of rows instead of billions三段式逐一拆解目标表events_hourly使用AggregatingMergeTree引擎聚合列的类型是AggregateFunction(count)、AggregateFunction(uniq, UInt64)。后台合并时相同ORDER BY (event_type, hour)键的行会把聚合中间态合并起来从而把一天内同小时、同类型的多条聚合中间态收敛成一行。event_type使用LowCardinality(String)符合技能包中schema-types-lowcardinality低基数字符串用 LowCardinality的配套建议。物化视图events_hourly_mvTO子句把结果定向写入目标表countState()、uniqState()这类带-State后缀的函数输出的是聚合中间态而非最终数值供后续合并。查询语句用countMerge(events)、uniqMerge(unique_users)把中间态展开为最终数值。由于数据已被压缩到小时粒度同样的 7 天窗口只读数千行。规则文件总结的要点如下必须严格遵循MV 内用-State函数查询内用-Merge函数——这是中间态与终态的正确配对方式增量语义——MV 只处理创建之后新插入的数据块已有历史数据不会自动纳入需要单独回填backfill插入时开销极小——预聚合只发生在数据写入路径上对集群的额外负担可以忽略。State/Merge 机制与 SimpleAggregateFunction 的差异深入源码可以发现Langfuse 实际生产表并未一律使用AggregateFunction-State而是大量使用SimpleAggregateFunction。以 packages/shared/clickhouse/migrations/canonical/0023_traces_aggregating_merge_trees.up.sql 中的traces_all_amt为例CREATE TABLE traces_all_amt {CLICKHOUSE_CLUSTER_CLAUSE} ( project_id String, id String, timestamp SimpleAggregateFunction(min, DateTime64(3)), end_time SimpleAggregateFunction(max, Nullable(DateTime64(3))), name SimpleAggregateFunction(anyLast, Nullable(String)), metadata SimpleAggregateFunction(maxMap, Map(String, String)), tags SimpleAggregateFunction(groupUniqArrayArray, Array(String)), cost_details SimpleAggregateFunction(sumMap, Map(String, Decimal(38, 12))), bookmarked AggregateFunction(argMax, Nullable(Bool), DateTime64(3)), input AggregateFunction(argMax, String, DateTime64(3)) CODEC (ZSTD(3)) ) Engine AggregatingMergeTree() ORDER BY (project_id, id);对比规则文件示例可以提炼出两条配套规律SimpleAggregateFunction(f, T)适用于min、max、anyLast、sumMap、groupUniqArrayArray这类合并结果与数据无关可交换、幂等性要求低的函数。它在 MV 里可以直接写min(...)、anyLast(...)等普通函数见同文件 MV 中min(tn.start_time) as start_time、max(coalesce(tn.end_time, tn.start_time)) as end_time查询端也不需要-Merge直接SELECT列即可ClickHouse 在合并时自动按类型语义收敛。AggregateFunction(f, T[, X])适用于argMax、uniq、count等需要保留中间态的复杂聚合。这类列在 MV 中必须显式使用-State变体同文件中argMaxState(tn.input, ...) as input查询端则需argMaxMerge(...)展开。traces_all_amt中还有一个值得注意的细节bookmarked、public、input、output这类最后一次写入胜出的字段用argMaxState(col, event_ts)以事件时间戳为版本确保合并时取的是时间上最新的值而非任意值——这正是-State中间态函数表达力所在。而规则文件示例中uniqState(user_id)同理保证uniq的去重统计在多次合并后依然精确。增量语义已有数据不会自动回填规则文件明确强调Incremental - existing data not automatically included (backfill separately)。增量 MV 从创建时刻起只消费之后插入到源表的数据块创建前已存在的数十亿历史行不会自动出现在目标表中。这意味着在 Langfuse 这类有存量数据的系统里落地新 MV 时必须配套回填流程。仓库中恰好有对应的回填实现可参考worker/src/backgroundMigrations/backfillEventsFullFromObservations.ts从 observations 表回填 events 明细worker/src/backgroundMigrations/backfillEventsFullFromDatasetRunItems.ts从 dataset run items 回填。这些后台迁移background migrations由 worker/src/backgroundMigrations/backgroundMigrationManager.ts 统一调度与 MV 的增量填充形成历史全量 实时增量的完整数据闭环。实操时建议先建 MV 保证新数据开始累积再跑回填任务补历史最后在查询侧把两部分结果按聚合键合并。Langfuse 仓库中的生产级应用实例实例一project_environments项目环境集合packages/shared/clickhouse/migrations/canonical/0009_add_project_environments.up.sql 是规则文件方案最简洁的仓库内印证——用聚合表保存每个项目出现过哪些环境CREATE TABLE project_environments {CLICKHOUSE_CLUSTER_CLAUSE} ( project_id String, environments SimpleAggregateFunction(groupUniqArrayArray, Array(String)) ) ENGINE {CLICKHOUSE_REPLICATION_PREFIX}AggregatingMergeTree ORDER BY (project_id); CREATE MATERIALIZED VIEW project_environments_traces_mv {CLICKHOUSE_CLUSTER_CLAUSE} TO project_environments AS SELECT project_id, groupUniqArray(environment) AS environments FROM traces GROUP BY project_id;注意这里同时存在三条 MVproject_environments_traces_mv、project_environments_observations_mv、project_environments_scores_mv分别从 traces、observations、scores 三个源表聚合写入同一个目标表。这展示了增量 MV 的一个进阶用法多源合并。三个数据流各自增量写入AggregatingMergeTree后台合并时按project_id收敛最终一行包含全部来源的环境集合。查询端直接SELECT environments FROM project_environments WHERE project_id ...即可无需任何-Merge。实例二traces 的 AMT 三级分层TTL 与版本化字段同文件0023_traces_aggregating_merge_trees.up.sql展示了更完整的工业级形态。Langfuse 为 traces 建了traces_nullEngine Null()的触发器表、traces_all_amt全量、traces_7d_amtTTL 7 天、traces_30d_amtTTL 30 天三张聚合表每张配一个 MV-- Null 引擎触发器表只触发 MV不落明细节省存储 CREATE TABLE traces_null {CLICKHOUSE_CLUSTER_CLAUSE} ( ... ) Engine Null(); -- 7 天 TTL 的 AMT 目标表 CREATE TABLE traces_7d_amt {CLICKHOUSE_CLUSTER_CLAUSE} ( ... ) Engine AggregatingMergeTree() ORDER BY (project_id, id) TTL toDate(start_time) INTERVAL 7 DAY; CREATE MATERIALIZED VIEW IF NOT EXISTS traces_7d_amt_mv {CLICKHOUSE_CLUSTER_CLAUSE} TO traces_7d_amt AS SELECT tn.project_id as project_id, tn.id as id, min(tn.start_time) as start_time, anyLast(tn.name) as name, argMaxState(tn.input, if(tn.input , tn.event_ts, toDateTime64(0, 3))) as input, ... FROM traces_null tn GROUP BY project_id, id;这里可以提炼出三个增量 MV 的设计模式Null 引擎触发器表真实数据写入traces_nullMV 从 Null 表消费并转发聚合结果明细本身不落地避免重复存储TTL 分层同一份数据按保留窗口建多张聚合表7 天 / 30 天 / 全量配合TTL toDate(start_time) INTERVAL 7 DAY自动淘汰旧分区让热查询只访问小表版本化聚合argMaxState(col, event_ts)保证合并后保留最新值避免乱序写入导致旧值覆盖新值。实例三events_core行级投影 MV并非所有 MV 都是聚合型。0041_create_events_core_mv.up.sql中的events_core_mv是行级投影MV——把events_full的字段裁剪、leftUTF8(input, 200)截断大字段后写入更轻量的events_coreCREATE MATERIALIZED VIEW IF NOT EXISTS events_core_mv {CLICKHOUSE_CLUSTER_CLAUSE} TO events_core AS SELECT project_id, trace_id, span_id, ..., leftUTF8(input, 200) as input, leftUTF8(output, 200) as output, ... FROM events_full;这印证了TO目标表 MV 的通用性无论目标是聚合表还是投影表插入时增量消费、写入独立目标表的机制一致。而 Langfuse 对events的查询统一走 packages/shared/src/server/queries/clickhouse-sql/event-query-builder.ts 查询构建器技能包中的 Langfuse 专属规则明确要求对events表的查询必须先尝试该构建器不要手写 SQL预聚合/投影落表后上层查询读的是聚合结果或精简列天然贴合读数千行而非数十亿行的目标。MV 运维修改与演进drop-recreate 禁区增量 MV 上线后仍会面临查询逻辑调整。技能包 SKILL.md 针对 Langfuse 给出了两条硬性约束直接适用于规则文件场景严禁对正在接收实时写入的源表执行先 DROP 再 CREATE的 MV 重建——DROP与CREATE之间的每一行新插入数据都会静默、永久地丢失不再进入目标表。正确做法是ALTER TABLE mv {CLICKHOUSE_CLUSTER_CLAUSE} MODIFY QUERY select在不中断写入的前提下替换转换逻辑。若新查询增加了列需先对目标表执行ALTER TABLE ... ADD COLUMN IF NOT EXISTS ...且这些目标表 ALTER 必须携带{CLICKHOUSE_CLUSTERED_ONLY: SETTINGS alter_sync 2}模板片段确保任何副本不会在新列就绪前应用新 MV 查询。MODIFY QUERY仅对TO目标表型 MV 可行——而 Langfuse 全部 MV 均使用TO。不要在迁移中使用CREATE OR REPLACE VIEW/CREATE OR REPLACE TABLE/EXCHANGE TABLES——其原子替换依赖renameat2在 NFS 挂载如 AWS EFS的自托管部署上会失败导致启动中止。普通视图请在同一迁移文件内拆成DROP VIEW IF EXISTS ...CREATE VIEW ...两条语句。另外Langfuse 的迁移模板使用{CLICKHOUSE_CLUSTER_CLAUSE}占位符在集群/单机两种部署模式下渲染 DDL复制上方任何 SQL 到自己的集群时需按自身拓扑决定是否追加ON CLUSTER子句。选型增量 MV 还是 Refreshable MV技能包把物化视图规则拆成两条query-mv-incremental与 query-mv-refreshable.md二者的适用边界可以这样区分维度增量 MV本规则Refreshable MV触发方式源表插入数据块时实时触发按计划周期性执行目标引擎常配AggregatingMergeTree部分结果合并MergeTree等全量重写或追加延迟秒级以内近乎实时取决于刷新周期如每 5 分钟典型场景实时聚合计数、去重统计、实时看板复杂多表 JOIN 反范式化、Top N 缓存、批处理 DAG数据新鲜度始终最新允许轻微滞后换取亚毫秒查询规则文件示例中的events_hourly小时级实时聚合属于前者而订单 JOIN 客户 JOIN 商品这类复杂关联查询见query-mv-refreshable的REFRESH EVERY 5 MINUTE示例属于后者。Langfuse 中 trace 的 AMT 聚合链路是纯增量 MV 的典型代表因为遥测写入是持续高频流实时聚合价值最高。选型时遵循技能包原则实时聚合选增量 MV复杂 JOIN 与批处理选 Refreshable MV同时谨记 Refreshable MV 的警告——查询执行耗时应远小于刷新间隔不要每 10 秒刷新一个要跑 10 秒以上的查询。总结规则query-mv-incremental的结论可以浓缩为一句话把聚合从查询时前移到插入时让AggregatingMergeTree在后台完成部分结果的合并。落地清单如下明细表之上建AggregatingMergeTree目标表聚合列用AggregateFunction(...)复杂聚合或SimpleAggregateFunction(...)可交换简单聚合建TO目标表的增量 MVMV 内对AggregateFunction列使用-State变体函数查询时用对应-Merge变体展开牢记增量语义历史数据需独立回填参考仓库的 backgroundMigrations 实现上线后修改 MV 用ALTER TABLE ... MODIFY QUERY永远不要 DROP 正在接收写入的 MV与 Refreshable MV 按实时 vs 批处理分流各司其职。Langfuse 在packages/shared/clickhouse/migrations/canonical/0023_traces_aggregating_merge_trees.up.sql、0009_add_project_environments.up.sql、0041_create_events_core_mv.up.sql中的真实迁移脚本就是这条规则最完整的生产级注脚——你可以直接打开这些文件对照本文逐段验证State/Merge配对、SimpleAggregateFunction简化、TTL 分层与多源合并的每一种写法。【免费下载链接】langfuse Open source AI engineering platform: LLM evals, observability, metrics, prompt management, playground, datasets. Integrates with OpenTelemetry, LangChain, OpenAI SDK, LiteLLM, and more. YC W23项目地址: https://gitcode.com/GitHub_Trending/la/langfuse创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考