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

在携程，我们为用户提供各种数字产品和服务，包括酒店和机票预订、景点门票、旅游套餐、商务旅行管理以及与旅行相关的各种内容。正如您可能猜到的，我们需要一个可扩展、健壮且快速的日志平台，这是我们运营顺利的关键。

在开始之前，为了激发您的好奇心，让我向您展示一些关于我们基于 ClickHouse 构建的平台的数字。

本文将讲述我们日志平台的故事，包括我们最初构建它的原因、我们使用的技术以及最终我们利用 SharedMergeTree 等功能在 ClickHouse 上规划的未来发展。

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

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

为了简化说明，让我们将其放在时间线上

每个伟大的故事都始于一个伟大的问题，就我们而言，这个项目开始于 2012 年之前，携程没有任何统一或集中的日志平台。每个团队和业务部门 (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 是携程开发和维护的一个开源应用程序，作为 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 生成段（segments），而 ClickHouse 写入数据块（parts）。虽然两者都随着时间的推移异步合并，创建更大的数据块和段，但 ClickHouse 通过基于 ORDER BY 键对数据进行排序的列式模型（columnar model）来实现差异化。这使得能够构建稀疏索引（sparse index）以进行快速过滤，并由于高压缩率而实现高效的存储利用率。您可以在 这篇优秀的指南中了解更多关于此索引机制的信息。

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

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

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

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

这是我们最终使用的初始表设计（请记住，这是几年前的事情，我们当时还没有 ClickHouse 中现在所有数据类型，例如 maps）。

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)

• 我们使用双列表方法来存储动态变化的标签（我们打算将来使用 maps），即我们有两个数组分别存储键和值。
• 按天分区以方便数据操作，对于我们的数据量，按天分区是有意义的，但在大多数情况下，更高的粒度，如按月或按周分区更好。
• 根据您将在查询中使用的过滤器，您可能希望使用与上表不同的 ORDER BY 键。上面的键针对使用 log_level 和 time 的查询进行了优化。例如，如果您的查询没有利用 log_level，那么只在键中包含 time 列是有意义的。
• Tokenbf_v1 布隆过滤器 用于优化术语查询和模糊查询。
• _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 的限制。我们使用 StatefulSets、反亲和性和 ConfigMaps 将这些集群部署到 Kubernetes 中。这将单个集群的交付时间从 2 天缩短到 5 分钟。同时，我们标准化了部署架构，简化了跨多个全球环境的部署流程。这种方法大大降低了我们的运营成本，并有助于实施上述方法。 

尽管上述方法解决了一些挑战，但它也引入了关于如何将用户的查询分配到特定集群的新复杂层。

让我们举个例子来说明。

假设我们有三个集群：集群 1、集群 2 和集群 3，以及三个表：A、B 和 C。在我们下面描述的虚拟表分区方法实施之前，单个表（如 A）只能驻留在一个数据集群中（例如，集群 1）。这种设计限制意味着当集群 1 的磁盘空间已满时，我们没有快速的方法将表 A 的数据迁移到集群 2 的相对空闲的磁盘空间中。相反，我们必须使用双写同时将表 A 的数据写入集群 1 和集群 2。然后，在集群 2 中的数据过期后（例如，7 天后），我们可以从集群 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 Enterprise Service（通过阿里云），它引入了 SharedMergeTree 表引擎。这提供了存储和计算的原生分离。通过这种新的架构，我们可以提供几乎无限的存储，以支持 trip.com 中更多日志用例。

阿里云提供的 ClickHouse Enterprise Service 与 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	平均值
阿里云企业版	第一轮	0.26	0.62	7.2	22.99	0.67
	第二轮	0.24	0.46	4.4	20.61	0.52
	第三轮	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%
阿里云社区版	第一轮	0.63	3.4	11.06	29.50	1.39
	第二轮	0.64	1.92	9.35	23.50	1.20
	第三轮	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的企业版服务使用对象存储作为数据存储方式，但它的性能仍然更好——尤其是在高并发工作负载方面。我们相信这种无缝就地升级可以为我们减轻大量的运维负担。

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