Netflix 时序数据冷热分层与 Cassandra 冷读

引言

在之前的一篇文章中，我们介绍了 TimeSeries Abstraction，这是我们高吞吐、低延迟的平台，旨在规模化地存储和查询时间事件数据。此后，它已成为 Netflix 最大数据集的运营骨干，而这些数据集如今已达到数 PB 级别。

然而，在热存储中管理如此庞大的数据集，针对低毫秒级延迟访问进行了调优，并不便宜。在 Netflix，成本效率是一项持续性的纪律，而非临时性的举措。本文分享两种解决方案，它们通过持续将较旧、访问频率较低的数据迁移到更便宜的对象存储中，显著降低了存储成本：

• 方案 1：使用Apache Kafka和Flink将数据直接流式传输到S3
• 方案 2：利用 S3 中的Cassandra备份和可插拔 Memtables直接从 S3 读取数据

我们将从基于 Kafka/Flink 的设计开始，然后描述我们如何演进到基于 Cassandra 原生的冷层。

挑战

随着 Netflix 的 TimeSeries 数据集不断扩展，我们遇到了若干交织叠加的挑战：

• 不断增长的保留需求：用户经常推动延长数据保留时间，以服务更广泛的使用场景，这反过来又增加了存储成本。
• 不断攀升的成本：为不常访问的历史数据维护大规模、高密度的集群，成本高得令人望而却步。
• 稀疏的历史访问：大多数查询针对的是近期事件，但较旧的数据必须随时可用，以应对偶尔的审计和调查。
• 运维压力：热节点上的高存储密度使扩缩容、部署和集群稳定性变得更加复杂。
• 延迟要求：即使是历史数据，有时也需要亚秒级的查询性能，以支持内部工具和分析。

这创造了一个明确的机会：将最新数据保留在快速存储上，并将较旧、很少读取的数据迁移到更便宜的存储上，同时保持一致的 API。如果做得好，这可以在满足用户和运维需求的同时，大幅降低我们的基础设施支出。

TimeSeries Abstraction

在我们深入探讨冷存储解决方案之前，这里快速回顾一下 TimeSeries 如何将其数据集分区为多个表，每个表覆盖一个时间窗口，我们称之为时间切片。

这带来了若干关键优化，其中最值得注意的是，一旦表越过写入边界，就可以变为不可变，这让我们能够安全地应用进一步的优化。更多细节请参阅我们之前的文章。

这些切片信息至关重要，并维护在一个专用的元数据表中：

创建新的时间切片并维护其元数据，是 TimeSeries 控制平面的主要职责。TimeSeries 数据平面使用这些信息，将读写流量正确路由到已配置的表。

有了这些背景知识，让我们来看看我们为数据分层构建的不同解决方案。

方案 1：使用 Apache Kafka 和 Flink 进行流式传输

我们的第一个方案遵循了一种熟悉的流式处理模式，使用 Apache Kafka 和 Flink，将数据实时流式传输到 S3，同时将其持久化到 Cassandra。

高效地将数据写入 S3

为了优化到 S3 的数据流式传输，系统对 Time Series ID 进行哈希分区，将其分入不同的桶。这会对流进行分区，并让专用的 Flink 任务处理每个桶，最终通过批处理降低写入 S3 的吞吐量。Flink 工作进程将事件批处理为Parquet文件，按小时内的的事件时间分组。这些文件非常适合高吞吐的流式写入，但不适合低延迟读取。

于是有了压缩作业。

用于优化读取的压缩作业

压缩作业每天运行，将小的 Parquet 文件合并为每个命名空间一个经过优化的每日 Parquet 文件。

该作业通过以下方式进一步优化 Parquet 文件：

• 小行组：创建小（4–8 MB）行组，以最大限度地降低查找成本。然后我们可以使用Parquet Java 库从 S3 对行组执行范围读取。这有意远低于Parquet 推荐的行组下限——更小的行组更好，因为我们的读取以点/范围查找为主，而非全量扫描。
• 事件排序：按时间对行组内的事件进行排序，在查询时间范围时消除了暴力扫描。
• 布隆过滤器：在EVCache中缓存布隆过滤器和文件元数据，使 TimeSeries 查询规划器能够高效地跳过不必要的 S3 读取。

一旦压缩作业完成，每个分区的所有已压缩文件都会记录在 TimeSeries 控制平面中，此时热存储切片就可以被截断了！

在我们一些最大的数据集上，此作业通常需要约 12 小时，优先考虑较低的计算成本，而非执行速度。

流式冷层读取

一旦压缩文件在冷存储中完成优化，它们就可供读取。TimeSeries 数据平面将给定的查询扇出到热层和冷层，并合并它们的结果。用户仍然使用相同的查询界面：

冷层数据平面使用压缩元数据进行积极的规划和剪枝，这大幅减少了 S3 查询扇出。因此，检索一整天数据最多需要三次顺序 S3 读取，即使是在 ID 跨越两个行组这种不常见的情况下也是如此。这带来了每次读取约 150 毫秒的 p50 延迟和约 500 毫秒的 p99 延迟。

在撰写本文时，流式冷层实现已在全球范围内每秒处理多达 1,250 次从冷存储的读取：

为什么不就此止步？

流式冷层带来了实实在在的节省，既缩小了 Cassandra 的EBS占用，也缩小了 S3 中 Cassandra 备份的占用。但流式基础设施本身成本高昂，并增加了许多活动部件，扩大了团队的值班范围：

我们希望进一步改进这一点，于是问自己：

• 我们能否避免将数据流式传输到对象存储以及相关的数据传输成本？
• 我们能否降低新组件的计算和运维开销？

这引出了一个更为激进的问题：我们能否直接从 S3 中现有的 Cassandra 备份读取数据？

方案 2：面向不可变数据的 Cassandra 冷存储

在 Netflix，我们定期将 Cassandra 数据备份到 S3 以用于灾难恢复。分层存储的思路很简单：

直接从这些 S3 备份读取，而不是为传输到 S3 支付单独的数据传输成本。

一旦 TimeSeries 将某个热时间切片标记为可供冷存储，我们就可以定制 Cassandra 以直接从其备份读取数据，从而删除磁盘上的数据副本。这同时还能将对 TimeSeries 读取路径的改动降到最低！

为了使其在生产环境中可行，我们必须回答一系列“如何……？”的问题。

我们如何高效地读取备份？

我们在 Cassandra 中引入了一种新的压缩策略，我们称之为 ReadOnlyCompactionStrategy（ROCS）。其工作原理如下：

一旦某个时间切片越过用户配置的 冷层边界，TimeSeries 就停止接受该时间切片的写入，使其变为不可变。这种不可变性带来了显著的优化机会：

• SSTable 重叠： 标准策略（LCS、TWCS）通常会产生在 token 空间上重叠的 SSTable，因此单个分区键可能存在于多个文件中。
• 读放大： 由于一次读取必须触及覆盖该键 token 范围的每个 SSTable，这种重叠将单次点读变成多次查找，从而增加所有层级的延迟。

ROCS 会重新压缩所有 SSTable 一次，使其与 token 范围精确对齐：

ROCS 更广泛地适用于所有 TimeSeries 用例，通过减少读操作期间的 SSTable 扇出，降低所有存储层级的尾延迟：

一旦不可变时间切片得到优化，就该触发备份了。

我们将如何利用备份？

Cassandra 会备份集群中的每个节点，将每个备份在 S3 上组织为一个声明式清单，以及支持该键空间各表的 SSTable：

 {bucket}/
  └── {prefix}/                             ← cluster identity (env_backup/hash_app)
      └── {token}/                          ← node identity (ring token)
          ├── META_V2/                      ← manifests, one per snapshot
          │   ├── {snapshotInstantMs}.json
          │   ├── {snapshotInstantMs}.json
          │   └── ...
          └── SST_V2/                       ← actual SSTable files
              └── {backupTimestamp}/        ← snapshot timestamp
                  └── {keyspace}/
                      └── {table-with-uuid}/ 
                          └── {compression}/
                              └── {encryption}/
                                  ├── {component-file}
                                  └── ...

该 备份清单 编排备份，列出其所有 SSTable 的路径：

manifest.json
  ├── info
  │   ├── appName              "cass-cluster"
  │   ├── region               "us-east-1"
  │   ├── snapshotInstantMs    1748908800000
  │   ├── backupPathPrefix     "prod_backup/1234_cass/token"
  └── data[]
      └── [0]
          ├── keyspaceName     "ks"
          ├── columnfamily     "tbl-a1b2..."
          └── sstables[]
              └── [0]
                  └── components[]
                      ├── na-1-big-Data.db
                      ├── na-1-big-Index.db
                      ├── na-1-big-Filter.db
                      ├── na-1-big-Statistics.db
                      ├── na-1-big-Summary.db
                      ├── na-1-big-CompressionInfo.db

除了大型 Data.db 文件外，备份还包含一组紧凑的索引、元数据和完整性文件。这种结构使我们能够快速定位、验证和恢复特定分区或数据范围，而无需扫描整个数据集。此外，由于清单编目了每个 S3 路径，我们可以将其中大部分组件（不包括 Data.db 文件）直接下载到 Cassandra 节点，它们占用的空间可以忽略不计。这些本地组件支持高效的查询规划，而无需从 S3 读取任何数据。

Cassandra 将如何从备份读取？

Cassandra 读取通常从两个来源获取数据：Memtable，它保存最近的未刷写写入，以及持久存储上已刷写的 SSTable。从 Cassandra 4.1 起，我们可以利用 可插拔 Memtable 将“冷”表的读取 直接重定向到 S3 中的备份存储。

-- create table for a given Time Slice
CREATE TABLE data_20260616_00_04 (
    ...
    ...,
    PRIMARY KEY ((time_series_id, time_bucket, event_bucket), event_time, event_id, key)
) WITH CLUSTERING ORDER BY (event_time DESC, event_id DESC, key DESC)
    ...
    AND compaction = ...
    AND compression = ...
    AND memtable = 'backupmemtable:timestamp=1781953200000' ← Custom memtable
    ...

该时间戳标记了备份保证覆盖全表的时点，且该表保证是不可变的。

打开一个备份 Memtable 会启动一个状态机初始化，它会一次性将索引数据从 S3 拉取到节点上：

此过程通过多种弹性机制确保所有 SSTable 组件都能可靠下载：

• 重试与退避： 所有 S3 操作和文件系统缓存写入都使用带指数退避的重试。
• 资源隔离： 初始化在专用线程池上运行，与正常 Cassandra 操作隔离。
• 部分下载保护： 如果下载的组件大小与清单不匹配，则将其删除并重新下载。
• DDL 变更保护： 一个 .timestamp 文件标记初始化完成，并将其与备份时间戳绑定。如果该表的备份时间戳发生变化，此文件将被删除，并为新时间戳重新开始初始化。
• 读取保护： Memtable 初始化是异步的；对正在初始化的 Memtable 的读取最多阻塞读取 RPC 超时的一半。

一旦初始化从 S3 拉取了所有 SSTable 组件（Data.db 除外）到本地磁盘，Memtable 就准备好提供读取服务。读取过程如下所示：

为了最小化延迟，系统在请求、SSTable 和单个数据块之间并行化 S3 读取。共享缓存会去重获取并预取数据，因此数据块在需要之前就已就绪。

我们进一步优化集群和表设置以减少 S3 调用次数：

• 更大的压缩块： 通过读取更少、更大的块来减少每个分区的 S3 调用（在小读取上接受一定的过度获取）。
• 禁用读取修复： 避免针对不可变、已预先修复的备份数据产生昂贵且不必要的修复流量。
• 禁用推测性重试： 防止 S3 较高的延迟触发冗余、重叠的读取。

我们如何知道何时可以安全地从备份读取？

然而，在 Memtable 能够指向备份之前，我们必须在整个环上确认两件事，我们使用 Cassandra 虚拟表 来做到这一点。

首先，ROCS 重新压缩成功：

cqlsh> SELECT * FROM netflix_views.cluster_view                                                                                                                                                          
       WHERE keyspace_name = 'system_views' AND table_name = 'sstable_tasks';

keyspace_name | table_name    | host     | value                                                                                                                                                        
--------------+---------------+----------+------------------------------------------------
system_views  | sstable_tasks | 10.0.0.1 | [{"table_name":"<.>","kind":"compaction", ...}]
system_views  | sstable_tasks | 10.0.0.2 | [] <---- no active compactions
system_views  | sstable_tasks | 10.0.0.3 | []

其次，备份已完全捕获（由 uploaded=True 表示）：

cqlsh> SELECT * FROM netflix_views.complete_backups                                                                                                                                                      
       WHERE keyspace_name = 'my_ks' AND table_name = 'my_table';

keyspace_name | table_name | timestamp     | num_tokens | num_uploaded | uploaded                                                                                                                       
--------------+------------+---------------+------------+--------------+----------
my_ks         | my_table   | 1765211400000 |          3 |            3 |     True
my_ks         | my_table   | 1765213200000 |          3 |            2 |    False

我们如何自动化整个流程？

TimeSeries 控制平面在切片越过不可变边界时自动完成整个转换：

每个切片在任何时候都可以处于任何生命周期阶段，而不会影响系统可用性，并且系统保持 对延迟备份或滞后压缩具有弹性。

Cassandra 将如何处理额外负载？

从 S3 提供数据比从本地存储读取要消耗多得多的 CPU。读取请求的突然激增可能会压垮 Cassandra 集群并影响热路径，t将一个节省成本的功能变成可靠性风险。

为避免这种情况，冷读可以选择运行在一个独立的 Cassandra 集群上，该集群将其备份 Memtable 指向热集群的备份：

该冷 Cassandra 集群在拓扑上必须与原始集群一致，但在硬件上无需匹配！ 我们可以在小得多、更便宜、存储更少的实例上运行冷集群。

真实世界性能

Cassandra 冷层已推广到存储最密集的 TimeSeries Abstraction 集群。它目前在 Cassandra 冷存储中存储了超过 15 PB 的压缩数据，并使我们能够大幅削减所有受影响集群的附加存储：

同时，它在任何给定区域每秒从冷层存储提供数千次读取：

凭借更高效的 S3 访问模式和更小的读取大小，Cassandra 冷层相比之前的冷存储实现将 p90 延迟改善了约 30%，同时将 p99 延迟保持在约 500 毫秒以下。Cassandra 已通过其块缓存优雅地处理冷读突发，因此该设计为基于 S3 的冷层保留了同样的韧性：

致谢：特别感谢我们出色的同事，他们为 TimeSeries Abstraction 的成功做出了贡献，并与我们合作开发了冷存储解决方案：Kaidan Fullerton、Matt Lehman、Sumanth Pasupuleti、Carlos Matias Herrero、Nick Pavlakis、Bryan Keller、Francesco Di Chiara

TimeSeries 分层存储之旅：从 Kafka/Flink 流到原生 Cassandra 冷读 最初发表于 Netflix TechBlog（Medium），人们正在那里通过高亮和回应这个故事来继续对话。