trip.com 如何从 Elasticsearch 迁移并使用 ClickHouse 构建 50PB 日志解决方案

在 trip.com，我们为用户提供广泛的数字产品，包括酒店和机票预订、景点、旅游套餐、商务旅行管理以及旅行相关内容。正如您可能猜到的那样，我们对可扩展、稳健且快速的日志记录平台的需求是我们运营良好运作的关键。

在我们开始之前，为了稍微激起您的好奇心，让我向您展示一些数字，突出我们在 ClickHouse 之上构建的平台

这篇博文将解释我们的日志记录平台的故事，我们最初构建它的原因，我们使用的技术，以及最终，我们利用 SharedMergeTree 等功能在 ClickHouse 之上对其未来的计划。

以下是我们旅程中将要讨论的不同主题

• 我们如何构建集中式日志记录平台
• 我们如何扩展日志记录平台并从 Elasticsearch 迁移到 ClickHouse
• 我们如何改善我们的运营体验
• 我们如何在阿里云上测试 ClickHouse Cloud

为了简化，让我们将其放在时间轴上

每个伟大的故事都始于一个伟大的问题，在我们的案例中，这个项目开始是因为在 2012 年之前，trip.com 没有任何统一或集中的日志记录平台。由于每个团队和业务部门 (BU) 都在收集和管理自己的日志，这带来了许多不同的挑战

• 需要大量人力来开发、维护和运营所有这些环境，这不可避免地导致了大量重复的工作。
• 数据治理和控制变得复杂。
• 公司内部没有统一的标准

有了这些，我们知道我们需要构建一个集中式和统一的日志记录平台。

2012 年，我们推出了第一个平台。它建立在 Elasticsearch 之上，并开始定义 ETL、存储、日志访问和查询的标准。

即使我们不再将 Elasticsearch 用于我们的日志记录平台，也可能值得探讨我们如何实施我们的解决方案。它推动了我们后续的大部分工作，我们在后来迁移到 ClickHouse 时不得不考虑这一点。

我们的 Elasticsearch 集群主要由主节点、协调器节点和数据节点组成。

每个 Elasticsearch 集群都至少由三个符合主节点条件的节点组成。在这些节点中，将选举一个主节点，负责维护集群状态。集群状态是元数据，包含有关各种索引、分片、副本等的信息。任何修改集群状态的操作都将由主节点执行。

数据节点存储数据，并将用于执行 CRUD 操作。这些可以分为多个层：热层、温层等。

这种类型的节点没有任何其他功能（主节点、数据节点、摄取节点、转换节点等），而是充当智能负载均衡器，通过考虑集群状态。如果协调器接收到带有 CRUD 操作的查询，它将被发送到数据节点。或者，如果他们收到添加或删除索引的查询，它将被发送到主节点。

在 Elasticsearch 之上，我们使用 Kibana 作为可视化层。您可以在下面看到可视化的示例

我们的用户有两种将日志发送到平台的选项：通过 Kafka 和通过代理。

第一种方法涉及使用公司的框架 TripLog 将数据摄取到 Kafka 消息代理（使用 Hermes）。

private static final Logger log = LoggerFactory.getLogger(Demo.class);

public void demo (){
  TagMarker marker = TagMarkerBuilder.newBuilder().scenario("demo").addTag("tagA", "valueA").addTag("tagA", "valueA").build();
  log.info(marker, "Hello World!");
}

这为我们的用户提供了一个框架，可以轻松地将日志发送到我们的平台。

另一种方法是使用代理，例如 Filebeat、Logstash、Logagent 或自定义客户端，它们将直接写入 Kafka。您可以在下面看到 Filebeat 配置的示例

filebeat.config.inputs:
  enabled: true
  path: "/path/to/your/filebeat/config"
filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /var/log/history.log
      - /var/log/auth.log
      - /var/log/secure
      - /var/log/messages
    harvester_buffer_size: 102400
    max_bytes: 100000
    tail_files: true
    fields:
      type: os
    ignore_older: 30m
    close_inactive: 2m
    close_timeout: 40m
    close_removed: true
    clean_removed: true
output.kafka:
  hosts: ["kafka_broker1", "kafka_broker2"]
  topic: "logs-%{[fields.type]}"
  required_acks: 0
  compression: none
  max_message_bytes: 1000000
processors:
  - rename:
      when:
        equals:
          source: "message"
          target: "log_message"

无论用户选择哪种方法，数据最终都会进入 Kafka，在那里可以使用 gohangout 将其管道传输到 Elasticsearch。

Gohangout 是 trip.com 开发和维护的开源应用程序，作为 Logstash 的替代品。它旨在从 Kafka 消费数据，执行 ETL 操作，并最终将数据输出到各种存储介质，例如 ClickHouse 和 Elasticsearch。Filter 模块中的数据处理包括用于数据清理的常用功能，例如 JSON 处理、Grok 模式匹配和时间转换（如下所示）。在下面的示例中，GoHangout 使用正则表达式匹配从 Message 字段中提取 num 数据，并将其存储为单独的字段。

许多人使用 Elasticsearch 进行可观测性，它在较小数据量的情况下表现出色。它们提供易于使用的软件、无模式体验、广泛的功能以及带有 Kibana 的流行 UI。但是，当在我们这种规模上部署时，它存在众所周知的挑战。

当我们在 Elasticsearch 中存储 4PB 数据时，我们开始面临围绕集群稳定性的多个问题

1. 集群上的高负载导致许多请求被拒绝、写入延迟和查询缓慢
2. 每天将 200 TB 数据从热节点迁移到冷节点导致明显的性能下降
3. 分片分配是一个挑战，并导致某些节点不堪重负
4. 大型查询导致内存不足 (OOM) 异常。

围绕集群性能

1. 查询速度受到整体集群状态的影响
2. 由于摄取期间的高 CPU 使用率，我们难以提高插入吞吐量

最后，围绕成本

1. 数据量、数据结构和缺乏压缩导致需要大量存储
2. 较弱的压缩率对业务产生影响，并迫使我们缩短保留期
3. Elasticsearch 的 JVM 和内存限制导致更高的 TCO（总拥有成本）

因此，在意识到以上所有问题后，我们寻找了替代方案，ClickHouse 应运而生！

Elasticsearch 和 ClickHouse 之间存在一些根本的区别；让我们一起了解一下。

Elasticsearch 依赖于一种特定的查询语言，称为 Query DSL（域特定语言）。即使现在有更多选项，这仍然是主要语法。另一方面，ClickHouse 依赖于 SQL，SQL 非常主流，并且用户友好，并且与许多不同的集成和 BI 工具兼容。

Elasticsearch 和 ClickHouse 在内部行为上有一些相似之处，Elasticsearch 生成段，ClickHouse 写入部件。虽然两者都会随着时间的推移异步合并，创建更大的部件和段，但 ClickHouse 以列式模型脱颖而出，其中数据通过 ORDER BY 键排序。这允许构建稀疏索引，以实现快速过滤和高效的存储使用，这归功于高压缩率。您可以在这份出色的指南中阅读有关此索引机制的更多信息。

Elasticsearch 中的数据存储在索引中，并分解为分片。这些需要保持在相对较小的尺寸范围内（在当时，建议分片大小约为 50GB）。相比之下，ClickHouse 数据存储在表中，这些表可能大得多（在 TB 范围内，如果不受磁盘大小限制，则更大）。最重要的是，ClickHouse 允许您创建分区键，这将数据物理地分隔到不同的文件夹中。然后可以有效地操作这些分区（如果需要）。

总的来说，ClickHouse 的功能和特性给我们留下了深刻的印象：它的列式存储、向量化查询执行、高压缩率和高插入吞吐量。这些满足了我们的日志记录解决方案对性能、稳定性和成本效益的需求。因此，我们决定使用 ClickHouse 替换我们的存储和查询层。

下一个挑战是如何在不中断服务的情况下无缝地从一个存储迁移到另一个存储。

在决定迁移到 Clickhouse 后，我们确定了几个需要完成的不同任务

这是我们最终确定的初始表设计（请记住，那是几年前，我们没有今天 ClickHouse 中的所有数据类型，例如 map）

CREATE TABLE log.example
(
  `timestamp` DateTime64(9) CODEC(ZSTD(1)),
  `_log_increment_id` Int64 CODEC(ZSTD(1)),
  `host_ip` LowCardinality(String) CODEC(ZSTD(1)),
  `host_name` LowCardinality(String) CODEC(ZSTD(1)),
  `log_level` LowCardinality(String) CODEC(ZSTD(1)),
  `message` String CODEC(ZSTD(1)),
  `message_prefix` String MATERIALIZED substring(message, 1, 128) CODEC(ZSTD(1)),
  `_tag_keys` Array(LowCardinality(String)) CODEC(ZSTD(1)),
  `_tag_vals` Array(String) CODEC(ZSTD(1)),
  `log_type` LowCardinality(String) CODEC(ZSTD(1)),
   ...
   INDEX idx_message_prefix message_prefix TYPE tokenbf_v1(8192, 2, 0) GRANULARITY 16,
   ...
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/example', '{replica}')
PARTITION BY toYYYYMMDD(timestamp)
ORDER BY (log_level, timestamp, host_ip, host_name)
TTL toDateTime(timestamp) + toIntervalHour(168)

• 我们使用双列表方法来存储动态变化的标签（我们计划将来使用 map），即我们有两个数组分别存储键和值。
• 按天分区以便于数据操作。对于我们的数据量，每日分区是有意义的，但在大多数情况下，更高级别的粒度（如每月或每周）更好。
• 根据您将在查询中使用的过滤器，您可能希望拥有与上表不同的 ORDER BY 键。上面的键针对使用 log_level 和 time 的查询进行了优化。例如，如果您的查询没有利用 log_level，则仅在键中包含 time 列是有意义的。
• Tokenbf_v1 Bloom 过滤器，用于优化术语查询和模糊查询。
• _log_increment_id 列包含全局唯一的增量 ID，以实现高效的滚动分页和精确的数据定位。
• ZSTD 数据压缩方法，节省超过 40% 的存储成本。

鉴于我们过去的设置和 Elasticsearch 的经验，我们决定复制类似的架构。我们的 ClickHouse-Keeper 实例充当主节点（类似于 Elasticsearch）。部署了多个查询节点，这些节点不存储数据，但持有指向 ClickHouse 服务器的分布式表。这些服务器托管数据节点，这些数据节点存储和写入数据。下图显示了我们最终的架构

我们希望在迁移到 ClickHouse 后为用户提供无缝体验。为了做到这一点，我们需要确保他们所有的可视化和仪表板都可以使用 ClickHouse。这带来了挑战，因为 Kibana 是最初在 Elasticsearch 之上开发的工具，并且不支持额外的存储引擎。因此，我们不得不对其进行自定义，以确保它可以与 ClickHouse 接口。这需要我们在 Kibana 中创建新的数据面板，这些面板可以与 ClickHouse 一起使用：chhistogram、chhits、chpercentiles、chranges、chstats、chtable、chterms 和 chuniq。

然后，我们创建了脚本，将 95% 的现有 Kibana 仪表板迁移到使用 Clickhouse。最后，我们增强了 Kibana，以便用户可以编写 SQL 查询。

我们的日志记录管道是自助服务式的，允许用户发送日志。这些用户需要能够创建索引并定义所有权、权限和 TTL 策略。因此，我们创建了一个名为 Triplog 的平台，为用户提供了一个界面来管理他们的表、用户和角色、监控他们的数据流以及创建警报。

现在一切都已迁移，是时候看看我们平台的新性能了！即使我们自动化了 95% 的迁移并实现了无缝过渡，重要的是回顾我们的成功指标，看看新平台的表现如何。最重要的两个是查询性能和总拥有成本 (TCO)。

我们原始成本的重要组成部分是存储。让我们比较一下 Elasticsearch 和 ClickHouse 在相同数据样本的存储方面的差异

存储空间节省超过 50%，使现有 Elasticsearch 服务器能够使用 ClickHouse 支持 4 倍的数据量增长。

查询速度比 ElasticSearch 快 4 到 30 倍，P90 小于 300 毫秒，P99 小于 1.5 秒。

自从我们在 2022 年完成从 Elasticsearch 的迁移以来，我们向我们的平台添加了更多的日志记录用例，将其从 4PB 增长到 20PB。随着它继续增长并扩展到 30PB，我们面临着新的挑战。

1. 在这个规模的单个 ClickHouse 集群很难管理。在部署时，没有 ClickHouse-Keeper 或 SharedMergeTree，我们面临着围绕 Zookeeper 的性能挑战，导致 DDL 超时异常。
2. 用户选择的索引不佳导致查询性能欠佳，并且需要使用更好的模式重新插入数据。
3. 编写不当且未优化的查询导致性能问题。

1. 集群构建依赖于 Ansible，导致部署周期长（数小时）。
2. 我们当前的 ClickHouse 实例落后于社区版本多个版本，并且当前的集群部署模式不便于执行更新。

为了解决上述性能挑战，我们首先放弃了单集群方法。在我们的规模下，在没有 SharedMergeTree 和 ClickHouse Keeper 的情况下，元数据的管理变得困难，并且由于 Zookeeper 瓶颈，我们会遇到 DDL 语句超时。因此，我们没有保留单个集群，而是创建了多个集群，如下所示

这种新架构帮助我们扩展并克服了 Zookeeper 的限制。我们将这些集群部署到 Kubernetes，使用 StatefulSets、反亲和性和 ConfigMaps。这使得单个集群的交付时间从 2 天缩短到 5 分钟。同时，我们标准化了部署架构，简化了全球多个环境的部署流程。这种方法显着降低了我们的运营成本，并有助于实施上述方法。

尽管上述方法解决了很多挑战，但它引入了一个新的复杂性层，即我们如何将用户的查询分配给特定的集群。

让我们举个例子来说明一下

假设我们有三个集群：集群 1、集群 2 和集群 3，以及三个表：A、B 和 C。在我们描述的虚拟表分区方法实施之前，单个表（如 A）只能驻留在一个数据集群中（例如，集群 1）。这种设计限制意味着，当集群 1 的磁盘空间已满时，我们没有快速的方法将表 A 的数据迁移到集群 2 相对空的磁盘空间。相反，我们必须使用双写将表 A 的数据同时写入集群 1 和集群 2。然后，在集群 2 中的数据过期后（例如，七天后），我们可以从集群 1 中删除表 A 的数据。这个过程很麻烦且缓慢，需要大量的人工工作来管理集群。

为了解决这个问题，我们设计了一个类分区架构，使表 A 能够在多个集群（集群 1、集群 2 和集群 3）之间来回移动。如右侧转型后所示，表 A 的数据根据时间间隔进行分区（可以精确到秒，但为简单起见，我们在此处使用天作为示例）。例如，6 月 8 日的数据写入集群 1，6 月 9 日的数据写入集群 2，8 月 10 日的数据写入集群 3。当查询命中 6 月 8 日的数据时，我们只查询集群 1 的数据。当查询需要 6 月 9 日和 10 日的数据时，我们同时查询集群 2 和集群 3 的数据。

我们通过建立不同的分布式表来实现此功能，每个分布式表代表特定时间段的数据，并且每个分布式表都与集群的逻辑组合（例如，集群 1、集群 2 和集群 3）相关联。这种方法解决了表跨集群的问题，并且不同集群之间的磁盘使用率趋于更加平衡。

您可以在上图中看到，每个查询都将根据其 WHERE 子句，通过代理智能地重定向到包含所需表的正确集群。

这种架构还可以帮助随着时间的推移进行模式演变。由于可以添加和删除列，因此某些表可能具有更多或更少的列。上述路由可以应用于列级别来解决此问题，代理能够过滤掉不包含查询所需列的表。

除了以上之外，这种架构还可以帮助我们支持不断演变的 ORDER BY 键 - 通常，对于 ClickHouse，您无法动态更改表的 ORDER BY 键。使用上述方法，您只需更改新表上的 ORDER BY 键，并让旧表过期（感谢 TTL）。

在查询层，我们使用 Antlr4 技术将用户的 SQL 查询解析为抽象语法树 (AST)。借助 AST 树，我们可以快速从 SQL 查询中获取表名、过滤条件和聚合维度等信息。有了这些信息，我们可以轻松地为 SQL 查询实施实时定向策略，例如数据统计、查询重写和治理流量控制。

我们为所有用户 SQL 查询实施了统一的查询网关代理。该程序根据元数据信息和策略重写用户 SQL 查询，以提供精确路由和自动性能优化等功能。此外，它还记录每个查询的详细上下文，用于集群查询的统一治理，对 QPS、大型表扫描和查询执行时间施加限制，以提高系统稳定性。

我们的平台已在 40PB+ 规模上得到验证，但仍有许多地方需要改进。我们希望更具动态可扩展性，以便更优雅地应对假期等期间的高峰使用量。为了应对这种增长，我们开始探索 ClickHouse 企业服务（通过阿里云），该服务引入了 SharedMergeTree 表引擎。这提供了存储和计算的本机分离。借助这种新架构，我们可以提供几乎无限的存储来支持 trip.com 中更多的日志记录用例。

阿里云提供的 ClickHouse 企业服务与 ClickHouse Cloud 使用的 ClickHouse 版本相同。

为了测试 ClickHouse 企业服务，我们首先对数据进行双写，将其同时插入到我们现有的部署和利用 SharedMergeTree 的新服务中。为了模拟真实的工作负载，我们

• 将 3TB 的数据加载到两个集群中，然后进行持续的插入负载。
• 收集了各种查询模板以用作测试集。
• 使用脚本，我们构建了查询，这些查询将查询随机的 1 小时时间间隔，并带有保证非空结果集的特定值。

当涉及到使用的基础设施时

• 3 个 32 CPU 节点，具有 128 GiB 内存，使用对象存储进行 ClickHouse 企业产品（带有 SMT）
• 2 个 40 CPU 节点，具有 176 GiB 内存，带有 HDD 用于社区版（开源）

为了执行我们的查询工作负载，我们使用了 clickhouse-benchmark 工具（ClickHouse 随附）用于两种服务。

1. 企业版和社区版都配置为使用文件系统缓存，因为我们希望重现与我们在生产环境中可能遇到的类似条件（考虑到数据量会大得多，我们应该预期生产环境中的缓存命中率会更低）
2. 我们将以 2 的并发性运行第一个测试，每个查询将在 3 个不同的轮次中执行。

	测试轮次	P50	P90	P99	P9999	平均值
阿里云企业版	第 1 轮	0.26	0.62	7.2	22.99	0.67
	第 2 轮	0.24	0.46	4.4	20.61	0.52
	第 3 轮	0.24	0.48	16.75	21.71	0.70
	平均值	0.246 40.3%	0.52 22.2%	7.05 71.4%	21.77 90.3	0.63 51.6%
阿里云社区版	第 1 轮	0.63	3.4	11.06	29.50	1.39
	第 2 轮	0.64	1.92	9.35	23.50	1.20
	第 3 轮	0.58	1.60	9.23	19.3	1.07
	平均值	0.61 100%	2.31 100%	9.88 100%	24.1 100%	1.07 100%

来自 ClickHouse 企业服务的结果以黄色显示，阿里云社区版的结果以红色显示。相对于社区版的性能百分比以绿色显示（越低越好）。

现在，随着我们增加并发性，社区版很快就无法处理工作负载，并开始返回错误。这实际上意味着企业版能够有效地处理三倍多的并发查询。

尽管 ClickHouse 的企业服务使用对象存储作为其存储数据的方式，但它的性能仍然更好 - 尤其是在高并发工作负载方面。我们相信这种无缝的原地升级可以消除我们的大量运营负担。

作为此测试的结果，我们决定开始将我们的业务指标迁移到企业服务。这包含有关付款完成率、订单统计信息等的信息，我们建议所有社区用户都尝试一下企业服务！