无论你是刚开始接触 ClickHouse，还是负责现有部署，都会不可避免地需要用历史数据回填表。在某些情况下，这项工作相对简单；但如果还需要填充 materialized view，复杂度就会增加。本指南整理了完成这项任务的一些流程，你可以根据自己的使用场景加以应用。
在本指南中，我们使用一个 PyPI 数据集。该数据集中的每一行都表示一次通过 pip 等工具下载 Python 软件包的记录。
例如，这个子集仅涵盖一天的数据——2024-12-17，并可在 https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/ 公开获取。你可以使用以下方式查询：
SELECT count()
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/*.parquet')
┌────count()─┐
│ 2039988137 │ -- 20.4 亿
└────────────┘
1 行，耗时：32.726 秒。处理了 20.4 亿行，170.05 KB（6234 万行/秒，5.20 KB/秒）
峰值内存占用：239.50 MiB。
DESCRIBE TABLE s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/*.parquet')
FORMAT PrettyCompactNoEscapesMonoBlock
SETTINGS describe_compact_output = 1
┌─name───────────────┬─type────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ timestamp │ Nullable(DateTime64(6)) │
│ country_code │ Nullable(String) │
│ url │ Nullable(String) │
│ project │ Nullable(String) │
│ file │ Tuple(filename Nullable(String), project Nullable(String), version Nullable(String), type Nullable(String)) │
│ installer │ Tuple(name Nullable(String), version Nullable(String)) │
│ python │ Nullable(String) │
│ implementation │ Tuple(name Nullable(String), version Nullable(String)) │
│ distro │ Tuple(name Nullable(String), version Nullable(String), id Nullable(String), libc Tuple(lib Nullable(String), version Nullable(String))) │
│ system │ Tuple(name Nullable(String), release Nullable(String)) │
│ cpu │ Nullable(String) │
│ openssl_version │ Nullable(String) │
│ setuptools_version │ Nullable(String) │
│ rustc_version │ Nullable(String) │
│ tls_protocol │ Nullable(String) │
│ tls_cipher │ Nullable(String) │
└────────────────────┴─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
完整的 PyPI 数据集包含超过 1 万亿行，可在我们的公开演示环境 clickpy.clickhouse.com 中查看。有关该数据集的更多信息，包括此演示如何利用 materialized view 提升性能，以及数据如何按日填充，请参见此处。
- 在现有数据摄取的情况下回填数据 - 新数据正在持续加载，同时需要回填历史数据。这些历史数据已经明确。
- 向现有表添加 materialized view - 需要在一套历史数据已完成填充且数据已在持续stream的系统中添加新的 materialized view。
我们假设数据将从对象存储中进行回填。在所有情况下，我们的目标都是避免暂停数据插入。
我们建议从对象存储回填历史数据。在可能的情况下，应将数据导出为 Parquet，以获得最佳读取性能和压缩效果 (减少网络传输) 。通常建议文件大小控制在 150MB 左右，但 ClickHouse 支持超过 70 种文件格式，并且能够处理各种大小的文件。
在所有场景中，我们都依赖”副本表与视图”的概念。这些表和视图是实时流式数据所用对象的副本，可在隔离环境中执行回填操作，并在发生故障时提供便捷的恢复手段。例如，以下是主 pypi 表和 materialized view，用于计算每个 Python 项目的下载次数：
CREATE TABLE pypi
(
`timestamp` DateTime,
`country_code` LowCardinality(String),
`project` String,
`type` LowCardinality(String),
`installer` LowCardinality(String),
`python_minor` LowCardinality(String),
`system` LowCardinality(String),
`on` String
)
ENGINE = MergeTree
ORDER BY (project, timestamp)
CREATE TABLE pypi_downloads
(
`project` String,
`count` Int64
)
ENGINE = SummingMergeTree
ORDER BY project
CREATE MATERIALIZED VIEW pypi_downloads_mv TO pypi_downloads
AS SELECT
project,
count() AS count
FROM pypi
GROUP BY project
INSERT INTO pypi SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/1734393600-000000000{000..100}.parquet')
0 rows in set. Elapsed: 15.702 sec. Processed 41.23 million rows, 3.94 GB (2.63 million rows/s., 251.01 MB/s.)
峰值内存占用: 977.49 MiB.
┌──count()─┐
│ 20612750 │ -- 2061万
└──────────┘
1 row in set. Elapsed: 0.004 sec.
SELECT sum(count)
FROM pypi_downloads
┌─sum(count)─┐
│ 20612750 │ -- 2061万
└────────────┘
1 row in set. Elapsed: 0.006 sec. Processed 96.15 thousand rows, 769.23 KB (16.53 million rows/s., 132.26 MB/s.)
Peak memory usage: 682.38 KiB.
{101..200}。虽然可以直接向 pypi 插入数据，但我们也可以通过创建副本表来隔离执行此次回填操作。
如果回填失败，主表不会受到任何影响，只需对副本表执行 truncate 操作并重新执行即可。
要创建这些视图的新副本，可以使用带有后缀 _v2 的 CREATE TABLE AS 子句：
CREATE TABLE pypi_v2 AS pypi
CREATE TABLE pypi_downloads_v2 AS pypi_downloads
CREATE MATERIALIZED VIEW pypi_downloads_mv_v2 TO pypi_downloads_v2
AS SELECT
project,
count() AS count
FROM pypi_v2
GROUP BY project
INSERT INTO pypi_v2 SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/1734393600-000000000{101..200}.parquet')
0 rows in set. Elapsed: 17.545 sec. Processed 40.80 million rows, 3.90 GB (2.33 million rows/s., 222.29 MB/s.)
峰值内存占用：991.50 MiB。
SELECT count()
FROM pypi_v2
┌──count()─┐
│ 20400020 │ -- 2040万
└──────────┘
1 row in set. Elapsed: 0.004 sec.
SELECT sum(count)
FROM pypi_downloads_v2
┌─sum(count)─┐
│ 20400020 │ -- 2040 万
└────────────┘
1 row in set. Elapsed: 0.006 sec. Processed 95.49 thousand rows, 763.90 KB (14.81 million rows/s., 118.45 MB/s.)
峰值内存占用: 688.77 KiB.
pypi_v2 和 pypi_downloads_v2，然后重新加载数据即可。
数据加载完成后，我们可以使用 ALTER TABLE MOVE PARTITION 子句，将数据从副本表移动到主表。
ALTER TABLE pypi_v2 MOVE PARTITION () TO pypi
0 rows in set. Elapsed: 1.401 sec.
ALTER TABLE pypi_downloads_v2 MOVE PARTITION () TO pypi_downloads
0 rows in set. Elapsed: 0.389 sec.
分区名称上面的 MOVE PARTITION 调用使用的分区名称是 ()。这表示该表只有一个分区 (即未分区) 。对于已分区的表，你需要多次调用 MOVE PARTITION——每个分区调用一次。当前分区的名称可以从 system.parts 表中查询，例如 SELECT DISTINCT partition FROM system.parts WHERE (table = 'pypi_v2')。 pypi 和 pypi_downloads 已包含完整数据。pypi_downloads_v2 和 pypi_v2 可以安全地删除。
┌──count()─┐
│ 41012770 │ -- 约 4101 万
└──────────┘
1 row in set. Elapsed: 0.003 sec.
SELECT sum(count)
FROM pypi_downloads
┌─sum(count)─┐
│ 41012770 │ -- 4101.2770 万
└────────────┘
1 row in set. Elapsed: 0.007 sec. Processed 191.64 thousand rows, 1.53 MB (27.34 million rows/s., 218.74 MB/s.)
SELECT count()
FROM pypi_v2
MOVE PARTITION 操作既轻量 (利用硬链接) ，又具备原子性；也就是说，它要么失败，要么成功，不会出现中间状态。
在下面的数据回填场景中，我们会大量利用这一过程。
请注意，这一过程要求用户自行确定每次插入操作的规模。
插入越大，也就是行数越多，所需的 MOVE PARTITION 操作就越少。不过，这需要与插入失败时 (例如因网络中断) 所带来的恢复成本进行权衡。你还可以通过将文件分批处理来降低风险。这既可以通过范围查询实现，例如 WHERE timestamp BETWEEN 2024-12-17 09:00:00 AND 2024-12-17 10:00:00，也可以通过 glob 模式实现。例如，
INSERT INTO pypi_v2 SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/1734393600-000000000{101..200}.parquet')
INSERT INTO pypi_v2 SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/1734393600-000000000{201..300}.parquet')
INSERT INTO pypi_v2 SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/1734393600-000000000{301..400}.parquet')
--持续执行直到所有文件加载完毕，或调用 MOVE PARTITION
ClickPipes 在从对象存储加载数据时会采用这种方法，自动为目标表及其 materialized view 创建副本，从而无需用户执行上述步骤。再配合多个工作线程，每个工作线程分别处理不同的子集 (通过 glob 模式划分) ，并使用各自的副本表，即可快速加载数据，同时保证精确一次语义。感兴趣的读者可在这篇博客中了解更多细节。
- 确定检查点——即需要恢复历史数据的时间戳或列值。
- 为主表和 materialized view 的目标表创建副本。
- 为所有指向步骤 (2) 中所创建目标表的 materialized view 创建副本。
- 向我们在步骤 (2) 中创建的主表副本插入数据。
- 将所有分区从副本表移动到原始表。删除副本表。
例如，在我们的 PyPI 数据中，假设已经有数据加载完成。我们可以识别出最小时间戳，从而确定“检查点”。
SELECT min(timestamp)
FROM pypi
┌──────min(timestamp)─┐
│ 2024-12-17 09:00:00 │
└─────────────────────┘
1 row in set. Elapsed: 0.163 sec. Processed 1.34 billion rows, 5.37 GB (8.24 billion rows/s., 32.96 GB/s.)
峰值内存占用: 227.84 MiB.
2024-12-17 09:00:00 之前的数据。按照前面介绍的流程，我们创建副本表和视图，并通过基于 timestamp 的过滤器加载这部分子集数据。
CREATE TABLE pypi_v2 AS pypi
CREATE TABLE pypi_downloads_v2 AS pypi_downloads
CREATE MATERIALIZED VIEW pypi_downloads_mv_v2 TO pypi_downloads_v2
AS SELECT project, count() AS count
FROM pypi_v2
GROUP BY project
INSERT INTO pypi_v2 SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/1734393600-*.parquet')
WHERE timestamp < '2024-12-17 09:00:00'
0 rows in set. Elapsed: 500.152 sec. Processed 2.74 billion rows, 364.40 GB (5.47 million rows/s., 728.59 MB/s.)
对 Parquet 中的 timestamp 列进行过滤通常非常高效。ClickHouse 只需读取 timestamp 列即可确定需要加载的完整数据范围，从而尽可能减少网络流量。ClickHouse 查询引擎还可以利用 Parquet 索引，例如最小值-最大值索引。
ALTER TABLE pypi_v2 MOVE PARTITION () TO pypi
ALTER TABLE pypi_downloads_v2 MOVE PARTITION () TO pypi_downloads
直接使用 ClickHouse Cloud 中的 ClickPipes如果你使用的是 ClickHouse Cloud，并且数据可以单独隔离在自己的存储桶中 (即不需要过滤器) ，那么应使用 ClickPipes 来恢复历史备份。除了通过多个工作线程并行加载来缩短加载时间外，ClickPipes 还会自动完成上述流程，并为主表和 materialized view 创建副本表。
场景 2：向现有表添加 materialized views
避免使用 POPULATE除了在已暂停摄取的小型数据集上为 materialized views 执行回填之外，我们不建议使用 POPULATE 命令。该操作可能会遗漏插入其源表的行，因为 materialized view 是在 POPULATE 完成后才创建的。此外，POPULATE 会对全部数据运行，在大型数据集上很容易受到中断或内存限制的影响。
- 创建 materialized view，并添加一个过滤条件，使其只处理大于最近未来某个任意时间点的行。
- 运行
INSERT INTO SELECT 查询，将数据插入 materialized view 的目标表，同时按该视图的聚合查询从源表读取数据。
这还可以进一步优化：在步骤 (2) 中仅针对部分数据执行，和/或为 materialized view 使用一个额外的目标表 (待插入完成后，再将分区附加到原始表) ，以便在发生故障后更轻松地恢复。
请看下面这个 materialized view，它用于计算每小时最热门的项目。
CREATE TABLE pypi_downloads_per_day
(
`hour` DateTime,
`project` String,
`count` Int64
)
ENGINE = SummingMergeTree
ORDER BY (project, hour)
CREATE MATERIALIZED VIEW pypi_downloads_per_day_mv TO pypi_downloads_per_day
AS SELECT
toStartOfHour(timestamp) as hour,
project,
count() AS count
FROM pypi
GROUP BY
hour,
project
SELECT 子句，加入一个过滤条件，只考虑时间晚于不久后某个任意时刻的行——这里我们假设 2024-12-17 09:00:00 是几分钟之后的时间。
CREATE MATERIALIZED VIEW pypi_downloads_per_day_mv TO pypi_downloads_per_day
AS SELECT
toStartOfHour(timestamp) AS hour,
project, count() AS count
FROM pypi WHERE timestamp >= '2024-12-17 09:00:00'
GROUP BY hour, project
INSERT INTO SELECT 将结果插入该视图的目标表。例如，对于上述视图：
INSERT INTO pypi_downloads_per_day SELECT
toStartOfHour(timestamp) AS hour,
project,
count() AS count
FROM pypi
WHERE timestamp < '2024-12-17 09:00:00'
GROUP BY
hour,
project
Ok.
0 rows in set. Elapsed: 2.830 sec. Processed 798.89 million rows, 17.40 GB (282.28 million rows/s., 6.15 GB/s.)
Peak memory usage: 543.71 MiB.
pypi_downloads_per_day_v2，向其中插入数据，然后将其生成的分区附加到 pypi_downloads_per_day。
materialized view 的查询通常会更复杂 (这并不少见，否则用户也不会使用视图！) ，而且会消耗资源。在极少数情况下，查询所需的资源甚至会超出服务器的承载能力。这也凸显了 ClickHouse materialized view 的一个优势——它们是增量式的，不会一次处理整个数据集！
在这种情况下，用户有以下几种选择：
- 修改查询以回填各个时间范围，例如
WHERE timestamp BETWEEN 2024-12-17 08:00:00 AND 2024-12-17 09:00:00、WHERE timestamp BETWEEN 2024-12-17 07:00:00 AND 2024-12-17 08:00:00 等。
- 使用 Null 表引擎 来填充 materialized view。这样可以模拟 materialized view 典型的增量填充过程，在数据块 (大小可配置) 上执行其查询。
(1) 是最简单的方法，而且通常已经足够。为简洁起见，这里不提供示例。
下面将进一步介绍 (2)。
使用 Null 表引擎填充 materialized view
/dev/null) 。虽然这看起来有些矛盾，但 materialized view 仍会基于插入到该表引擎中的数据执行。这使我们能够在不持久化原始数据的情况下构建 materialized view，从而避免 I/O 以及相关的存储开销。
需要注意的是，任何关联到该表引擎的 materialized views 在数据插入时仍会按数据块执行，并将结果发送到目标表。这些块的大小可以配置。更大的块虽然可能更高效 (处理速度也更快) ，但也会消耗更多资源 (主要是内存) 。使用这种表引擎意味着我们可以以增量方式构建 materialized view，即一次处理一个块，从而避免将整个聚合保存在内存中。
请看下面的示例：
CREATE TABLE pypi_v2
(
`timestamp` DateTime,
`project` String
)
ENGINE = Null
CREATE MATERIALIZED VIEW pypi_downloads_per_day_mv_v2 TO pypi_downloads_per_day
AS SELECT
toStartOfHour(timestamp) as hour,
project,
count() AS count
FROM pypi_v2
GROUP BY
hour,
project
pypi_v2,，用于接收将被用来构建 materialized view 的行。请注意，我们将 schema 限制为仅包含所需的列。我们的 materialized view 会对插入到该表中的行执行聚合 (一次处理一个块) ，并将结果发送到目标表 pypi_downloads_per_day。
这里我们使用 pypi_downloads_per_day 作为目标表。为了提高弹性，用户可以创建一个副本表 pypi_downloads_per_day_v2，并像前面的示例所示那样将其用作该视图的目标表。插入完成后，pypi_downloads_per_day_v2 中的分区随后可以移动到 pypi_downloads_per_day。这样一来，如果插入因内存问题或服务器中断而失败，我们就能够进行恢复；也就是说，只需 TRUNCATE pypi_downloads_per_day_v2、调整 settings，然后重试即可。
pypi 中用于回填的相关数据插入到 pypi_v2 中。
INSERT INTO pypi_v2 SELECT timestamp, project FROM pypi WHERE timestamp < '2024-12-17 09:00:00'
0 rows in set. Elapsed: 27.325 sec. Processed 1.50 billion rows, 33.48 GB (54.73 million rows/s., 1.23 GB/s.)
峰值内存占用: 639.47 MiB.
639.47 MiB。
性能与资源调优
上述场景中的性能和资源消耗受多种因素影响。在尝试调优之前，建议读者先了解 Optimizing for S3 Insert and Read Performance guide 中 Using Threads for Reads 部分详细介绍的插入机制。概括如下：
简单 INSERT SELECT 查询提示：对于不含复杂转换的简单 INSERT INTO t1 SELECT * FROM t2 查询，可考虑启用 optimize_trivial_insert_select=1。该设置 (自 24.7 版本起默认禁用) 会自动调整 SELECT 的并行度，使其与 max_insert_threads 一致，从而减少资源占用以及创建的 parts 数量。这对于在表之间进行大批量数据迁移尤其实用。
min_insert_block_size_bytes_for_materialized_views 和 min_insert_block_size_rows_for_materialized_views 来提升性能。如需修改这些参数，请遵循针对 min_insert_block_size_rows 和 min_insert_block_size_bytes 所讨论的最佳实践。
若要降低内存占用，可以尝试调整这些设置，但这不可避免地会影响性能。以下使用前面的查询展示相关示例。
将 max_insert_threads 降低至 1 可减少内存开销。
INSERT INTO pypi_v2
SELECT
timestamp,
project
FROM pypi
WHERE timestamp < '2024-12-17 09:00:00'
SETTINGS max_insert_threads = 1
0 行。耗时：27.752 秒。已处理 15 亿行，33.48 GB（5389 万行/秒，1.21 GB/秒）
峰值内存占用：506.78 MiB。
max_threads 设置调低至 1，以进一步减少内存占用。
INSERT INTO pypi_v2
SELECT timestamp, project
FROM pypi
WHERE timestamp < '2024-12-17 09:00:00'
SETTINGS max_insert_threads = 1, max_threads = 1
Ok.
0 rows in set. Elapsed: 43.907 sec. Processed 1.50 billion rows, 33.48 GB (34.06 million rows/s., 762.54 MB/s.)
Peak memory usage: 272.53 MiB.
min_insert_block_size_rows 设为 0 (不再将其作为决定块大小的因素) ，并将 min_insert_block_size_bytes 设为 10485760 (10MiB) ，以进一步降低内存占用。
INSERT INTO pypi_v2
SELECT
timestamp,
project
FROM pypi
WHERE timestamp < '2024-12-17 09:00:00'
SETTINGS max_insert_threads = 1, max_threads = 1, min_insert_block_size_rows = 0, min_insert_block_size_bytes = 10485760
0 rows in set. Elapsed: 43.293 sec. Processed 1.50 billion rows, 33.48 GB (34.54 million rows/s., 773.36 MB/s.)
Peak memory usage: 218.64 MiB.
- 暂停向主表插入数据。
- 使用
CREATE AS 语法为主目标表创建一个副本表。
- 使用
ALTER TABLE ATTACH 将原始目标表中的分区附加到副本表。注意： 此处的 attach 操作不同于前文使用的 move。虽然它依赖硬链接，但原始表中的数据会被保留。
- 创建新的 materialized view。
- 恢复插入。注意： 插入操作只会更新目标表，不会更新副本表；副本表只会引用原始数据。
- 对 materialized view 进行回填，沿用上文针对带时间戳数据的相同步骤，并将副本表作为来源表。
请看下面这个示例：使用 PyPI 和我们之前新建的 materialized view pypi_downloads_per_day (假设我们无法使用时间戳) ：
┌────count()─┐
│ 2039988137 │ -- 20.4 亿
└────────────┘
1 row in set. Elapsed: 0.003 sec.
-- (1) 暂停写入
-- (2) 创建目标表的副本
CREATE TABLE pypi_v2 AS pypi
SELECT count() FROM pypi_v2
┌────count()─┐
│ 2039988137 │ -- 20.4 亿
└────────────┘
结果中有 1 行。已用时间：0.004 秒。
-- (3) 将原始目标表的分区挂载到副本表。
ALTER TABLE pypi_v2
(ATTACH PARTITION tuple() FROM pypi)
-- (4) 创建新的 materialized view
CREATE TABLE pypi_downloads_per_day
(
`hour` DateTime,
`project` String,
`count` Int64
)
ENGINE = SummingMergeTree
ORDER BY (project, hour)
CREATE MATERIALIZED VIEW pypi_downloads_per_day_mv TO pypi_downloads_per_day
AS SELECT
toStartOfHour(timestamp) as hour,
project,
count() AS count
FROM pypi
GROUP BY
hour,
project
-- (4) 重启写入。此处通过插入单行来模拟复制。
INSERT INTO pypi SELECT *
FROM pypi
LIMIT 1
SELECT count() FROM pypi
┌────count()─┐
│ 2039988138 │ -- 20.4 亿
└────────────┘
1 row in set. Elapsed: 0.003 sec.
-- 注意 pypi_v2 包含的行数与之前相同
SELECT count() FROM pypi_v2
┌────count()─┐
│ 2039988137 │ -- 20.4亿
└────────────┘
-- (5) 使用备份表 pypi_v2 回填视图
INSERT INTO pypi_downloads_per_day SELECT
toStartOfHour(timestamp) as hour,
project,
count() AS count
FROM pypi_v2
GROUP BY
hour,
project
0 rows in set. Elapsed: 3.719 sec. Processed 2.04 billion rows, 47.15 GB (548.57 million rows/s., 12.68 GB/s.)
INSERT INTO SELECT 方法，对 pypi_downloads_per_day 执行回填。也可以进一步采用上文所述的 Null 表方法，并可选择使用副本表来增强稳健性。
虽然此操作确实需要暂停插入，但中间步骤通常都能很快完成，从而将数据中断降至最低。 
