专栏首页CSDN技术头条整合Kafka到Spark Streaming——代码示例和挑战

整合Kafka到Spark Streaming——代码示例和挑战

作者Michael G. Noll是瑞士的一位工程师和研究员,效力于Verisign,是Verisign实验室的大规模数据分析基础设施(基础Hadoop)的技术主管。本文,Michael详细的演示了如何将Kafka整合到Spark Streaming中。期间,Michael还提到了将Kafka整合到Spark Streaming中的一些现状,非常值得阅读,虽然有一些信息在Spark 1.2版本中已发生了一些变化,比如HA策略:通过Spark Contributor、Spark布道者陈超我们了解到,在Spark 1.2版本中,Spark Streaming开始支持fully HA模式(选择使用),通过添加一层WAL(Write Ahead Log),每次收到数据后都会存在HDFS上,从而避免了以前版本中的数据丢失情况,但是不可避免的造成了一定的开销,需要开发者自行衡量。

以下为译文

作为一个实时大数据处理工具,Spark Sreaming近日一直被广泛关注,与Apache Storm的对比也经常出现。但是依我说,缺少与Kafka整合,任何实时大数据处理工具都是不完整的,因此我将一个示例Spark Streaming应用程序添加到kafka-storm-starter,并且示范如何从Kafka读取,以及如何写入到Kafka。在这个过程中,我还使用Avro作为数据格式,以及Twitter Bijection进行数据序列化。

在本篇文章,我将详细地讲解这个Spark Streaming示例;同时,我还会穿插当下Spark Streaming与Kafka整合的一些焦点话题。免责声明:这是我首次试验Spark Streaming,仅作为参考。

当下,这个Spark Streaming示例被上传到GitHub,下载访问:kafka-storm-starter。项目的名称或许会让你产生某些误解,不过,不要在意这些细节:)

什么是Spark Streaming

Spark Streaming是Apache Spark的一个子项目。Spark是个类似于Apache Hadoop的开源批处理平台,而Spark Streaming则是个实时处理工具,运行在Spark引擎之上。

Spark Streaming vs. Apache Storm

Spark Streaming与Apache Storm有一些相似之处,后者是当下最流行的大数据处理平台。前不久,雅虎的Bobby Evans和Tom Graves曾发表过一个“Spark and Storm at Yahoo!”的演讲,在这个演讲中,他们对比了两个大平台,并提供了一些选择参考。类似的,Hortonworks的P.Taylor Goetz也分享过名为Apache Storm and Spark Streaming Compared的讲义。

这里,我也提供了一个非常简短的对比:对比Spark Streaming,Storm的产业采用更高,生产环境应用也更稳定。但是从另一方面来说,对比Storm,Spark拥有更清晰、等级更高的API,因此Spark使用起来也更加愉快,最起码是在使用Scala编写Spark应用程序的情况(毫无疑问,我更喜欢Spark中的API)。但是,请别这么直接的相信我的话,多看看上面的演讲和讲义。

不管是Spark还是Storm,它们都是Apache的顶级项目,当下许多大数据平台提供商也已经开始整合这两个框架(或者其中一个)到其商业产品中,比如Hortonworks就同时整合了Spark和Storm,而Cloudera也整合了Spark。

附录:Spark中的Machines、cores、executors、tasks和receivers

本文的后续部分将讲述许多Spark和Kafka中的parallelism问题,因此,你需要掌握一些Spark中的术语以弄懂这些环节。

  • 一个Spark集群必然包含了1个以上的工者作节点,又称为从主机(为了简化架构,这里我们先抛弃开集群管理者不谈)。
  • 一个工作者节点可以运行一个以上的executor
  • Executor是一个用于应用程序或者工作者节点的进程,它们负责处理tasks,并将数据保存到内存或者磁盘中。每个应用程序都有属于自己的executors,一个executor则包含了一定数量的cores(也被称为slots)来运行分配给它的任务。
  • Task是一个工作单元,它将被传送给executor。也就是说,task将是你应用程序的计算内容(或者是一部分)。SparkContext将把这些tasks发送到executors进行执行。每个task都会占用父executor中的一个core(slot)。
  • Receiver(API,文档)将作为一个长期运行的task跑在一个executor上。每个receiver都会负责一个所谓的input DStream(比如从Kafka中读取的一个输入流),同时每个receiver(input DStream)占用一个core/slot。
  • input DStream:input DStream是DStream的一个类型,它负责将Spark Streaming连接到外部的数据源,用于读取数据。对于每个外部数据源(比如Kafka)你都需要配置一个input DStream。一个Spark Streaming会通过一个input DStream与一个外部数据源进行连接,任何后续的DStream都会建立标准的DStreams。

在Spark的执行模型,每个应用程序都会获得自己的executors,它们会支撑应用程序的整个流程,并以多线程的方式运行1个以上的tasks,这种隔离途径非常类似Storm的执行模型。一旦引入类似YARN或者Mesos这样的集群管理器,整个架构将会变得异常复杂,因此这里将不会引入。你可以通过Spark文档中的Cluster Overview了解更多细节。

整合Kafka到Spark Streaming

概述

简而言之,Spark是支持Kafka的,但是这里存在许多不完善的地方。

Spark代码库中的KafkaWordCount对于我们来说是个非常好的起点,但是这里仍然存在一些开放式问题。

特别是我想了解如何去做:

  • 从kafaka中并行读入。在Kafka,一个话题(topic)可以有N个分区。理想的情况下,我们希望在多个分区上并行读取。这也是Kafka spout in Storm的工作。
  • 从一个Spark Streaming应用程序向Kafka写入,同样,我们需要并行执行。

在完成这些操作时,我同样碰到了Spark Streaming和/或Kafka中一些已知的问题,这些问题大部分都已经在Spark mailing list中列出。在下面,我将详细总结Kafka集成到Spark的现状以及一些常见问题。

Kafka中的话题、分区(partitions)和parallelism

详情可以查看我之前的博文:Apache Kafka 0.8 Training Deck and Tutorial和Running a Multi-Broker Apache Kafka 0.8 Cluster on a Single Node。

Kafka将数据存储在话题中,每个话题都包含了一些可配置数量的分区。话题的分区数量对于性能来说非常重要,而这个值一般是消费者parallelism的最大数量:如果一个话题拥有N个分区,那么你的应用程序最大程度上只能进行N个线程的并行,最起码在使用Kafka内置Scala/Java消费者API时是这样的。

与其说应用程序,不如说Kafka术语中的消费者群(consumer group)。消费者群,通过你选择的字符串识别,它是逻辑消费者应用程序集群范围的识别符。同一个消费者群中的所有消费者将分担从一个指定Kafka话题中的读取任务,同时,同一个消费组中所有消费者从话题中读取的线程数最大值即是N(等同于分区的数量),多余的线程将会闲置。

多个不同的Kafka消费者群可以并行的运行:毫无疑问,对同一个Kafka话题,你可以运行多个独立的逻辑消费者应用程序。这里,每个逻辑应用程序都会运行自己的消费者线程,使用一个唯一的消费者群id。而每个应用程序通常可以使用不同的read parallelisms(见下文)。当在下文我描述不同的方式配置read parallelisms时,我指的是如何完成这些逻辑消费者应用程序中的一个设置。

这里有一些简单的例子

  • 你的应用程序使用“terran”消费者群id对一个名为“zerg.hydra”的kafka话题进行读取,这个话题拥有10个分区。如果你的消费者应用程序只配置一个线程对这个话题进行读取,那么这个线程将从10个分区中进行读取。
  • 同上,但是这次你会配置5个线程,那么每个线程都会从2个分区中进行读取。
  • 同上,这次你会配置10个线程,那么每个线程都会负责1个分区的读取。
  • 同上,但是这次你会配置多达14个线程。那么这14个线程中的10个将平分10个分区的读取工作,剩下的4个将会被闲置。

这里我们不妨看一下现实应用中的复杂性——Kafka中的再平衡事件。在Kafka中,再平衡是个生命周期事件(lifecycle event),在消费者加入或者离开消费者群时都会触发再平衡事件。这里我们不会进行详述,更多再平衡详情可参见我的Kafka training deck一文。

你的应用程序使用消费者群id“terran”,并且从1个线程开始,这个线程将从10个分区中进行读取。在运行时,你逐渐将线程从1个提升到14个。也就是说,在同一个消费者群中,parallelism突然发生了变化。毫无疑问,这将造成Kafka中的再平衡。一旦在平衡结束,你的14个线程中将有10个线程平分10个分区的读取工作,剩余的4个将会被闲置。因此如你想象的一样,初始线程以后只会读取一个分区中的内容,将不会再读取其他分区中的数据。

现在,我们终于对话题、分区有了一定的理解,而分区的数量将作为从Kafka读取时parallelism的上限。但是对于一个应用程序来说,这种机制会产生一个什么样的影响,比如一个Spark Streaming job或者 Storm topology从Kafka中读取数据作为输入。

1. Read parallelism:通常情况下,你期望使用N个线程并行读取Kafka话题中的N个分区。同时,鉴于数据的体积,你期望这些线程跨不同的NIC,也就是跨不同的主机。在Storm中,这可以通过TopologyBuilder#setSpout()设置Kafka spout的parallelism为N来实现。在Spark中,你则需要做更多的事情,在下文我将详述如何实现这一点。

2. Downstream processing parallelism:一旦使用Kafka,你希望对数据进行并行处理。鉴于你的用例,这种等级的parallelism必然与read parallelism有所区别。如果你的用例是计算密集型的,举个例子,对比读取线程,你期望拥有更多的处理线程;这可以通过从多个读取线程shuffling或者网路“fanning out”数据到处理线程实现。因此,你通过增长网络通信、序列化开销等将访问交付给更多的cores。在Storm中,你通过shuffle grouping将Kafka spout shuffling到下游的bolt中。在Spark中,你需要通过DStreams上的repartition转换来实现。

通常情况下,大家都渴望去耦从Kafka的parallelisms读取,并立即处理读取来的数据。在下一节,我将详述使用Spark Streaming从Kafka中的读取和写入。

从Kafka中读取

Spark Streaming中的Read parallelism

类似Kafka,Read parallelism中也有分区的概念。了解Kafka的per-topic话题与RDDs in Spark中的分区没有关联非常重要。

Spark Streaming中的KafkaInputDStream(又称为Kafka连接器)使用了Kafka的高等级消费者API,这意味着在Spark中为Kafka设置read parallelism将拥有两个控制按钮。

1. Input DStreams的数量。因为Spark在每个Input DStreams都会运行一个receiver(=task),这就意味着使用多个input DStreams将跨多个节点并行进行读取操作,因此,这里寄希望于多主机和NICs。

2. Input DStreams上的消费者线程数量。这里,相同的receiver(=task)将运行多个读取线程。这也就是说,读取操作在每个core/machine/NIC上将并行的进行。

在实际情况中,第一个选择显然更是大家期望的。

为什么会这样?首先以及最重要的,从Kafka中读取通常情况下会受到网络/NIC限制,也就是说,在同一个主机上你运行多个线程不会增加读的吞吐量。另一方面来讲,虽然不经常,但是有时候从Kafka中读取也会遭遇CPU瓶颈。其次,如果你选择第二个选项,多个读取线程在将数据推送到blocks时会出现锁竞争(在block生产者实例上,BlockGenerator的“+=”方法真正使用的是“synchronized”方式)。

input DStreams建立的RDDs分区数量:KafkaInputDStream将储存从Kafka中读取的每个信息到Blocks。从我的理解上,一个新的Block由spark.streaming.blockInterval在毫秒级别建立,而每个block都会转换成RDD的一个分区,最终由DStream建立。如果我的这种假设成立,那么由KafkaInputDStream建立的RDDs分区数量由batchInterval/spark.streaming.blockInterval决定,而batchInterval则是数据流拆分成batches的时间间隔,它可以通过StreamingContext的一个构造函数参数设置。举个例子,如果你的批时间价格是2秒(默认情况下),而block的时间间隔是200毫秒(默认情况),那么你的RDD将包含10个分区。如果有错误的话,可以提醒我。

选项1:控制input DStreams的数量

下面这个例子可以从Spark Streaming Programming Guide中获得:

在这个例子中,我们建立了5个input DStreams,因此从Kafka中读取的工作将分担到5个核心上,寄希望于5个主机/NICs(之所以说是寄希望于,因为我也不确定Spark Streaming task布局策略是否会将receivers投放到多个主机上)。所有Input Streams都是“terran”消费者群的一部分,而Kafka将保证topic的所有数据可以同时对这5个input DSreams可用。换句话说,这种“collaborating”input DStreams设置可以工作是基于消费者群的行为是由Kafka API提供,通过KafkaInputDStream完成。

在这个例子中,我没有提到每个input DSream会建立多少个线程。在这里,线程的数量可以通过KafkaUtils.createStream方法的参数设置(同时,input topic的数量也可以通过这个方法的参数指定)。在下一节中,我们将通过实际操作展示。

但是在开始之前,在这个步骤我先解释几个Spark Streaming中常见的几个问题,其中有些因为当下Spark中存在的一些限制引起,另一方面则是由于当下Kafka input DSreams的一些设置造成:

当你使用我上文介绍的多输入流途径,而这些消费者都是属于同一个消费者群,它们会给消费者指定负责的分区。这样一来则可能导致syncpartitionrebalance的失败,系统中真正工作的消费者可能只会有几个。为了解决这个问题,你可以把再均衡尝试设置的非常高,从而获得它的帮助。

然后,你将会碰到另一个坑——如果你的receiver宕机(OOM,亦或是硬件故障),你将停止从Kafka接收消息。

Spark用户讨论markmail.org/message/…

这里,我们需要对“停止从Kafka中接收”问题做一些解释。当下,当你通过ssc.start()开启你的streams应用程序后,处理会开始并一直进行,即使是输入数据源(比如Kafka)变得不可用。也就是说,流不能检测出是否与上游数据源失去链接,因此也不会对丢失做出任何反应,举个例子来说也就是重连或者结束执行。类似的,如果你丢失这个数据源的一个receiver,那么你的流应用程序可能就会生成一些空的RDDs。

这是一个非常糟糕的情况。最简单也是最粗糙的方法就是,在与上游数据源断开连接或者一个receiver失败时,重启你的流应用程 序。但是,这种解决方案可能并不会产生实际效果,即使你的应用程序需要将Kafka配置选项auto.offset.reset设置到最小——因为Spark Streaming中一些已知的bug,可能导致你的流应用程序发生一些你意想不到的问题,在下文Spark Streaming中常见问题一节我们将详细的进行介绍。

选择2:控制每个input DStream上小发着线程的数量

在这个例子中,我们将建立一个单一的input DStream,它将运行3个消费者线程——在同一个receiver/task,因此是在同一个core/machine/NIC上对Kafka topic“zerg.hydra”进行读取。

KafkaUtils.createStream方法被重载,因此这里有一些不同方法的特征。在这里,我们会选择Scala派生以获得最佳的控制。

结合选项1和选项2

下面是一个更完整的示例,结合了上述两种技术:

我们建立了5个input DStreams,它们每个都会运行一个消费者线程。如果“zerg.hydra”topic拥有5个分区(或者更少),那么这将是进行并行读取的最佳途径,如果你在意系统最大吞吐量的话。

Spark Streaming中的并行Downstream处理

在之前的章节中,我们覆盖了从Kafka的并行化读取,那么我们就可以在Spark中进行并行化处理。那么这里,你必须弄清楚Spark本身是如何进行并行化处理的。类似Kafka,Spark将parallelism设置的与(RDD)分区数量有关,通过在每个RDD分区上运行task进行。在有些文档中,分区仍然被称为“slices”。

在任何Spark应用程序中,一旦某个Spark Streaming应用程序接收到输入数据,其他处理都与非streaming应用程序相同。也就是说,与普通的Spark数据流应用程序一样,在Spark Streaming应用程序中,你将使用相同的工具和模式。更多详情可见Level of Parallelism in Data Processing文档。

因此,我们同样将获得两个控制手段:

1. input DStreams的数量,也就是说,我们在之前章节中read parallelism的数量作为结果。这是我们的立足点,这样一来,我们在下一个步骤中既可以保持原样,也可以进行修改。

2. DStream转化的重分配。这里将获得一个全新的DStream,其parallelism等级可能增加、减少,或者保持原样。在DStream中每个返回的RDD都有指定的N个分区。DStream由一系列的RDD组成,DStream.repartition则是通过RDD.repartition实现。接下来将对RDD中的所有数据做随机的reshuffles,然后建立或多或少的分区,并进行平衡。同时,数据会在所有网络中进行shuffles。换句话说,DStream.repartition非常类似Storm中的shuffle grouping。

因此,repartition是从processing parallelism解耦read parallelism的主要途径。在这里,我们可以设置processing tasks的数量,也就是说设置处理过程中所有core的数量。间接上,我们同样设置了投入machines/NICs的数量。

一个DStream转换相关是union。这个方法同样在StreamingContext中,它将从多个DStream中返回一个统一的DStream,它将拥有相同的类型和滑动时间。通常情况下,你更愿意用StreamingContext的派生。一个union将返回一个由Union RDD支撑的UnionDStream。Union RDD由RDDs统一后的所有分区组成,也就是说,如果10个分区都联合了3个RDDs,那么你的联合RDD实例将包含30个分区。换句话说,union会将多个DStreams压缩到一个DStreams或者RDD中,但是需要注意的是,这里的parallelism并不会发生改变。你是否使用union依赖于你的用例是否需要从所有Kafka分区进行“in one place”信息获取决定,因此这里大部分都是基于语义需求决定。举个例子,当你需要执行一个不用元素上的(全局)计数。

注意:RDDs是无序的。因此,当你union RDDs时,那么结果RDD同样不会拥有一个很好的序列。如果你需要在RDD中进行sort。

你的用例将决定需要使用的方法,以及你需要使用哪个。如果你的用例是CPU密集型的,你希望对zerg.hydra topic进行5 read parallelism读取。也就是说,每个消费者进程使用5个receiver,但是却可以将processing parallelism提升到20。

在下一节中,我将把所有部分结合到一起,并且联合实际数据处理进行讲解。

写入到Kafka

写入到Kafka需要从foreachRDD输出操作进行:

通用的输出操作者都包含了一个功能(函数),让每个RDD都由Stream生成。这个函数需要将每个RDD中的数据推送到一个外部系统,比如将RDD保存到文件,或者通过网络将它写入到一个数据库。需要注意的是,这里的功能函数将在驱动中执行,同时其中通常会伴随RDD行为,它将会促使流RDDs的计算。

注意:重提“功能函数是在驱动中执行”,也就是Kafka生产者将从驱动中进行,也就是说“功能函数是在驱动中进行评估”。当你使用foreachRDD从驱动中读取Design Patterns时,实际过程将变得更加清晰。

在这里,建议大家去阅读Spark文档中的Design Patterns for using foreachRDD一节,它将详细讲解使用foreachRDD读外部系统中的一些常用推荐模式,以及经常出现的一些陷阱。

在我们这个例子里,我们将按照推荐来重用Kafka生产者实例,通过生产者池跨多个RDDs/batches。我通过Apache Commons Pool实现了这样一个工具,已经上传到GitHub。这个生产者池本身通过broadcast variable提供给tasks。

最终结果看起来如下:

更多的细节和解释可以点击阅读原文看所有源代码。

就我自己而言,我非常喜欢Spark Streaming代码的简洁和表述。在Bobby Evans和Tom Graves讲话中没有提到的是,Storm中这个功能的等价代码是非常繁琐和低等级的:kafka-storm-starter中的KafkaStormSpec会运行一个Stormtopology来执行相同的计算。同时,规范文件本身只有非常少的代码,当然是除下说明语言,它们能更好的帮助理解;同时,需要注意的是,在Storm的Java API中,你不能使用上文Spark Streaming示例中所使用的匿名函数,比如map和foreach步骤。取而代之的是,你必须编写完整的类来获得相同的功能,你可以查看AvroDecoderBolt。这感觉是将Spark的API转换到Java,在这里使用匿名函数是非常痛苦的。

最后,我同样也非常喜欢Spark的说明文档,它非常适合初学者查看,甚至还包含了一些进阶使用。关于Kafka整合到Spark,上文已经基本介绍完成,但是我们仍然需要浏览mailing list和深挖源代码。这里,我不得不说,维护帮助文档的同学做的实在是太棒了。

本文分享自微信公众号 - CSDN技术头条(CSDN_Tech)

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2015-01-05

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Spark App自动化分析和故障诊断

    非常高兴有机会可以代表我们团队在“CCTC 2017——Spark技术峰会”上给大家分享我们在Spark平台化上所做的一些工作,下面是分享的一些笔录。 苏宁大...

    CSDN技术头条
  • 取代而非补充,Spark Summit 2014精彩回顾

    Apache Spark开源生态系统在2014上半年大幅增长,已迅速成为大数据领域中最活跃的开源项目,HDFS位列第二,其代码变动次数(commits)和行数仅...

    CSDN技术头条
  • Apache Spark的承诺及所面临的挑战

    Spark并非完美无瑕,目前发展到了什么程度呢?我们来一起看看Spark的优劣之处吧。 可以读一读Panopoly带来的The Evolution of the...

    CSDN技术头条
  • Tensorflow的C语言接口部署DeeplabV3+语义分割模型

    tensorflow框架一般都是基于Python调用,但是有些时候跟应用场景,我们希望调用tensorflow C语言的接口,在C++的应用开发中使用它。要这么...

    OpenCV学堂
  • [python][基础]判断变量是否可

    py3study
  • linux升级python到2.7

    py3study
  • 利用Metasploit 打入ThinkPHP内网...

    (1)通过msfconsole命令启动Metasploit在这个过程中,系统会主动加载数据库,如图2所示:

    徐焱
  • 走进Java接口测试之解决超大文本数据驱动报OOM问题

    上篇文章 走进Java接口测试之测试框架TestNG数据驱动(入门篇)阐述测试框架 TestNG 中的一些基本的概念和玩法,本文带着大家来解决一个实际的工程问题...

    高楼Zee
  • 分布式系统事务一致性

    现今互联网界,分布式系统和微服务架构盛行。业界著名的CAP理论也告诉我们,在设计和实现一个分布式系统时,需要将数据一致性、系统可用性和分区容忍性放在一起考虑。

    黄泽杰
  • 【答疑释惑】Android多国语言实现原理

    先来看一下网友的问题: ? 网友们问的问题是android中布局文件中android:text="hello_world"和 android:text="@s...

    程序员互动联盟

扫码关注云+社区

领取腾讯云代金券