Design Data-Intensive Application 笔记

TODO: 目前并没能为这个笔记理出一个主线，书的最后一章的总结其实比较到位，这里也是在最终读到最后一章时，尝试按其脉络进行的总结，目前仍是一团散沙。

本书的一条主线就是考虑如何设计组织的数据流，充分利用各类数据/存储系统的优势，让让数据以合适的形态存储在合适的地方，使它们构成有机的整体，构建可靠（reliable）、可扩展（scalable）和可维护（maintainable）的应用或系统。

很大一部分业务需求可以归结为“存储一些数据，后续将它们查询出来”。对于足够简单的系统，其所有的需求都能通过单一的数据库完美解决，可能充分利用数据库特性就足够了。但现实中，随着需求的不断迭代，我们的应用复杂度会逐渐变大，比如新业务不断产生、要处理的历史数据不断累加、组织架构划分导致本身耦合的相关领域需要由不同团队队列维护，某个时候，我们可能会发现，单一的中心数据库不再能很好契合现状；另一个现状是目前并不存在一个完美的数据软件产品能处理好所有的处理场景，我们需要选择合适的软件产品去做合适的事情，如：

• OLTP 数据库用于处理各类业务事务
• 全文检索用于支持任意关键字搜索
• 数据缓存的维护
• 数据仓库、批处理、流处理用于各类数据分析，进行机器学习、分类、排名、推荐
• 基于数据进行通知的发送

此书的目标是探索如何创建可靠（reliable）、可扩展（scalable）和可维护（maintainable）的应用和系统。

• 可靠性：容错算法
• 可扩展性：分区（partitioning）
• 可维护：各类演化、抽象机制

系统存在多个存储系统时，需要搞清楚数据流的方向，每个存储的输入和输出。

跨系统的事务读写需要借助相关分布式事务进行处理。或者考虑让所有用户输入流入单一系统对写操作进行定序（全序广播），子系统可以直接按序处理这些写操作派生所需数据。

• 分布式事务使用锁来确保写顺序。CDC 和 event sourcing 使用日志进行写定序。
• 分布式事务使用 atomic commit 保证操作执行 exactly once，基于日志的系统基于执行的确定性和幂等重试来保证这点
• 但是，分布式事务一般能提供 linearizability，这能提供 reading your own write. Derived data system 一般异步更新数据，因此通常不能提供 timing guarantee.

或者可以通过 log-based event based 系统去达成跨系统是事务希望达成的保证。

Total order 对于存在因果关系（causality）的事件是至关重要的。因果依赖的产生方式可能极其微妙：因果关系

例如对于社交软件，社交关系和消息发送之间若设计初未给定这两者中事件的因果关系，可能会导致一个用户看见了被解除关系用户发送的消息。

例如，应用数据写入特定记录库，通过 CDC (Change Data Capture) 进行索引的更新。

如果允许应用直接对记录库和索引库进行更新，可以会产生并发写问题。

流式处理和批处理主要的区别是流处理器处理的是无限的的数据集，而批处理处理的是有限、已知的数据。但是它们的处理引擎直接的差别正在弥合。

• Spark 通过将 stream 分割成 microbatch 进行流处理
• Apache Flink 通过流处理引擎进行批处理

异步 derived data 和事务相比

• 异步可以让系统中本地错误留在本地，当然，代价是需要考虑维护各子系统流程的一致性
• 事务则会导致所有参与子系统同时出错

存在两大类消息中间件

• AMQP/JMS 风格的 broker: 它将独立消息分派给 consumer，消息在 ack 后被删除，适用于异步 RPC，例如任务队列；其中任务的执行顺序并不重要，且无读取处理过任务的需求
• Log-based message broker: 特定分区消息全分派给给定 consumer，消息以相同顺序发送；Broker 将持久化消息，可以后续重放历史消息。这种方式和数据库中的 replication log 比较类似。

可以渐近的演化系统:

• Lambda Architecture
• BigData: Principles and Best Practices of Scalable Real-Time Data Systems. 497

Lambda Architecture 存在一些实践上的问题

• 需要为批处理和流处理维护相同的逻辑，将引入不少工作量
• 流处理和批处理两个流水线产生的结果独立，需要合并后以响应用户请求
  • 如果是简单对于 tumbling window 的聚合还比较容易
  • 对于复杂操作，如 join，sessionization 或者输出不是 time series 的场景，比较难实现
• 直接对完整历史数据进行批处理将消耗过多资源，不太实际，一般会做成进行增量的批处理，这会面临处理窗口的问题（关于时间）

该架构的问题在通过 batch 和 stream 处理的弥合中逐渐被消弭：

• 通过处理近期事件流的处理引擎对历史事件进行重放；Log-based 消息 broker 支持消息重放
• 流处理器提供 exactly-once (effectively-once) 语义；容错
• 通过事件时间而不是处理时间划分窗口

最抽象的角度来说，它们都提供了数据存储、处理、查询的能力。

• Unix 简单在是对于硬件资源的薄封装
• 数据库 简单在可以使用简单的声明式查询驱动强大的基础架构（查询优化，索引，join，并发控制，replication 等）得到所需数据

数据库提供的特性包括：

• 二级索引：根据特定字段进行高效搜索
• Materialized view: 可以认为是特定查询预计算出的缓存
• Replication log: 保证各个节点之间的同步
• 全文索引：允许通过文本中关键字进行搜索

一个组织中的整个数据流很像一个大型数据库，其中批处理、流处理、ETL 进程将数据从一个地方移动到另一个地方，从一个形式转换成另一种形式；和数据库中保持索引或 materialized view 保持最新的机制一样。

批、流处理器可以类比于数据库中 trigger、存储过程、materialized view 维护例行程序；由它们维护的 derived data 可以看成各类索引。例如，关系数据库支持 B-tree 索引、hash 索引、spatial 索引等。

如果认同不存在单一的数据模型满足所有访问场景的话，可能存在两条不同存储、处理工具组成有机系统的道路：

• Federated database: unifying reads
  • 写仍然通过访问各类底层引擎实现，对查询进行统一
  • 但是如何同步写需要自行考虑
• Unbundled¹ databases: unifying writes
  • 单个数据库中，一致性的索引的维护是内建功能；存在多个系统时，可靠的组合存储系统如同把数据库的索引维护特性拆分，让它能够自动同步不同存储直接的写。
  • 这种方式和 unix 传统类似，让每个小工具做一件事，把一件事做好，然后让它们之间可以通过统一的接口交互。

Unifying write 方式相对复杂，一般有两种方式

1. 使用跨异构存储系统的分布式事务，这种方式可能过于复杂，更好的方式是
2. 使用异步 eveng log 加幂等写入，这种方式更加健壮与实际

Stream 处理器中使用分布式事务实现 exactly-once 效果挺好，但是对于来自不同团队的不同系统，缺乏一个标准的事务协议，让它们之间的事务协调很难集成。而使用 ordered event log 加幂等 consumer 是个更简单的抽象，因此在异构的存储系统中实现起来更为可行。

1. 异步事件流让整个系统在面对单个组件性能降级时更为健壮；Consumer 运行慢或者异常时，producer 可以继续不受影响的执行，消息能被缓存，当消费者恢复后，可以正常继续处理所有未处理的消息而不会丢失任何数据；若使用分布式事务进行同步交互，本地异常可能升级成更大规模的异常
2. 拆分的数据系统允许软件组件、服务可以独立的开发、改进和维护。团队可以专注于将一件事做好，为其它团队提供良好定义的接口。同时 event log 提供足够强大的接口，能用于实现相对强一致的系统属性（通过持久化和排序的事件），同时它也足够通用，方便任意系统的对接。

使用应用代码对专用存储和处理系统进行组合以达到分拆数据库的方式也被称为 "database inside-out" 方式。

如果一份数据集派生自另一数据集，一般会经过一些转换函数

• 次级索引派生自相对直接的转换函数：对于每行、文档，提取需要索引的列、字段，对它们进行排序（假设使用 B-tree 或者 SSTable 进行索引）
• 全文搜索索引一般通过各类自然语言处理函数（如语言检测（language detection），分词（word segmentation）， stemming 或 lemmatization，spelling correction，synonym identification）处理后构建高效查询的数据结构（如 inverted index）
• 机器学习系统中，可以认为模型是使用各种 feature extraction、统计分析函数从训练集派生得出。当输入新数据到模型，得出的数据可以认为从输入和模型（也即间接的从训练集）派生得到。
• 缓存通常包含为了 UI 展示所需的聚合数据。对于缓存的构建通常需要 UI 相关的领域知识，UI 的改变可能会影响到缓存的构建定义。

派生数据的维护与异步任务执行并不一样，传统消息系统一般是为异步任务执行设计的：

• 维护派生数据时，状态改变的顺序通常比较重要
• 容错对于派生数据至关重要，丢失单一消息可能导致整个数据集损坏

严格的排序要求和容错处理能力是相对严格的要求，但是相比分布式事务仍然成本相对较低。

流处理和服务的对比

• 目前微服务化的趋势是将传统的应用切分成一组通过同步网络请求调用（如 REST 接口）组合的服务；这类面向服务的架构（soa, service-oriented architecture）相对传统单体的优势主要在于通过低耦合实现的组织的可扩展性：不同团队可以专注于不同的服务；服务可以独立部署、更新，降低了不同团队之间的沟通成本。
• 将 stream operator 组合进数据流系统与微服务有很多相似之处，不过底层的数据通讯机制大为不同，它通过单向、异步的消息流交互，而非同步、请求/响应方式。

为何我们需要派生数据？一般是需要后续进行查询。

• Write path: 特定数据写入系统后，更新派生数据（如搜索引擎、预测模型等）
• Read path: 为响应用户请求，从派生数据中读取数据，可能再做一些处理后构造给用户的响应

Write path 和 read path 在派生数据集衔接，这里存在写和读操作时工作量的权衡。

一般来说，我们会考虑将写操作放入存储，记录进 event log，而把读当作对于特定存储数据的瞬态请求。但是如果我们将读请求也作为事件传入系统，然后订阅其处理结果，对于读请求的处理将和流处理器对于派生数据处理完美融合；同时我们还能更好的捕获事件之间的因果关系，不过问题是这将引入更大的 I/O 成本。

我们希望构建可靠、正确的（程序的语义定义良好，可被理解；可能发生的异常也能被定义）程序。事务，因其提供的 atomicy, isolation 和 durability 属性保证，一直是构建正确应用的首选工具，在数据流架构中如何构建正确应用？

End-to-end argument: 应用层需要达成的数据属性保障只能通过应用层实现，下层基础设施提供的各类保障只能作为其实现手段；也即，数据库事务可能为我们提供了 durability，应用层不会因为使用了它就能让应用数据自动达成了 durability，比如你的 BUG 误删了数据。

像唯一性这种约束需要共识，

• 达成共识最简单的方式是让单一 leader 进行决策
• 对于唯一性检测，一种 scalable 的方式是使用希望保持唯一的值进行分区；但这种方式和异步、多 master 复制方式不相容，多 master 可能并发地收到冲突的写操作，导致唯一性约束被破坏。如果需要立即拒绝破坏唯一性约束的写请求，同步协同不可避免。

Total order broadcast 的事件日志等价于共识要求，也可以用于确保唯一性约束。流处理器对于同一分区的日志按序处理，如果分区是使用要维护唯一性的值进行分区的，流处理器将可以确定性、无歧义的确定两个冲突的请求哪个先到。除了唯一性约束，对于可能发生冲突被路由到同一分区的场景，这种方式都能适用。

enforcing constraint

一致性（consistency）可认为由两个属性构成

• Timeliness: 确保用户看到的系统状态是最新状态；CAP 中的一致性需要满足 linearizability，这是一种相对较强的属性要求，它能达成 timeliness；较弱的 timeliness 属性、如 read-after-write 一致性有些场景也足够
• Integrity: 表示不存在数据损坏。例如，数据库索引需要正确反应数据库里的数据，缺失了部分记录的索引用途不大。违反 integrity 导致的不一致将是永久的，这种异常一般需要显式地检查与修复。在 ACID 事务中，一致性通常被理解成某种应用级别对于 integrity 的表示。Atomicity 和 durability 是维持 integrity 的重要工具。

ACID 事务一般同时提供 timeliness (如 linearizability) 和一致性（如 atomic commit）保证。

对于基于事件的数据流系统，其很多有趣的属性来自对于 timeliness 和 integrity 的解耦。若事件流的处理是异步的，除非显式的构建一个消费者监听处理结束的消息，这类系统一般不提供 timeliness 保证。而 integrity 则是流处理系统的核心。

Exactly-once 或 effectively-once 语义是实现 integrity 的一种机制。

可以通过一组机制实现 integrity:

• 将写操作表示为一条独立的、可以 atomically 写入的消息
• 使用确定的 derivation 函数，从消息派生出所有状态变更
• 在消息处理的各个层级，传递客户端生成的请求 ID，实现 end-to-end 的去重和幂等
• 消息设计为不可变的；允许派生数据能重复处理，减轻 bug 修复后的处理负担

Uniqueness 约束算是一个全局、较强的约束，很多时候可以换成弱一点的约束，很多业务场景下，事实上可以接受约束被短暂的违反：

• Compensating transaction.
• 可能依赖人、业务设计进行冲突处理（如过多的卖出票可以道歉退票、或升舱之类方式处理）

可以发现

• 数据流系统可以不依赖 atomic commit, linearizability 或 synchronous cross-partition coordination 达成 integrity 要求
• 严格的 uniqueness 约束需要 timeliness 和 coordination，但是大部分应用其实可以容忍这种约束被短暂打破、后续修复

在讨论诸如 correctness, integrity, 容错时，我们会做出某些事情可能会出错、某些事情不会出错的假设，我们称这些假设为我们的 system model.

通过事件日志派生数据通常可以做成确定、幂等的，使得错误恢复也很容易。

• 大部分情况，构造 totally ordered log 需要所有事件流经一个单一 leader 节点进行定序，若吞吐量高于单机处理能力，需要进行分区，不同分区事件的顺序可能存在歧义
• 若服务器跨不同地区的数据中心，一般希望某个数据中心出现问题，其它数据中心仍能支撑系统继续工作，这意味着不同数据中心产生的事件顺序不定
• 微服务应用中，常见设计是不同服务不共享持久化状态，因此不同服务产生的事件顺序不定
• 特定应用维护客户端状态，且在离线状态下仍能工作，这种类型应用其客户端和服务端见到事件顺序也可能产生不一致

对事件进行全序定序被称为 total order broadcast，它等价于 consensus. 大部分共识算法被设计成单个节点足够处理系统的事件吞吐，它们并未给出通过多节点进行事件定序的机制。设计一个跨地区分布的、能扩展到超越单机负载能力的共识算法仍是个开放问题。

Causal order 不是 total order: 有些事件的顺序可比，有些不可比。

• Logical timestamp 可以在不借助 coordination 的情况下提供 total ordering, 可以在无法使用 order broadcast 的情况下提供一些用处；但是它们需要接收人根据元信息对乱序消息进行处理
• 系统将用户做出决定前对于系统的观测记录为一个事件，后续事件可以引用该事件 ID 以维护因果关系
• 冲突解决算法可以用于处理那些意外排序的事件，对于维护状态有用，但是对于触发了外部副作用的操作无能为力

Extended Architecture Transaction (XA Transaction).

不同数据系统数据的一致性需要通过分布式事务确保。如使用 2PC。

因果关系（causality）导致事件存在顺序：因产生果，消息发送才能被接收，提出问题后给出答案。

1. 数据进行分区、复制时，如果某些分区的复制速度较慢，观察者可能观察到错误的因果关系（比如对于一个应答过程，先在特定分区看到答案，然后再看见问题）
2. 因果关系可以使用 happens before 关系描述：如果 A happens before B，表示 B 可能知道 A 或它建立于 A 之上，或者它依赖 A；如果 A 和 B 是并发的，它们之间不存在因果关系，换句话，我们确定它们对彼此毫无感知
3. 在 snapshot isolation 中的一致性表示的是因果关系上的一致：即如果 snapshot 中包含“答案”，其中也必须包含“问题”。能够观测数据库特定时间的状态蕴含着因果一致性：所有给定时间点之前的操作产生的效果可见，所有该时间点之后操作产生的效果不可见。Read skew 表示读取的数据违反了因果关系
4. 因果关系还可能通过观测引入，两个进行并发对同一份数据进行观测，依据观测结果做出决定最终修改了被观测的数据导致决定的前提被改变，也就是 write screw 的场景；在 serializable snapshot isolation 中，可以通过追踪事务之间的因果依赖检测 write screw
5. Cross-channel timing dependencies: 考虑 Fiture 9-5 中 image resize 功能的实现，引入了 image resizer 对于文件存储服务的依赖，需要文件存储服务的 replicate 功能满足 linearizability 才能百分百确保工作

流处理经常需要处理时间相关问题，比如处于分析目的，经常要使用类似“上 5 分钟平均值”这样的时间窗口。

批处理一般处理历史数据，系统时钟对数据处理意义不大，直接使用时间时间戳去确定事件时间，这使得处理结果可以是确定性的。

很多流处理框架使用处理机器的本地系统时钟决定时间窗口，这种方式简单，且只在处理时间和事件创建时间之间很短的情况下有意义。但当处理时间延迟过大，将可能导致结果存在问题，如

• 消息乱序
• 一种微妙的问题是，如何确定事件窗口，即确定给定逻辑窗口内是否仍然有新的事件
  • 可以进行超时处理，这种情况对于后续过来满足窗口的事件
    1. 忽略这类 straggler 事件
    2. 发布一个修正
  • 使用特殊消息表示“从现在开始将不会有早于时间 t 的消息”，可以用于结束窗口；若存在多个 producer 这仍是个问题

消息延迟可能因各种原因产生：

• 队列
• 网络异常
• Message broker 或者处理器性能问题
• 流 consumer 故障
• 修复问题后重新对历史事件的处理
• …

考虑为事件给定一个时间戳，问题是：应该使用谁的时间？

对于移动应用，向服务器上报使用统计信息，要统计的事件的发生时间是用户操作的发生时间，应该使用用户设备的本地时钟；但是用户控制设备上的时间不可信，一般需要校准，一种方式是记录三个时间戳

• 时间发生时间（用户设备时钟）
• 时间发送到服务器的时间（用户设备时钟）
• 服务器接收事件的事件（服务器时钟）

事件窗口的类型

• Tumbling window: 固定长度、固定边界的时间窗口（如 10:03:00-10:03:59, =10:04:00-10:04:59=）
• Hopping window: 固定长度、固定边界时间窗口，但是重叠（如 10:03:00-10:07:59, =10:04:00-10:08:59=）
• Sliding window: 如最新 5 分钟，其边界是滑动的
• Session window: 按用户在相近时间内同一组行为分割；Sessionization 是网站分析中一种常见需求

对于批处理框架来说，容错相对简单，直接重新执行对应任务，出错的执行的输出直接忽略；这种透明重试的前提假设是输入文件不可变，处理过程确定，然后只保证正确结果对外可见达到的。

498 流处理不能“完全执行完得到结果文件”，因为要处理的流是无限的。

• Microbatching and checkpointing：对于 microbatch 输出结果后，产生副作用这种事无能为力
• Atomic commit: 依赖幂等仰赖几个假设：失败任务的重启需要将所有消息按相同顺序重放；处理过程需要是确定性的，不存在并发对于相同值的更新

确保可以安全重试，而不会产生执行多次的效果

• Distributed transacion
• 幂等

• Differential data flow:
  • http://cidrdb.org/cidr2013/Papers/CIDR13_Paper111.pdf
  • https://www.microsoft.com/en-us/research/
  • https://timelydataflow.github.io/differential-dataflow/chapter_0/chapter_0_0.html