本文将主要介绍如何利用Lambda架构来跟踪数据实时更新的项目实现，以一个新闻服务功能为例。

当前股票市场的交易者可以了解丰富的股票交易信息。从金融新闻到传统的报纸和杂志再到博客和社交媒体，汇聚着海量的数据，远比股票交易者想关注的股 票信息要大得多，这就需要为股票交易者提供信息的有效过滤。这里将开发一个新闻服务功能给股票证券投资交易者使用，并为股票交易者提供个性化新闻。

这个新闻服务就叫"自动获取金融新闻"，输入各个数据源的金融新闻，也同时输入用户实时股票交易信息。不管何时，在股票交易者所拥有资产证券中占比 较大的公司，它们的新闻一到达，将会显示到股票交易者的仪表板上。随着大量股票交易者进行交易，相应的交易信息会发送过来，所以希望拥有一个大数据系统来 存储所有交易者的历史交易信息作为真实数据源，然而，处理海量数据会非常慢以至于不能进行实时的数据更新。为了达到实时跟踪和维持数据结果为最新这两个要求，可以采用Lambda架构来实现。

Lambda架构优势

在传统SQL系统，更新一个表只是对已存在字段的值进行更改，这在少量的服务器上的数据库工作的很好，可以水平扩展到从库或者备份库。但是当数据库 扩展到大量数据服务器上时，硬件崩溃等情况下恢复数据到失败点就比较困难和耗时，而且由于历史不在数据库中，仅仅存在log日志，数据崩溃将导致一些不可见的数据错误，即脏数据。

而相对应地，一个分布式、多副本消息队列的大数据系统可以保证数据一旦进入系统就不会丢失，即使在硬件或者网络失败的情况下。存储更新的所有历史可 以重建真实的数据源，并能保证每次批处理之后结果正确，然而，为了在实时数据更新后得到最新完整的数据集，需要重新处理整个历史数据集，将会耗费太长的时 间。为了解决这个问题，可以在Lambda架构中增加一个实时组件，此组件只存储数据更新的当前值，可以保证快速实时得到结果，工作过程类似于传统的 SQL系统。实时处理层的脏数据将会被后续批处理覆盖掉，这个高可用、最终一致性的系统可以实现准确的结果。当前值的任何错误，实时处理层的报告，硬件或 者网络错误，数据崩溃，或者软件Bug等将会在下一次批处理时自动修复。

自动获取金融新闻项目的数据管道

整个数据管道流动如图1：

图1

输入数据格式为JSON，主要来自综合交易信息和Twitter新闻。JSON格式的消息会push到Kafka，并被批处理层(batch layer)和实时处理层(real-time layer)消费。使用Kafka作为数据管道的输入起点，是因为Kafka可以保证即使在硬件或者网络失败的情况下，消息也会被传输到整个系统。

在批处理层，Camus(Linkin开源的项目，现已更名为Gobblin)消费所有Kafka过来的消息并保存到HDFS上，然后Spark处理所有的交易历史计算每个股票交易者持有的股票准确数量，对应的结果会写入Cassandra数据库。

在流式处理层，Spark Streaming实时消费Kafka消息，但并不像Storm那样完全实时，Spark Streaming可以达到500ms的micro-batch数据流处理。Spark Streaming可以重用批处理层的Spark代码，并且micro-batch数据流处理可以得到足够小的延迟。

批处理层和实时处理层的结果都会写入到Cassandra数据库，并通过Flask提供一个web接口服务。随着海量交易数据写入系统，Cassandra数据库的快速写入能力基本可以满足。

如何调度实时处理层和批处理层的结果

当最新的消息进入大数据系统，web接口提供的结果服务总能保持最新，综合批处理层和实时层的处理结果。用一个例子来展示如何简单的使用批处理结果和实时处理结果。

从下图2看到，有三个数据库表：一个存储批处理结果(图2中Batch表);一个存储自上次批处理完成时间点到当前时间的实时交易数据，即增量数据(图2中Real Time 2表);另外一个存储最新数据，即状态表(图2中高亮的Real Time 1表)。

任何软件、硬件或者网络问题引起批处理结果异常，都通过单独一个数据库表记录数据增量，并在批处理成功后更新为对应的批处理结果数来保证最终数据一致性。

在这个例子中，假设第一轮批处理起始时间点为t0，一个交易者做了一笔交易后获得了3M公司的5000股股票。

图2

在t0时间点，批处理开始，处理完之后最新结果存储在Real Time 1表，当前值为5000股。

图3

在批处理过程中，交易者卖掉3M公司1000股股票，Real Time 1表更新数据值为4000股，同时Real Time 2表存储从t0到当前的增量-1000股，如图4所示。

图4

当批处理结束，三个表的值分别为5000，4000，-1000。这时，交换active数据库表为Real Time 2表，进行合并批处理结果和实时结果获得最新结果值。然后重置Real Time 1表为0，后续用来存储从t1时间点开始的增量数据。接下来新的一轮以存储最新数据的Real Time 2表为起点，循环前面的过程。

图5

图6

图7

以上每步处理过程完全成功并写入数据库，可以保证展示给交易者的数据准确性。数据集 处理时间取决于数据集大小，处理任务的计划按序处理而不是按自然天时间。在一个系统中需要工作流支持复杂处理、多任务依赖和资源共享。这里采用 Airbnb的项目Airflow，可以调度程序和监控工作流。Airflow把task和上游各种依赖构建成一个有向无环图(DAG)，基于 Python实现，可以把多个任务写成Bash脚本，Bash命令能直接调用任何模块，并且Bash脚本可以被Airflow使用，这样使得 Airflow易操作。Airflow编程接口比基于XML配置的调度系统Oozie简单;Airflow的Bash脚本编码量比Luigi要少很多，Luigi的每个job都是一个python工程。每步合并实时和批量数据的job运行都是前一步成功完成退出后。

最后简单总结一下，Lambda架构涉及批量处理层和实时处理层处理历史数据以及实时更新的数据。 为了Lambda架构的实现切实可行，数据处理要设计成批处理层和实时处理层结合。本项目中，有一个“备用”数据库表专门用来存储输入的总数，而不从批处 理层读取数据，并允许对批处理层和实时处理层的结果进行简单的聚合。以上就是用Lambda架构实现的一个高可用、高数据最终一致性的系统。

作者：侠天

来源：51CTO