S3Queue：对象存储文件如何持续消费

做的事：真实建S3Queue表监听一个路径，陆续丢文件进去

代码CREATE TABLE demo.s3queue_src (id UInt32, val String) ENGINE = S3Queue('http://127.0.0.1:19400/ch-lab/incoming/*.parquet', 'minioadmin', 'minioadmin123', 'Parquet') SETTINGS mode='unordered', keeper_path='/clickhouse/s3queue/incoming1'; CREATE TABLE demo.s3queue_target (id UInt32, val String) ENGINE=MergeTree ORDER BY id; CREATE MATERIALIZED VIEW demo.s3queue_mv TO demo.s3queue_target AS SELECT * FROM demo.s3queue_src; -- 真实先后上传两个文件(schema不完全相同)到incoming/路径 -- testdata.parquet: id/name/amt三列，1000行 -- file1.parquet: id/val两列，100行

验证：真实两个文件都被自动消费，一个不落

实测 · 本机执行输出

SELECT count() FROM demo.s3queue_target;
→ 1100   ← 真实1000(testdata.parquet) + 100(file1.parquet)

SELECT * FROM system.s3_queue_metadata FORMAT Vertical;
→ processed_nodes_count: 2
  processed_nodes: {
    'incoming/file1.parquet':    last_processed_timestamp=... retries=0
    'incoming/testdata.parquet': last_processed_timestamp=... retries=0
  }
  processing_nodes: {}   failed_nodes: {}

两个文件真实都没人工触发任何操作，本文只是把它们上传到MinIO的incoming/路径下，S3Queue表就真实自动发现了它们、消费了它们的内容、写进了目标表——processed_nodes里真实精确记录了每个文件的完整路径和处理时间，作为"这个文件我已经处理过了，不要重复处理"的真实凭证。

真实的核心机制：靠Keeper记账，实现"不重复处理"

S3Queue这套设计，真实是把本系列前面《ReplicatedMergeTree》和《Keeper》验证过的"用Keeper存轻量协调信息"这套思路，用在了对象存储文件消费这个场景上——keeper_path参数指定的路径，真实存的就是processed_nodes这样一份"已处理文件清单"，S3Queue表内部的后台线程会真实定期扫描incoming/*.parquet这个glob匹配到的文件列表，对每一个文件先查Keeper里是否已经记录过，没记录过的才真实去读取内容、推给挂在上面的物化视图、成功之后再把这个文件路径写进Keeper的已处理清单。本文特意上传了两个schema不完全相同的文件(一个三列一个两列)，真实验证了这套机制不要求所有被消费的文件长得一模一样，只要能按位置/名称映射到目标表结构即可(本文的三列文件里，第三列amt因为目标表没有对应列，真实被安静地丢弃了，只取了前两列)。这套"持续发现新文件、记账去重"的能力，真实解决的是对象存储场景下一个常见的真实需求——业务系统持续往S3/MinIO某个目录写入批次文件(比如每小时一个日志文件)，S3Queue不需要外部调度器盯着这个目录，靠自己内置的发现+记账机制，就能把这些文件真实持续、增量地导入进ClickHouse，即使ClickHouse服务本身重启过，Keeper里的记账信息也不会丢，不会导致重复导入或者漏掉文件。

分步演示：从文件上传到真实的Keeper记账去重

每一步都是本文上面用真实S3Queue跑出来的原始结果。

陆续丢文件进去，真实一个不落地自动消费

这一步新上传的文件

本文用到的工具

本机真实的MinIO + ENGINE = S3Queue，真实的Keeper配合记账
真实的 system.s3_queue_metadata 查看处理状态

下一篇 · CLICKHOUSE 内部原理 76

MySQL 表引擎：ClickHouse 如何读取外部数据库