[译]在 ClickHouse 中处理实时更新

OLAP 数据库通常不欢迎使用可变数据（Mutable data）。ClickHouse 也不例外。与其他 OLAP 产品一样，ClickHouse 最初甚至不支持更新。后来，添加了更新，但是像许多其他内容一样，它们以ClickHouse方式添加。即使是现在，ClickHouse 更新也是异步的，这使得它们很难在交互式应用程序中使用。尽管如此，在许多用例中，用户需要对现有数据进行修改，并希望立即看到效果。ClickHouse 可以做到吗？当然可以。

ClickHouse 更新的简短历史

早在 2016 年，ClickHouse 团队就发表了一篇题为如何在 ClickHouse 中更新数据^[1]的文章。ClickHouse 当时不支持数据修改。为了模拟更新，只能使用特殊的插入结构，并且数据必须由分区删除。

在 GDPR 的压力下，ClickHouse 团队在 2018 年提出了 UPDATE 和 DELETE 的要求。后续文章 ClickHouse 中的Updates 和 Deletes^[2]仍然是 Altinity 博客中阅读最多的文章之一。这些异步的，非原子的更新被实现为 ALTER TABLE UPDATE 语句，并且有可能打乱很多数据。当不需要立即结果时，这对于批量操作和不频繁的更新很有用。ClickHouse 仍然缺少常规SQL 更新，尽管它们每年确实可靠地出现在路线图中。如果需要实时更新行为，我们必须使用其他方法。让我们考虑一个实际的用例，并比较 ClickHouse 中使用它的不同方法。

用例

考虑一个生成各种警报的系统。用户或机器学习算法会不时查询数据库，以查看新警报并进行确认。确认操作需要修改数据库中的警报记录。确认后，警报应从用户视图中消失。这看起来像是 ClickHouse 的 OLTP 操作。

由于我们无法使用更新，因此我们将不得不插入修改后的记录。数据库中有两条记录后，我们需要一种有效的方法来获取最新的一条。为此，我们将尝试三种不同的方法：

• ReplacingMergeTree
• Aggregate functions
• AggregatingMergeTree

ReplacingMergeTree

让我们从创建一个存储警报的表开始。

CREATE TABLE alerts(
  tenant_id     UInt32,
  alert_id      String,
  timestamp     DateTime Codec(Delta, LZ4),
  alert_data    String,
  acked         UInt8 DEFAULT 0,
  ack_time      DateTime DEFAULT toDateTime(0),
  ack_user      LowCardinality(String) DEFAULT ''
)
ENGINE = ReplacingMergeTree(ack_time)
PARTITION BY tuple()
ORDER BY (tenant_id, timestamp, alert_id);

为简单起见，所有警报特定列都打包到一个通用 alert_data 列中。但是您可以想象警报可能包含数十甚至数百列。另外，alert_id在我们的示例中是一个随机字符串。

注意 ReplacingMergeTree引擎。ReplacecingMergeTee 是一种特殊的表引擎，它用主键（ORDER BY）替换数据-具有相同键值的行的较新版本将替换较旧的行。在我们的案例中，数据新旧由ack_time列确定。替换是在后台合并操作期间执行的。它不会立即发生，也不能完全保证会发生，因此查询结果的一致性是一个问题。不过，ClickHouse 具有可用于此类表格的特殊语法，我们将在下面的查询中使用它。

在运行查询之前，让我们在表中填充一些数据。我们为 1000 个租户生成了 1000 万个警报：

INSERT INTO alerts(tenant_id, alert_id, timestamp, alert_data)
SELECT
  toUInt32(rand(1)%1000+1) AS tenant_id,
  randomPrintableASCII(64) as alert_id,
  toDateTime('2020-01-01 00:00:00') + rand(2)%(3600*24*30) as timestamp,
  randomPrintableASCII(1024) as alert_data
FROM numbers(10000000);

接下来，让我们确认 99％的警报，为acked，ack_user和ack_time列提供新值。代替更新，我们只是插入一个新行。

INSERT INTO alerts (tenant_id, alert_id, timestamp, alert_data, acked, ack_user, ack_time)
SELECT tenant_id, alert_id, timestamp, alert_data, 1 as acked,
concat('user', toString(rand()%1000)) as ack_user, now() as ack_time
FROM alerts WHERE cityHash64(alert_id) % 99 != 0;

如果现在查询此表，我们将看到类似以下内容的内容：

SELECT count() FROM alerts
┌──count()─┐
│ 19898060 │
└──────────┘
1 rows in set. Elapsed: 0.008 sec.

因此，我们肯定在表中既有已确认的行也有未确认的行。因此，替换尚未发生。为了查看真实数据，我们必须添加FINAL关键字。

SELECT count() FROM alerts FINAL
┌──count()─┐
│ 10000000 │
└──────────┘
1 rows in set. Elapsed: 3.693 sec. Processed 19.90 million rows, 1.71 GB (5.39 million rows/s., 463.39 MB/s.)

现在该计数是正确的，但请查看查询时间！借助 FINAL，ClickHouse 必须扫描所有行并在查询时间内通过主键合并它们。这会产生正确的答案，但会带来很多开销。让我们看看是否可以通过仅过滤未确认的行来做得更好。

SELECT count() FROM alerts FINAL WHERE NOT acked
┌─count()─┐
│  101940 │
└─────────┘
1 rows in set. Elapsed: 3.570 sec. Processed 19.07 million rows, 1.64 GB (5.34 million rows/s., 459.38 MB/s.)

查询时间和处理的数据量是相同的，即使数量要小得多。筛选无助于加快查询速度。随着表大小的增加，成本可能会更大。它不缩放。

注意：为了便于阅读，所有查询和查询时间都显示为好像在clickhouse-client中运行一样。实际上，我们多次尝试查询以确保结果一致，并使用clickhouse-benchmark实用工具进行确认。

好的，查询整个表并没有帮助。我们仍然可以将 ReplacingMergeTree 用于我们的用例吗？让我们选择一个随机的 tenant_id，然后选择所有尚未确认的记录-假设有一个用户正在查看的仪表板。我喜欢 Ray Bradbury，所以选择了 451。由于alert_data只是随机垃圾，因此我们将计算一个校验和，并将其用于确认结果在多种方法中是相同的：

SELECT
  count(),
  sum(cityHash64(*)) AS data
FROM alerts FINAL
WHERE (tenant_id = 451) AND (NOT acked)
┌─count()─┬─────────────────data─┐
│      90 │ 18441617166277032220 │
└─────────┴──────────────────────┘
1 rows in set. Elapsed: 0.278 sec. Processed 106.50 thousand rows, 119.52 MB (383.45 thousand rows/s., 430.33 MB/s.)

太快了！在 278 毫秒内，我们可以查询所有未确认的数据。为什么这次速度这么快？区别在于过滤条件。tenant_id是主键的一部分，因此 ClickHouse 可以在 FINAL 之前过滤数据。在这种情况下，ReplacecingMergeTree 变得有效。

我们也尝试使用用户过滤器，并查询特定用户确认的警报数量。列的基数相同-我们有 1000 个用户，可以尝试 user451。

SELECT count() FROM alerts FINAL
WHERE (ack_user = 'user451') AND acked
┌─count()─┐
│    9725 │
└─────────┘
1 rows in set. Elapsed: 4.778 sec. Processed 19.04 million rows, 1.69 GB (3.98 million rows/s., 353.21 MB/s.)

这下很慢，因为未使用索引。ClickHouse 扫描了所有 1,904 万行。请注意，我们无法将ack_user添加到索引，因为它会破坏 ReplacingMergeTree 的语义。不过，我们可以使用 PREWHERE 来解决问题：

SELECT count() FROM alerts FINAL
PREWHERE (ack_user = 'user451') AND acked
┌─count()─┐
│    9725 │
└─────────┘
1 rows in set. Elapsed: 0.639 sec. Processed 19.04 million rows, 942.40 MB (29.80 million rows/s., 1.48 GB/s.)

PREWHERE 是 ClickHouse 的特殊提示，可以不同地应用过滤器。通常，ClickHouse 足够聪明，可以自动将条件移至 PREWHERE，因此用户无需理会。这次没有发生，因此我们检查了很好。

Aggregate Functions(聚合函数）

ClickHouse 以支持多种聚合功能而闻名。在最新版本中，它已超过 100 个。结合 9 个聚合函数组合器（请参阅combinators^[3]），这为经验丰富的用户提供了极大的灵活性。对于此用例，我们不需要任何高级功能，仅使用 3 个函数：argMax，max和any。

可以使用argMax聚合函数对第 451 个租户执行相同的查询，如下所示：

SELECT count(), sum(cityHash64(*)) data FROM (
  SELECT tenant_id, alert_id, timestamp,
         argMax(alert_data, ack_time) alert_data,
         argMax(acked, ack_time) acked,
         max(ack_time) ack_time_,
         argMax(ack_user, ack_time) ack_user
  FROM alerts
  GROUP BY tenant_id, alert_id, timestamp
)
WHERE tenant_id=451 AND NOT acked;
┌─count()─┬─────────────────data─┐
│      90 │ 18441617166277032220 │
└─────────┴──────────────────────┘
1 rows in set. Elapsed: 0.059 sec. Processed 73.73 thousand rows, 82.74 MB (1.25 million rows/s., 1.40 GB/s.)

结果相同，行数相同，但性能提高了 4 倍！这是 ClickHouse 聚合效率。缺点是查询变得更加复杂。但是我们可以使其更简单。

让我们注意，确认警报时，我们仅更新 3 列：

• 确认：0 => 1
• ack_time：0 => now()
• ack_user：”=>’user1′

在所有 3 种情况下，列值都会增加！因此，我们可以使用max代替笨重的argMax。由于我们不更改alert_data，因此在此列上不需要任何实际的汇总。ClickHouse 为此提供了很好的any聚合功能。它可以选择任何值而不会产生额外的开销：

查询变得简单，并且速度稍快！原因是使用any功能，ClickHouse 无需在alert_data列上计算max！

AggregatingMergeTree 聚合合并树

AggregatingMergeTree 是 ClickHouse 最强大的功能之一。与实例化视图结合使用时，它可以实现实时数据聚合。由于我们在先前的方法中使用了聚合函数，因此我们可以通过 AggregatingMergeTree 使其更好吗？实际上，这没有太大的改进。

我们只更新一次行，因此一个组只需要聚合两行。对于这种情况，AggregatingMergeTree 不是最佳选择。但是，我们可以做个技巧。我们知道，警报总是先插入为未确认，然后再被确认。用户确认警报后，只需修改 3 列。如果我们不重复其他列的数据，是否可以节省磁盘空间并提高性能？

让我们创建一个表，该表使用max聚合函数实现聚合。除了max之外，我们还可以使用any，但这将要求列为可空的-any会选择一个非空值。

DROP TABLE alerts_amt_max;
CREATE TABLE alerts_amt_max (
  tenant_id     UInt32,
  alert_id      String,
  timestamp     DateTime Codec(Delta, LZ4),
  alert_data    SimpleAggregateFunction(max, String),
  acked         SimpleAggregateFunction(max, UInt8),
  ack_time      SimpleAggregateFunction(max, DateTime),
  ack_user      SimpleAggregateFunction(max, LowCardinality(String))
)
Engine = AggregatingMergeTree()
ORDER BY (tenant_id, timestamp, alert_id);

由于原始数据是随机的，因此我们将使用来自alert的现有数据填充新表。我们将像以前一样在两个插入中执行此操作，一个插入未确认的警报，另一个插入已确认的警报：

INSERT INTO alerts_amt_max SELECT * FROM alerts WHERE NOT acked;
INSERT INTO alerts_amt_max
SELECT tenant_id, alert_id, timestamp,
  '' as alert_data,
  acked, ack_time, ack_user
FROM alerts WHERE acked;

请注意，对于已确认的事件，我们将插入一个空字符串而不是alert_data。我们知道数据不会改变，并且只能存储一次！聚合函数将填补空白。在实际的应用程序中，我们可以跳过所有未更改的列，并让它们获取默认值。

有了数据后，让我们首先检查数据大小：

SELECT
    table,
    sum(rows) AS r,
    sum(data_compressed_bytes) AS c,
    sum(data_uncompressed_bytes) AS uc,
    uc / c AS ratio
FROM system.parts
WHERE active AND (database = 'last_state')
GROUP BY table
┌─table──────────┬────────r─┬───────────c─┬──────────uc─┬──────────────ratio─┐
│ alerts         │ 19039439 │ 20926009562 │ 21049307710 │ 1.0058921003373666 │
│ alerts_amt_max │ 19039439 │ 10723636061 │ 10902048178 │ 1.0166372782501314 │
└────────────────┴──────────┴─────────────┴─────────────┴────────────────────┘

好吧，由于随机字符串，我们几乎没有压缩。但是聚合小了两倍，因为我们不必两次存储alerts_data。

现在，让我们尝试对聚合表进行查询：

SELECT count(), sum(cityHash64(*)) data FROM (
   SELECT tenant_id, alert_id, timestamp,
          max(alert_data) alert_data,
          max(acked) acked,
          max(ack_time) ack_time,
          max(ack_user) ack_user
     FROM alerts_amt_max
   GROUP BY tenant_id, alert_id, timestamp
)
WHERE tenant_id=451 AND NOT acked;
┌─count()─┬─────────────────data─┐
│      90 │ 18441617166277032220 │
└─────────┴──────────────────────┘
1 rows in set. Elapsed: 0.036 sec. Processed 73.73 thousand rows, 40.75 MB (2.04 million rows/s., 1.13 GB/s.)

多亏了 AggregatingMergeTree，我们可以处理更少的数据（以前是 40MB，之前是 82MB），现在它的效率更高。

实现更新

ClickHouse 将尽其所能在后台合并数据，删除重复的行并执行聚合。但是，有时为了强制释放磁盘空间（例如释放磁盘空间），这是有意义的。这可以通过 OPTIMIZE FINAL 语句完成。优化是一项阻塞性且昂贵的操作，因此不能过于频繁地执行。让我们看看它对查询性能是否有任何影响。

OPTIMIZE TABLE alerts FINAL
Ok.
0 rows in set. Elapsed: 105.675 sec.
OPTIMIZE TABLE alerts_amt_max FINAL
Ok.
0 rows in set. Elapsed: 70.121 sec.

在 OPTIMIZE FINAL 之后，两个表具有相同的行数和相同的数据。

┌─table──────────┬────────r─┬───────────c─┬──────────uc─┬────────────ratio─┐
│ alerts         │ 10000000 │ 10616223201 │ 10859490300 │ 1.02291465565429 │
│ alerts_amt_max │ 10000000 │ 10616223201 │ 10859490300 │ 1.02291465565429 │
└────────────────┴──────────┴─────────────┴─────────────┴──────────────────┘

不同方法之间的性能差异变得不太明显。这是汇总表：

结论

ClickHouse 提供了丰富的工具集来处理实时更新，例如 ReplacecingMergeTree，CollapsingMergeTree（此处未进行评论），AggregatingMergeTree 和聚合函数。所有这些方法具有三个共同的属性：

• 通过插入新版本来修改数据。ClickHouse 中的插入速度非常快。
• 有许多有效的方法可以模拟与 OLTP 数据库相似的更新语义
• 但是，实际的修改不会立即发生。

特定方法的选择取决于应用程序用例。ReplacecingMergeTree 简单明了，对用户来说最方便，但仅可用于中小型表，或者始终由主键查询数据时使用。聚合函数的使用提供了更大的灵活性和性能，但需要大量的查询重写。最后，AggregatingMergeTree 可以节省存储空间，仅保留修改后的列。这些是 ClickHouse DB 设计人员中可以使用的好工具，并在需要时应用。

参考资料