ClickHouse动态生成OHLC：如何排除查询结果中的最后一分钟

问题场景 #

在生成分钟级OHLC（开盘/收盘/最高/最低）数据时，需要排除最后一分钟（未结束的分钟），避免通过物化视图动态生成时出现同一分钟多条未聚合的记录。原有实现中，插入新tick数据后，同分钟会产生多条重复的OHLC记录，无法自动合并。

原参考的分钟级OHLC生成逻辑：

SELECT 
  id,
  minute,
  max(value) AS high,
  min(value) AS low,
  avg(value) AS avg,
  argMin(value, timestamp) AS first,
  argMax(value, timestamp) AS last
FROM security
GROUP BY id, toStartOfMinute(timestamp) AS minute
ORDER BY minute

完整示例场景代码：

create table ticks_data
(
    symbol       String,
    datetime_msc DateTime64(3),
    price        Float64,
    volume       UInt64
)
engine = MergeTree PARTITION BY toYYYYMM(datetime_msc)
ORDER BY (symbol, datetime_msc);

INSERT INTO ticks_data (symbol, datetime_msc, price, volume)
VALUES ('EURUSD', '2022-06-03 18:01:51.265',1.07084,0);
INSERT INTO ticks_data (symbol, datetime_msc, price, volume)
VALUES ('EURUSD', '2022-06-03 18:01:51.027',1.071429,0);
INSERT INTO ticks_data (symbol, datetime_msc, price, volume)
VALUES ('EURUSD', '2022-06-03 18:01:51.948',1.07089,0);

CREATE TABLE charts
(
    symbol       String,
    datetime     DATETIME,
    high         Float64,
    low          Float64,
    vol          UInt64,
    open         Float64,
    close        Float64
)
ENGINE = AggregatingMergeTree
order by (symbol,datetime);

insert into table charts
SELECT
  symbol,
  datetime,
  max(price) AS high,
  min(price) AS low,
  sum(volume) AS vol,
  arrayElement(arraySort((x,y)->y,groupArray(price), groupArray(datetime_msc)), 1) AS open,
  arrayElement(arraySort((x,y)->y, groupArray(price), groupArray(datetime_msc)), -1) AS close
FROM ticks_data
GROUP BY symbol, toStartOfMinute(datetime_msc) AS datetime
ORDER BY datetime;

CREATE MATERIALIZED VIEW charts_MV to charts
AS
SELECT
  symbol,
  datetime,
  max(price) AS high,
  min(price) AS low,
  sum(volume) AS vol,
  arrayElement(arraySort((x,y)->y,groupArray(price), groupArray(datetime_msc)), 1) AS open,
  arrayElement(arraySort((x,y)->y, groupArray(price), groupArray(datetime_msc)), -1) AS close
FROM ticks_data
GROUP BY symbol, toStartOfMinute(datetime_msc) AS datetime
ORDER BY datetime;

INSERT INTO ticks_data (symbol, datetime_msc, price, volume)
VALUES ('EURUSD', '2022-06-03 18:01:54.265',1.07098,0)

解决方案 #

1. 修正表结构与物化视图（解决重复聚合问题） #

原有实现直接插入聚合结果到AggregatingMergeTree，导致同分钟多次插入生成多条记录。正确做法是存储聚合状态而非直接结果，让ClickHouse自动合并状态：

-- 重新定义charts表，使用聚合函数State类型
DROP TABLE IF EXISTS charts;
CREATE TABLE charts
(
    symbol       String,
    datetime     DATETIME,
    high         AggregateFunction(max, Float64),
    low          AggregateFunction(min, Float64),
    vol          AggregateFunction(sum, UInt64),
    open         AggregateFunction(argMin, Float64, DateTime64(3)),
    close        AggregateFunction(argMax, Float64, DateTime64(3))
)
ENGINE = AggregatingMergeTree
ORDER BY (symbol, datetime);

-- 创建物化视图，插入聚合状态而非直接结果，同时过滤未结束的分钟
DROP MATERIALIZED VIEW IF EXISTS charts_MV;
CREATE MATERIALIZED VIEW charts_MV TO charts
AS
SELECT
  symbol,
  toStartOfMinute(datetime_msc) AS datetime,
  maxState(price) AS high,
  minState(price) AS low,
  sumState(volume) AS vol,
  argMinState(price, datetime_msc) AS open,
  argMaxState(price, datetime_msc) AS close
FROM ticks_data
-- 过滤当前未结束的分钟，只处理已完整的分钟
WHERE toStartOfMinute(datetime_msc) < toStartOfMinute(now())
GROUP BY symbol, datetime;

2. 查询最终OHLC数据 #

通过Merge函数将聚合状态计算为最终的OHLC值，如需动态排除最后一分钟，可添加过滤条件：

SELECT
  symbol,
  datetime,
  maxMerge(high) AS high,
  minMerge(low) AS low,
  sumMerge(vol) AS vol,
  argMinMerge(open) AS open,
  argMaxMerge(close) AS close
FROM charts
-- 可选：排除最后一分钟的记录
-- WHERE datetime != (SELECT max(datetime) FROM charts)
GROUP BY symbol, datetime
ORDER BY datetime;

3. 关键优化点 #

• 用argMinState/argMaxState替代数组排序获取开盘/收盘价，大幅提升性能，避免大数据量下的内存占用问题。
• 物化视图中直接过滤未结束的分钟，确保只处理已完成的分钟数据，无需后续手动排除。
• AggregatingMergeTree会自动合并相同(symbol, datetime)的聚合状态，最终每条分钟数据仅保留一条聚合记录，避免重复。

内容的提问来源于stack exchange，提问作者Pavel