你好我是程序员贵哥。在上节课里我们已经了解了Kafka的基本架构。不过对于基于Kafka的流式数据处理我们还有两个重要的问题没有回答第一个Kafka的分布式是如何实现的呢我们已经看到了Kafka会对数据进行分区以进行水平扩展。那么如果我们可以动态添加Broker来增加Kafka集群的吞吐量这个集群的上下游是怎么知道的呢第二个在我们有了Kafka和Storm这样的系统之后我们的流式处理系统应该怎么搭建呢我们如何解决可能遇到的各种故障带来的数据不准确的问题呢那么今天这节课就是要帮助我们回答这两个问题。一方面今天我们会深入来看一下Kafka是如何随着Broker的增加和减少协调上下游的Producer和Consumer去收发消息的。另一方面我们会从整个大数据系统的全局视角来看一下在有了Kafka和Storm这样的利器之后我们的大数据系统的整体架构应该如何搭建。Kafka的分布式系统的实现首先Kafka系统并没有一个Master节点。不过这一点倒是不让人意外主要是Kafka的整体架构实在太简单了。我们在上一讲就看到了单个的Broker节点底层就是一堆顺序读写的文件。而要能够分布式地分摊压力只需要用好ZooKeeper就好了。每一个Kafka的Broker启动的时候就会把自己注册到ZooKeeper上注册信息自然是Broker的主机名和端口。在ZooKeeper上Kafka还会记录这个Broker里包含了哪些主题Topic和哪些分区Partition。而ZooKeeper本身提供的接口则和我们之前讲解过的Chubby类似是一个分布式锁。每一个Kafka的Broker都会把自己的信息像一个文件一样写在一个ZooKeeper的目录下。另外ZooKeeper本身也提供了一个监听-通知的机制。上游的Producer只需要监听Brokers的目录就能知道下游有哪些Broker。那么无论是随机发送还是根据消息中的某些字段进行分区上游都可以很容易地把消息发送到某一个Broker里。当然Producer也可以无需关心ZooKeeper而是直接把消息发送给一个负载均衡由它去向下游的Broker进行数据分发。高可用机制而在Kafka最初的论文里还没有包括Kafka的高可用机制。在这种情况下一旦某个Broker节点挂了它就会从ZooKeeper上消失对应的分区也就不见了自然数据我们也就没有办法访问了。不过在了解了这么多的分布式高可用方案之后相信我们要自己实现一个Kafka的高可用方案自然也不困难。在Kafka发布了0.8版本之后它就支持了由多副本带来的高可用功能。在现实中Kafka是这么做的首先为了让Kafka能够高可用我们需要对于每一个分区都有多个副本和GFS一样Kafka的默认参数选择了3个副本。其次这些副本中有一个副本是Leader其余的副本是Follower。我们的Producer写入数据的时候只需要往Leader写入就好了。Leader自然也就是将对应的数据写入到本地的日志文件里。然后每一个Follower都会从Leader去拉取最新的数据一旦Follower拉到数据之后会向Leader发送一个Ack的消息。我们可以设定有多少个Follower成功拉取数据之后就能认为Producer写入完成了。这个可以通过在发送的消息里设定一个acks的字段来决定。如果acks0那就是Producer的消息发送到Broker之后不管数据是否刷新到本地硬盘我们都认为写入已经完成了而如果设定acks2意味着除了Leader之外至少还有一个Follower也把数据写入完成并且返回Leader一个Ack消息之后消息才写入完成。我们可以通过调整acks这个参数来在数据的可用性和性能之间取得一个平衡。Producer发送数据、Broker接收并存储下来的逻辑就这么简单。不过下游Consumer去消费数据的逻辑就稍微复杂一点了。主要的挑战来自于我们可以动态增减Broker和Consumer。负载均衡机制Kafka的Consumer一样会把自己“注册”到ZooKeeper上。在同一个Consumer Group下一个Partition只会被一个Consumer消费这个Partition和Consumer的映射关系也会被记录在ZooKeeper里。这部分信息被称之为“所有权注册表”。而Consumer会不断处理Partition的数据一旦某一段的数据被处理完了对应这个Partition被处理到了哪个Offset的位置也会被记录到ZooKeeper上。这样即使我们的Consumer挂掉由别的Consumer来接手后续的消息处理它也可以知道从哪里做起。那么在这个机制下一旦我们针对Broker或者Consumer进行增减Kafka就会做一次数据“再平衡Rebalance”。所谓再平衡就是把分区重新按照Consumer的数量进行分配确保下游的负载是平均的。Kafka的算法也非常简单就是每当有Broker或者Consumer的数量发生变化的时候会再平均分配一次。如果我们有X个分区和Y个Consumer那么Kafka会计算出 NX/Y然后把0到N-1的分区分配给第一个ConsumerN到2N-1的分配给第二个Consumer依此类推。而因为之前Partition的数据处理到了哪个Offset是有记录的所以新的Consumer很容易就能知道从哪里开始处理消息。而和Storm一样本质上Kafka对于消息的处理也是“至少一次”的。如果消息成功处理完了那么我们会通过更新ZooKeeper上记录的Offset来确认这一点。而如果在消息处理的过程中Consumer出现了任何故障我们都需要从上一个Offset重新开始处理。这样我们自然也就避免不了重复处理消息。如果你希望能够避免这一点你需要在实际的消息体内有类似message-id这样的字段并且要通过其他的去重机制来解决但是这并不容易做到。顺序保障机制不过Kafka虽然有很强的性能也在发布之后很快提供了基于多副本的高可用机制。但是Kafka本身其实也是有很多限制的。首先是Kafka很难提供针对单条消息的事务机制。因为我们在ZooKeeper上保存的是最新处理完的消息的一个Offset而不是哪些消息被处理完了、哪些消息没有被处理完这样的message-id status的映射关系。所以Consumer没法说我有一条新消息已经处理完了但是还有一条旧消息还在处理中。而是只能按照消息在Partition中的偏移量来顺序处理。其次是Kafka里对于消息是没有严格的“顺序”定义的。也就是我们无法保障先从应用服务器发送出来的消息会先被处理。因为下游是一个分布式的集群所以先发送的消息X可能被负载均衡发送到Broker A后发送的消息反而被负载均衡发送到Broker B。但是Broker B里的数据可能会被下游的Consumer先处理而Broker A里的数据后被处理。不过对于快速统计实时的搜索点击率这样的统计分析类的需求来说这些问题都不是问题。而Kafka的应用场景也主要在这里而不是用来作为传统的消息队列完成业务系统之间的异步通信。数据处理的Lambda架构其实有了Storm和Kafka这样的实时数据处理架构之后另一个问题也就浮出了水面。既然我们已经可以获得分钟级别的统计数据那我们还需要MapReduce这样的批处理程序吗答案当然还是需要的因为在目前的框架下我们的流式计算还有几个问题没有处理好。首先是我们的流式数据处理只能保障“至少一次At Least Once”的数据处理模式而在批处理下我们做到的是“正好一次Exactly Once”。也就意味着批处理计算出来的数据是准确的而流式处理计算的结果是有误差的。其次是当数据处理程序需要做修改的时候批处理程序很容易修改而流式处理程序则没有那么容易。比如增加一些数据分析的维度和指标。原先我们只计算点击率现在可能还需要计算转化率原先我们只需要有分国家的统计数据现在还要有分省份和分城市的数据。我们原先的计算结果已经保存在数据库或者HDFS上了。那么对于批处理程序来说我们的解决方案也很容易那就是选定一个我们希望新的报表需要覆盖的时间范围比如过去30天。我们撰写一个新的MapReduce程序运行出新的计算结果保存成新的数据表。我们可以把旧的数据表删除用新的数据表替换就好了。通常我们的Hadoop集群不只要承担报表任务也会承担很多临时的分析任务。所以一般来说像Hadoop这样的批处理集群的计算资源对于单个报表来说是足够富余的重跑30天的数据分析往往也可以在1~2天内完成。流式数据处理的性能压力但是对于流式处理问题就有些麻烦了特别是在没有Kafka的时候。我们重新撰写一个新的Storm的Topology来支持新的分析维度和指标并不困难。困难的地方在于我们需要在不影响正在线上运行的程序的情况下进行新版本程序的发布。一个解决方案是我们写了一个新的Topology然后需要重放Replay过去30天的日志数据。如果我们用的是Scribe这样的流式日志传输系统我们会发现日志流都已经上传到了HDFS上我们还需要有一个程序从HDFS上把数据拉出来发送到Scribe里。而即使你用了Kafka数据都存放在了Kafka Broker的本地硬盘上重放日志的动作你还是少不了的。这个时候你会面临的问题是重放日志需要花费很多时间、或者短时间内会消耗很多计算资源。一般来说你最多也就为流式数据处理预留平时日志量35倍的计算能力。那如果你需要重放30天的日志你就需要等上610天才能重放完。这样就意味着每次修改程序要么你只能更新新的数据产生的报表要么你就要等上好几天才能看到最后的计算结果。其实最常会发生的变更既不来自于硬件故障导致的数据重复处理也不是来自于业务需求变更导致我们需要修改程序。**最常会发生的变更来自于解决分析程序里的各种Bug。**在这种场景下我们的输入数据不会发生变化输出的表结构也不会发生变化。但是我们可能需要反复修改数据处理程序并且反复在同一份日志数据集上运行这个程序。这样的程序运行场景对于大数据的批处理来说压力并不大但是对于流式数据处理一样会有大量重放日志的工作量。Lambda架构的基本思想有鉴于此Storm的作者南森·马茨Nathan Marz提出了Lamda架构把大数据的批处理和实时数据结合在一起变成一个统一的架构。Nathan的思路是这样的我们先不去看具体数据是通过什么计算框架来处理的而是把整个数据处理流程抽象成View Query(Data)这样一个函数。我们所有的报表、看板、数据分析结果都是来自对于原始日志的分析。所以原始日志就是我们的主数据Master Data不管是MapReduce这样的大数据批处理还是Storm这样的大数据流式处理都是对于原始数据的一次查询Query。而这些计算结果其实就是一个基于特定查询的视图View。当我们的程序有Bug其实就是查询写错了我们的主数据没有变我们视图的含义也没有变我们只需要重新写一个查询就好了。而如果我们有需求层面的变更就是我们需要一个新的视图以及对应的新的查询了。而对于实际数据分析系统的用户来说其实他关心的既不是Query也不是Master Data而是一个个View。那么我们在系统的整体架构上就只需要对这些用户暴露出View而不需要告诉他们具体下面的Query和Master Data的细节就好了。这样我们可以按照Hadoop和Storm本身合适的场景进行选择。一方面我们可以通过Storm进行实时的数据处理能够尽快获得想要的报表和数据分析结果。另一方面我们同样会定时运行MapReduce程序获得更准确的数据结果。在MapReduce程序运行完之前我们的分析决策基于Storm的实时计算结果但是当MapReduce更准确的计算结果出来了我们就可以拿这个结果替换掉之前的实时计算结果。而对于外部用户来说他们看到的始终是同一个视图只是这个视图会随着时间的变化不断修正数据结果罢了。所以Nathan Marz总结的Lambda结构是由这样几部分组成的第一部分是输入数据也就是Master Data这部分也就是我们的原始日志。然后是一个批处理层Batch Layer和一个实时处理层Speed Layer它们分别是一系列MapReduce的任务和一组Storm的Topology获取相同的输入数据然后各自计算出自己的计算结果。最后是一个服务层Serving Layer通常体现为一个数据库。批处理层的计算结果和实时处理层的结果会写入到这个数据库里。并且后生成的批处理层的结果会不断替代掉实时处理层的计算结果也就是对于最终计算的数据进行修正。对于外部的用户来说他不需要和批处理层以及实时处理层打交道而只需要通过像SQL这样的查询语言直接去查询服务层就好了。数据处理的Kappa架构可以看到Lambda架构很好地结合了MapReduce和Storm的优点。而这个Lambda结构最终也变成了Twitter的一个开源项目SummingBird。但是这个Lambda架构也有一个显著的缺点也就是什么事情都需要做两遍。这个做两遍体现在两个方面首先所有的视图既需要在实时处理层计算一次又要在批处理层计算一次。即使我们没有修改任何程序也需要双倍的计算资源。其次我们所有的数据处理的程序也要撰写两遍。MapReduce的程序和Storm的程序虽然要计算的是同样的视图但是因为底层的框架完全不同代码我们就需要写两套。这样意味着我们需要双倍的开发资源。而且因为批处理层和实时处理层的代码不同我们还不得不解决两遍对于同样视图的理解不同采用了不同的数据处理逻辑引入新的Bug的问题。不过在Kafka还没有成熟的时候把数据分成批处理层和实时处理层是很难避免的。主要问题在于我们重放实时处理层的日志是个开销很大的动作。通过Scribe这样的日志收集器我们的Master Data最终是以一个个固定文件落地到HDFS的文件系统上。一旦我们想要重放日志我们就需要把日志从HDFS上分片拉到不同的服务器上再搭建起多个Scribe的集群去重放日志。但是在有了Kafka之后重放日志一下子变得简单了。因为我们所有的日志都会在Kafka集群的本地硬盘上。而通过重放日志来重新进行数据计算也只是设定一下新的分析程序在ZooKeeper上的Offset就好了。有鉴于此Kafka的作者杰伊·克雷普斯Jay Kreps就提出了一个新的数据计算框架称之为Kappa架构。Kappa架构在View Query(Data)这个基本的抽象理念上和Lambda架构没有变化。但是相比于Lambda架构Kappa架构去掉了Lambda架构的批处理层而是在实时处理层支持了多个视图版本。我们之所以要有View Query(Data)这么一个抽象是因为我们的原始日志也就是Data是不会变化的而我们想要的View也不会变化。但是具体的Query可能会因为程序有Bug而比较频繁地被修改。在Kappa架构下如果要对Query进行修改我们原来的实时处理层的代码可以先不用动而是可以先部署一个新版本的代码比如一个新的Topology上去。然后我们会对这个Topology进行对应日志的重放在服务层生成一份新的数据结果表也就是视图的一个新的版本。在日志重放完成之前外部的用户仍然会查询旧的实时处理层产生的结果。而一旦日志重放完成新的Topology能够赶上进度处理到最新产生的日志那么我们就可以让查询切换到新的视图版本上来旧的实时处理层的代码也就可以停掉了。而随着Kappa架构的提出大数据处理又开始迈入了一个新的阶段也就是“流批一体”逐步进入主流的阶段。而这个也是我们接下来的课程中要探讨的主题了。小结好了到这里我们对于Kafka、Lambda架构以及Kappa架构也就学习完了。在今天这节课里我们看到Kafka的分布式架构其实非常简单。Kafka本身没有Master每一个Broker节点都会把自己注册到ZooKeeper上。所有的Broker本身也不维护任何状态对应的状态信息也是放在ZooKeeper上而下游的Consumer也是一样。对应Consumer处理数据到哪里了就是简单地在ZooKeeper上为每一个分区维护了一个最新的Offset。而对应的数据分区分配给哪一个Consumer也是通过ZooKeeper里的“所有权注册表”记录下来的分配的逻辑也非常简单也就是按照分区和Consumer的数量均匀地根据ID顺序分配。而有了Storm和Kafka之后工程界开始思考如何将数据的批处理和流式处理统一起来。Storm的作者Nathan Marz提出了一个抽象概念那就是 View Query(Data)。我们的数据处理程序只是针对数据的一个抽象函数本身没有状态这个函数运行在一个主数据集上可以拿到一个对应的结果视图。所以基于这个概念他把自己设计的架构称之为Lambda架构。在Lambda架构里我们的数据处理程序被分成批处理层、实时处理层以及服务层。批处理层的结果会不断替换掉实时处理层的计算结果以不断给出更准确的视图。而外部的应用只会查询服务层并不需要关心底层的数据处理的实现是怎么样的。不过在Lambda架构下我们的数据需要处理两遍我们的程序代码也要在不同的计算框架下实现两遍。为了减少这样的双倍开销Jay Kreps提出了Kappa架构。Kappa架构利用了Kafka把日志数据保留在Broker本地硬盘重放非常容易这样一个特点提出了放弃批处理层转而在实时处理层提供多个程序版本的思路。而这个思路也会是接下来几年里大数据处理进一步进化的主要方向。