应使用ReplacingMergeTree引擎并按sensorid和timestamp排序建表，写入时注入version字段（如Kafka offset），再通过OPTIMIZE TABLE FINAL手动合并分区实现去重。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
你需要在火山引擎中对从Kafka或API批量导入的JSON结构化数据做去重处理，避免sensorid+timestamp组合重复导致温度、湿度等指标被错误覆盖或统计失真。
确认数据是否已具备天然去重能力
检查你的JSON数据中是否包含明确的业务主键字段组合。例如：sensorid与timestamp共同构成唯一标识——这是最理想的去重前提。若缺失其中任一字段，【后续所有去重逻辑都将失效】，必须先回溯上游补全。
用SELECT语句快速验证：SELECT sensorid, timestamp, count(*) FROM heat_sensor_data_local GROUP BY sensorid, timestamp HAVING count(*) > 1；若返回结果非空，说明原始数据已存在重复。
建表阶段启用ReplacingMergeTree引擎
在创建本地表heat_sensor_data_local时，必须选用ReplacingMergeTree而非普通MergeTree：
CREATE TABLE heat_sensor_data_local (sensorid String, timestamp DateTime, temperature Float64, humidity Float64, alertflags Array(String), gpslatitude Float64, gpslongitude Float64) ENGINE = ReplacingMergeTree() ORDER BY (sensorid, timestamp) PARTITION BY toYYYYMM(timestamp);
注意：ORDER BY必须包含全部去重字段（sensorid, timestamp），且顺序不能颠倒；PARTITION BY仅影响存储分布，不参与去重判定。
ReplacingMergeTree不会实时删除旧记录，而是在后台合并分区时自动保留version字段最大或时间戳最新的那一行。因此，你必须在INSERT前确保每条JSON记录附带一个可比较的新鲜度标识。
写入时注入version字段控制保留优先级
方法一：用Kafka消息的offset作为version
在Kafka Producer端发送JSON时，显式添加"version": {kafka_offset}字段。ByteHouse消费时通过JSONExtractUInt(payload, 'version')提取该值，并在建表语句中声明VERSION version字段：
ENGINE = ReplacingMergeTree(version) ORDER BY (sensorid, timestamp)
方法二：用timestamp自身作version（仅当毫秒级精度可用）
若timestamp字段精确到毫秒且上游严格保序，可直接设为VERSION timestamp。但【一旦出现时钟回拨或乱序写入，将导致旧数据覆盖新数据】，生产环境慎用。
触发后台去重合并
第一步：等待自动合并
ByteHouse默认每8分钟触发一次后台合并任务，自动清理同一分区内的重复行。无需人工干预，但延迟不可控。
第二步：手动强制合并（立即生效）
执行OPTIMIZE TABLE heat_sensor_data_local PARTITION '202610' FINAL；将指定分区（如2026年10月）内所有数据块彻底重写，立即完成去重。注意：该操作会短暂占用磁盘IO和CPU，建议避开业务高峰。
第三步：验证去重结果
运行SELECT count(*) FROM (SELECT sensorid, timestamp FROM heat_sensor_data_local GROUP BY sensorid, timestamp)；若结果等于原始总行数，则无重复；若小于，则说明去重已生效。
