注意： 这一章还未完成编辑，你阅读的内容中还存在大量机器翻译的结果
第3章 日志和实时流处理
到目前为止，我仅描述了什么是从一个地方到另一个地方复制数据的理想方法。但是，在存储系统之间切换字节并不是故事的结局。事实证明，“日志”是“流”的另一个词，而日志是流处理的核心。
但是等等，流处理到底是什么？
如果你喜欢1990年代末和2000年代初的数据库文献或半成功的数据基础结构产品，则可能会将流处理与为事件驱动的处理构建SQL引擎或“框和箭头”接口的工作相关联。
如果你关注开源数据系统的爆炸式增长，则可能会将流处理与该空间中的某些系统相关联，例如Storm，Akka，S4和Samza。大多数人将它们视为一种异步消息处理系统，它与可感知群集的RPC层没有什么不同（实际上，该领域中的某些事情就是这样）。我听说过流处理被描述为一种模型，你可以在其中立即处理所有数据，然后将其丢弃。
这两种观点都有些局限。流处理与SQL无关。它也不限于实时处理。没有内在的原因，你无法使用多种不同的语言来表示昨天或一个月前的数据流。你也不必（或你应该）丢弃捕获的原始数据。
我认为流处理的范围更广：用于连续数据处理的基础结构。我认为计算模型可以像MapReduce或其他分布式处理框架一样通用，但是具有产生低延迟结果的能力。处理模型的真正驱动程序是数据收集的方法。批量收集的数据自然会批量处理。连续收集数据时，自然会对其进行连续处理。
美国人口普查为批量数据收集提供了一个很好的例子。人口普查定期启动，并通过让人们挨家挨户地进行对美国公民的暴力发现和枚举。在1790年首次开始普查时，这很有道理（见图3-1）。当时的数据收集本来就是面向批处理的，因为它涉及骑马骑马并在纸上写下记录，然后将这批记录传输到一个中央位置，在此人员将所有计数加起来。如今，当你描述人口普查过程时，人们立即想知道为什么我们不保留出生和死亡的日志，而是连续地或按需要的粒度来生成人口计数。
这是一个极端的例子，但是许多数据传输过程仍然依赖于进行定期转储以及批量传输和集成。处理批量转储的唯一自然方法是分批处理。随着这些流程被连续供稿所替代，我们自然会开始朝着连续处理迈进，以平滑所需的处理资源并减少延迟。
现代网络公司不需要有任何批量数据收集的。一个网站生成数据是活动数据或数据库的改变，这两者连续地发生。实际上，当你考虑任何业务时，底层机制几乎总是一个连续的过程—正如Jack Bauer告诉我们的那样，事件是实时发生的。当数据以成批收集，它几乎总是由于某些手动步骤或缺乏数字化，或者它是从一些非数字过程的自动化遗留历史遗迹。当机械师需要运送纸片并由人来进行处理时，数据的传输和响应通常非常缓慢。自动化的第一遍始终保持原始过程的形式，因此这通常会在介质更换后很长时间内徘徊。
每天运行的生产“批处理”作业通常有效地模仿了一种每天窗口大小的连续计算。当然，基础数据总是在变化。
这有助于清除有关流处理的一个常见混淆区域。通常认为，某些类型的处理不能在流处理系统中完成，而必须分批完成。我听说过的一个典型示例是计算百分位数，最大值，平均值或其他需要查看所有数据的摘要统计信息。但这在某种程度上使问题变得混乱。的确，例如，对于计算而言，最大值是一项阻塞操作，需要查看窗口中的所有记录才能选择最大记录。这种计算绝对可以在流处理系统中进行。确实，如果你查看有关流处理的最早的学术文献，实际上要做的第一件事就是为窗口提供精确的语义，以便仍然可以对窗口进行阻塞操作。从这个角度来看，很容易分享我对流处理的看法，这种看法更为笼统：与阻塞操作与非阻塞操作无关；只是在处理中的基础数据中包含时间概念的处理，因此不需要静态数据快照即可对其进行操作。这意味着流处理系统以用户控制的频率产生输出，而不是等待数据集的“结尾”。从这个意义上讲，流处理是批处理的概括，在实时数据盛行的情况下，流处理是非常重要的概括。
那么，为什么传统上将流处理视为利基应用呢？我认为最大的原因是缺乏实时数据收集，使得连续处理成为了理论上的关注点。
当然，我认为缺少实时数据收集可能注定了商业流处理系统的失败。他们的客户仍在进行面向文件的ETL和数据集成的每日批处理。建立流处理系统的公司专注于提供连接到实时数据流的处理引擎，但事实证明，当时很少有人真正拥有实时数据流。实际上，在我加入LinkedIn的职业生涯的早期，一家公司曾试图向我们出售一个非常酷的流处理系统，但是由于当时我们的所有数据都是按小时收集的，因此我们唯一想到的就是我们收集了每小时的文件，并在小时结束时将它们输入到流系统中！流处理公司的工程师指出，这是一个相当普遍的问题。例外实际上证明了这里的规则：金融是流处理取得一定成功的一个领域，而这正是实时数据流已经成为规范的领域，而处理数据流则成为紧迫的问题。
即使存在健康的批处理生态系统，流处理作为基础结构样式的实际适用性也很广泛。它涵盖了实时请求/响应服务与脱机批处理之间的基础架构差距。对于现代互联网公司，我认为他们的代码中约有25％属于此类。
事实证明，该日志解决了流处理中一些最关键的技术问题，我将对其进行描述，但是它解决的最大问题是仅在实时多用户数据Feed中提供数据。
对于那些对日志和流处理之间的关系的更多细节感兴趣的人，我们提供了开源的Samza，这是一种明显基于许多想法的流处理系统。我们在文档中更详细地描述了许多这些应用程序。但这并非特定于特定的流处理系统，因为所有主要的流处理系统都与Kafka集成在一起，可以作为数据日志进行处理。
数据流图
流处理最有趣的方面与流处理系统的内部结构无关，而是与它如何扩展我们从较早的数据集成讨论中得出的数据馈送概念有关。我们主要讨论了原始数据的提要或日志-即在执行各种应用程序时直接产生的事件和数据行。流处理使我们还可以包括根据其他提要计算得出的提要。对于用户而言，这些派生的提要看起来与从中计算得出的主要数据的提要没有什么不同（见图3-2）。
这些派生的提要可以封装处理过程中的任意复杂性和智能，因此它们可能非常有价值。例如，Google在流处理系统之上描述了一些有关如何重建其Web爬网，处理和索引管道的细节，该网络必须是地球上最复杂，规模最大的数据处理系统之一。
那么什么是流处理？就我们的目的而言，流处理作业将是从日志读取并将输出写入日志或其他系统的任何内容。用于输入和输出的日志将这些过程连接到处理阶段的图形中。以这种方式使用集中式日志，你可以查看组织的所有数据捕获，转换和流，就像一系列日志以及从中读取和写入的流程一样。
流处理器不一定完全需要一个精美的框架；它可以是从日志读取和写入的任何进程或一组进程。可以提供其他基础结构和支持，以帮助管理和扩展这种近实时处理代码，这就是流处理框架所要做的。
日志与流处理
为什么在流处理中根本需要日志？为什么处理器不使用简单的TCP或其他轻量级消息传递协议进行更直接的通信？有两个很强的理由赞成这种模型。
首先，它使每个数据集成为多用户。每个流处理输入可用于任何需要它的处理器。每个输出都可供需要的任何人使用。这不仅方便了生产数据流，而且还方便了复杂数据处理管道中的调试和监视阶段。能够快速利用输出流并检查其有效性，计算一些监视统计信息，甚至只是看看数据是什么样子，就可以使开发更加容易处理。
日志的第二个用途是确保在每个数据使用者进行的处理中保持顺序。某些事件数据可能仅按时间戳松散排序，但并非所有方法都是这种方式。考虑来自数据库的更新流。我们可能有一系列流处理作业，这些作业将获取这些数据并准备将其编入搜索索引中。如果我们不按顺序处理同一记录的两个更新，则可能会导致搜索索引的最终结果错误。
日志的最终用途可以说是最重要的，那就是为各个进程提供缓冲和隔离。如果处理器产生的输出快于下游消费者所能承受的速度，那么我们有三种选择：
- 我们可以阻止上游作业，直到下游作业赶上（如果仅使用TCP而且没有日志，这可能会发生）。
- 我们可以将数据放在地板上。
- 我们可以在两个进程之间缓冲数据。
删除数据在某些方面是可以的，但是通常是不可接受的，而且从来都不是真正希望的。
起初，听起来听起来像是可以接受的选择，但实际上这是一个巨大的问题。考虑到我们想要的不仅是对单个应用程序的处理进行建模，而且还对整个组织的全部数据流进行建模。不可避免地，这将是由不同团队拥有并以不同SLA运行的处理器之间非常复杂的数据流网络。在这个复杂的数据处理网络中，如果任何使用者失败或无法跟上，上游生产者将阻塞，并且阻塞将在整个数据流图中逐步累积，从而使一切停滞不前。
剩下的唯一实际选择是：缓冲。日志充当非常大的缓冲区，使进程可以重新启动或失败，而不会减慢处理图的其他部分。这意味着消费者可以在不影响任何上游图表的情况下长时间掉下来。只要它能够在重新启动时赶上，其他所有内容都不会受到影响。
这在其他地方并不罕见。大型，复杂的MapReduce工作流程使用文件来检查点并共享其中间结果。大型，复杂的SQL处理管道会创建很多中间表或临时表。这只是将模式与适用于运动中的数据的抽象（即日志）一起应用。
Storm和Samza是以此方式构建的两个流处理系统，可以使用Kafka或其他类似系统作为日志。
重新处理数据：Lambda体系结构和替代方案
这种面向日志的数据建模的有趣应用是Lambda体系结构。这是Nathan Marz提出的一个想法，他在博客中发表了一篇广泛阅读的文章，描述了将流处理与脱机处理相结合的方法（“如何击败CAP定理”）。事实证明，通过专门的网站和即将发行的书籍，这是一个令人惊讶的流行想法。
什么是Lambda架构，我如何成为一体？
Lambda体系结构看起来类似于图3-3。
这种工作方式是捕获不可变的记录序列，并将其并行输入到批处理和流处理系统中。你实施转换逻辑两次，一次在批处理系统中，一次在流处理系统中。然后，你可以在查询时将两者的结果拼接在一起，以得出完整的答案。
这方面有很多变化，我特意简化一下。例如，你可以在Kafka，Storm和Hadoop的各种类似系统中进行交换，并且人们经常使用两个不同的数据库来存储输出表：一个针对实时进行了优化，而另一个针对批量更新进行了优化。
这有什么好处？
Lambda体系结构强调保持原始输入数据不变。我认为这是一个非常重要的方面。查看复杂的数据流的能力很大程度上取决于查看输入的内容和输出的内容的能力。
我也喜欢这种体系结构突出了重新处理数据的问题。重新处理是流处理的主要挑战之一，但经常被忽略。
“重新处理”是指再次处理输入数据以重新得出输出。这是一个完全显而易见但经常被忽略的要求。代码将始终更改。因此，如果你有从输入流中获取输出数据的代码，则每当代码更改时，就需要重新计算输出以查看更改的效果。
为什么代码会更改？可能会有所变化，因为你的应用程序在不断发展，你想计算以前不需要的新输出字段。否则它可能会更改，因为你发现了一个错误并需要对其进行修复。无论如何，你需要重新生成输出。我发现，许多尝试构建实时数据处理系统的人并没有对此问题投入过多的思考，而最终得到了一个由于无法方便地处理再处理而无法快速发展的系统。
坏处……
Lambda体系结构的问题在于，在两个复杂的分布式系统中维护需要产生相同结果的代码，就像看起来那样痛苦。我认为这个问题无法解决。
像Storm和Hadoop这样的分布式框架的编程很复杂。不可避免地，代码最终会针对其运行所在的框架进行专门设计。实现Lambda体系结构的系统所带来的操作复杂性似乎是每一个人都普遍认为的一件事。
解决此问题的一种建议方法是拥有一种在实时和批处理框架上都进行抽象的语言或框架。你使用此高级框架编写代码，然后对其进行“编译”以在后台进行流处理或MapReduce。Summingbird是执行此操作的框架。这肯定会使事情变得更好，但我认为这不能解决问题。最终，即使你可以避免对应用程序进行两次编码，运行和调试两个系统的操作负担也将非常高。任何新的抽象只能提供两个系统的交集所支持的功能。更糟糕的是，致力于这种新的超级框架将丰富的工具和语言生态系统隔离开来，从而使Hadoop如此强大（Hive，Pig，Crunch，Cascading，Oozie等）。
通过类推，考虑使跨数据库对象关系映射（ORM）真正透明的众所周知的困难。并考虑这只是在非常相似的系统上进行抽象的问题，该系统使用（几乎）标准化的接口语言提供了几乎相同的功能。在勉强稳定的分布式系统之上构建完全不同的编程范例的抽象问题要困难得多。
一种替代方案
作为设计基础架构的人，我认为一个明显的问题是：为什么不能仅仅改进流处理系统以真正处理其目标领域中存在的全部问题？为什么需要在另一个系统上粘贴？为什么在代码更改时既不能实时处理又要处理重新处理？流处理系统已经有了并行性的概念，为什么不仅仅通过增加并行性和快速重放历史来处理重新处理呢？答案是你可以做到这一点，如果你今天正在构建这种类型的系统，我认为这实际上是一种合理的替代体系结构。
当我与人们讨论时，他们有时会告诉我，对于高吞吐量的历史数据处理，流处理是不合适的。这是一种直觉，主要是基于他们使用的系统的局限性，即系统扩展性很差或无法保存历史数据。没有任何理由应该如此。流处理中的基本抽象是数据流DAG，它与传统数据仓库（如Volcano）中的基本抽象完全相同，并且是MapReduce后继者Tez中的基本抽象。流处理只是该数据流模型的概括，它向最终用户公开了中间结果和连续输出的检查点。
那么如何直接从流处理工作中进行重新处理呢？我首选的方法实际上非常简单：
- 使用Kafka或其他某些系统，该系统可以让你保留要重新处理的数据的完整日志，并允许多个订户。例如，如果你要重新处理最多30天的数据，请将你在Kafka中的保留时间设置为30天。
- 当你要进行重新处理时，请启动流处理作业的第二个实例，该实例从保留数据的开头开始进行处理，但是将输出数据定向到新的输出表。
- 在完成第二项作业后，切换应用程序以从新表中读取。
- 停止作业的旧版本并删除旧的输出表。
该体系结构类似于图3-4。
与Lambda体系结构不同，在这种方法中，你仅在处理代码更改时才进行重新处理，而实际上你需要重新计算结果。当然，进行重新计算的工作只是在相同框架上运行，采用相同输入数据的相同代码的改进版本。
自然地，你将希望在你的后处理工作上提高并行度，以使其很快完成。
当然，你可以进一步优化它。在许多情况下，你可以合并两个输出表。但是，我认为同时拥有这两种方式有一些好处。这使你仅需具有一个将应用程序重定向到旧表的按钮，即可立即恢复到旧逻辑。在特别重要的情况下（例如，你的广告定位条件），你可以使用自动A / B测试或强盗算法来控制转换，以确保你推出的任何错误修复或代码改进都不会意外降低性能与以前的版本相比。
请注意，这并不意味着你的数据无法发送到Hadoop，而仅意味着你不在那里进行重新处理。Kafka与Hadoop集成良好，因此加载任何Kafka主题都很容易。通常将流处理作业的输出或中间流镜像到Hadoop，以在诸如Hive之类的工具中进行分析，或用作其他脱机数据处理流的输入，通常很有用。
我们已经记录了这种方法的实现，以及使用Samza进行处理架构的其他变化。
两种方法之间的效率和资源折衷在一定程度上是洗礼。Lambda体系结构需要始终运行重新处理和实时处理，而我提出的建议仅在需要重新处理时才运行作业的第二个副本。但是，我的建议需要在输出数据库中暂时拥有两倍的存储空间，并且需要一个支持大容量写入的数据库。在这两种情况下，重新处理的额外负载可能会平均化。如果你有很多这样的工作，它们将不会一次全部重新处理，因此在具有几十个这样的工作的共享群集上，你可能会为在任何给定时间主动进行重新处理的少数工作预算额外的百分之几的容量。
真正的优势根本不在于效率，而在于允许人们在单个处理框架之上开发，测试，调试和操作系统。
因此，在简单性很重要的情况下，除了Lambda体系结构之外，还可以考虑将此方法作为另一种选择。
有状态的实时处理
日志与流处理的关系并不以重新处理而结束。如果流处理系统的实际计算确实需要维护状态，那么对我们的好朋友日志来说，这是另一种用途。
某些实时流处理只是一次无状态记录转换，但是许多用途是更复杂的计数，聚合或流中窗口上的联接。例如，你可能想要用有关执行点击的用户的信息来丰富事件流（例如点击流），实际上是将点击流添加到用户帐户数据库中。始终，这种处理最终需要处理器维护某种状态；例如，当计算一个计数时，到目前为止你需要维护该计数。如果处理器本身可能发生故障，如何正确维护这种状态？
最简单的选择是将状态保留在内存中。但是，如果进程崩溃，它将失去其中间状态。如果仅在窗口上保持状态，则该过程可能会退回到日志中窗口开始的位置。但是，如果要计数一个小时以上，则可能不可行。
一种替代方法是简单地将所有状态存储在远程存储系统中，然后通过网络加入该存储。问题是没有数据的局部性和大量的网络往返。
我们如何支持像表这样根据处理进行分区的东西？
好吧，回想一下关于表和日志双重性的讨论。这给了我们确切的工具，它能够将流转换为与我们的处理并置的表，并提供了处理这些表的容错能力的机制。
流处理器可以将其状态保存在本地表或索引中：bdb，RocksDB甚至是更不寻常的东西，例如Lucene或fastbit索引。该存储库的内容从其输入流中馈送（首先可能是在应用任意转换之后）。它可以记录此本地索引的变更日志，以使其在崩溃时恢复其状态并重新启动。这允许一种通用机制，用于将任意索引类型中的共分区状态保持在传入流数据本地。
当过程失败时，它将从变更日志中还原其索引。日志是将本地状态转换为一次增量记录备份。
这种状态管理方法具有优雅的特性，即处理器的状态也保持为日志。我们可以像考虑对数据库表进行更改的日志一样来考虑该日志。实际上，处理器具有与它们一起维护的非常类似的共分区表。由于此状态本身是日志，因此其他处理器可以订阅它。在处理的目标是更新最终状态（该状态是处理的自然输出）的情况下，这实际上可能非常有用。
当与出于数据库集成目的而从数据库中发出的日志结合使用时，日志/表双重性的功能将变得显而易见。可以从数据库中提取变更日志，并由各种流处理器以不同的形式对索引进行索引，以加入事件流。
我们将在Samza中详细介绍这种管理状态处理的样式，并提供更多实际示例。
日志压缩
当然，我们不能希望为所有时间的状态变化保留完整的日志。除非你要使用无限空间，否则必须以某种方式清除日志。我将在Kafka稍作讨论，以使其更加具体。
在Kafka中，清理有两个选项，具体取决于数据是包含纯事件数据还是键更新。通过事件数据，我的意思是不相关的事件，例如页面视图，单击或你在应用程序日志中可以找到的其他内容。所谓键更新，是指专门记录由某个键标识的实体中状态变化的事件。数据库的变更日志就是一个典型的例子。
对于事件数据，Kafka支持保留数据窗口。可以根据时间（天）或空间（GB）来定义窗口，大多数人只是坚持使用一周的默认保留时间。如果你想无限保留，只需将此窗口设置为无限，就不会丢弃你的数据。
但是，对于键控数据，完整日志的一个不错的属性是你可以重播它以重新创建源系统的状态。也就是说，如果我有更改日志，则可以将该日志重播到另一个数据库的表中，并在任何时间点重新创建表的状态。这也适用于不同的系统：你可以将最初进入数据库的更新日志重播到任何其他通过主键维护数据（搜索索引，本地存储等）的系统中。
但是，随着时间的流逝，保留完整的日志将占用越来越多的空间，并且重放将花费越来越长的时间。因此，在Kafka中，我们支持旨在支持该用例的不同类型的保留，其示例如图3-5所示。我们不是简单地完全丢弃旧日志，而是从日志末尾垃圾收集过时的记录。日志尾部中具有最新更新的任何记录都可以进行这种清除。通过这样做，我们仍然可以保证日志包含源系统的完整备份，但是现在我们不再能够重新创建源系统的所有先前状态，而只能重新创建最近的状态。我们称此功能为日志压缩。
章节列表：
