【编者按】作者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是Apache Spark的一个子项目。Spark是个类似于Apache Hadoop的开源批处理平台，而Spark Streaming则是个实时处理工具，运行在Spark引擎之上。 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和Kafka中的parallelism问题，因此，你需要掌握一些Spark中的术语以弄懂这些环节。什么是Spark Streaming
Spark Streaming vs. Apache Storm
附录：Spark中的Machines、cores、executors、tasks和receivers
在Spark的执行模型，每个应用程序都会获得自己的executors，它们会支撑应用程序的整个流程，并以多线程的方式运行1个以上的tasks，这种隔离途径非常类似Storm的执行模型。一旦引入类似YARN或者Mesos这样的集群管理器，整个架构将会变得异常复杂，因此这里将不会引入。你可以通过Spark文档中的 Cluster Overview了解更多细节。
概述
简而言之，Spark是支持Kafka的，但是这里存在许多不完善的地方。
Spark代码库中的 KafkaWordCount对于我们来说是个非常好的起点，但是这里仍然存在一些开放式问题。
特别是我想了解如何去做：
在完成这些操作时，我同样碰到了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时，我指的是如何完成这些逻辑消费者应用程序中的一个设置。
这里有一些简单的例子
这里我们不妨看一下现实应用中的复杂性——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中的读取和写入。
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中获得：
val ssc: StreamingContext = ??? // ignore for now val kafkaParams: Map[String, String] = Map("group.id" -> "terran", /* ignore rest */) val numInputDStreams = 5 val kafkaDStreams = (1 to numInputDStreams).map { _ => KafkaUtils.createStream(...) }
在这个例子中，我们建立了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”进行读取。
val ssc: StreamingContext = ??? // ignore for now val kafkaParams: Map[String, String] = Map("group.id" -> "terran", ...) val consumerThreadsPerInputDstream = 3 val topics = Map("zerg.hydra" -> consumerThreadsPerInputDstream) val stream = KafkaUtils.createStream(ssc, kafkaParams, topics, ...)
KafkaUtils.createStream方法被重载，因此这里有一些不同方法的特征。在这里，我们会选择Scala派生以获得最佳的控制。
结合选项1和选项2
下面是一个更完整的示例，结合了上述两种技术：
val ssc: StreamingContext = ??? val kafkaParams: Map[String, String] = Map("group.id" -> "terran", ...) val numDStreams = 5 val topics = Map("zerg.hydra" -> 1) val kafkaDStreams = (1 to numDStreams).map { _ => KafkaUtils.createStream(ssc, kafkaParams, topics, ...) }
我们建立了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。
val ssc: StreamingContext = ??? val kafkaParams: Map[String, String] = Map("group.id" -> "terran", ...) val readParallelism = 5 val topics = Map("zerg.hydra" -> 1) val kafkaDStreams = (1 to readParallelism).map { _ => KafkaUtils.createStream(ssc, kafkaParams, topics, ...) } //> collection of five *input* DStreams = handled by five receivers/tasks val unionDStream = ssc.union(kafkaDStreams) // often unnecessary, just showcasing how to do it //> single DStream val processingParallelism = 20 val processingDStream = unionDStream(processingParallelism) //> single DStream but now with 20 partitions
在下一节中，我将把所有部分结合到一起，并且联合实际数据处理进行讲解。
写入到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。
最终结果看起来如下：
val producerPool = { // See the full code on GitHub for details on how the pool is created val pool = createKafkaProducerPool(kafkaZkCluster.kafka.brokerList, outputTopic.name) ssc.sparkContext.broadcast(pool) } stream.map { ... }.foreachRDD(rdd => { rdd.foreachPartition(partitionOfRecords => { // Get a producer from the shared pool val p = producerPool.value.borrowObject() partitionOfRecords.foreach { case tweet: Tweet => // Convert pojo back into Avro binary format val bytes = converter.value.apply(tweet) // Send the bytes to Kafka p.send(bytes) } // Returning the producer to the pool also shuts it down producerPool.value.returnObject(p) }) })
需要注意的是， Spark Streaming每分钟都会建立多个RDDs，每个都会包含多个分区，因此你无需为Kafka生产者实例建立新的Kafka生产者，更不用说每个Kafka消息。上面的步骤将最小化Kafka生产者实例的建立数量，同时也会最小化TCP连接的数量（通常由Kafka集群确定）。你可以使用这个池设置来精确地控制对流应用程序可用的Kafka生产者实例数量。如果存在疑惑，尽量用更少的。
下面的代码是示例Spark Streaming应用程序的要旨（所有代码参见 这里）。这里，我做一些解释：
// Set up the input DStream to read from Kafka (in parallel) val kafkaStream = { val sparkStreamingConsumerGroup = "spark-streaming-consumer-group" val kafkaParams = Map( "zookeeper.connect" -> "zookeeper1:2181", "group.id" -> "spark-streaming-test", "zookeeper.connection.timeout.ms" -> "1000") val inputTopic = "input-topic" val numPartitionsOfInputTopic = 5 val streams = (1 to numPartitionsOfInputTopic) map { _ => KafkaUtils.createStream(ssc, kafkaParams, Map(inputTopic -> 1), StorageLevel.MEMORY_ONLY_SER).map(_._2) } val unifiedStream = ssc.union(streams) val sparkProcessingParallelism = 1 // You'd probably pick a higher value than 1 in production. unifiedStream.repartition(sparkProcessingParallelism) } // We use accumulators to track global "counters" across the tasks of our streaming app val numInputMessages = ssc.sparkContext.accumulator(0L, "Kafka messages consumed") val numOutputMessages = ssc.sparkContext.accumulator(0L, "Kafka messages produced") // We use a broadcast variable to share a pool of Kafka producers, which we use to write data from Spark to Kafka. val producerPool = { val pool = createKafkaProducerPool(kafkaZkCluster.kafka.brokerList, outputTopic.name) ssc.sparkContext.broadcast(pool) } // We also use a broadcast variable for our Avro Injection (Twitter Bijection) val converter = ssc.sparkContext.broadcast(SpecificAvroCodecs.toBinary[Tweet]) // Define the actual data flow of the streaming job kafkaStream.map { case bytes => numInputMessages += 1 // Convert Avro binary data to pojo converter.value.invert(bytes) match { case Success(tweet) => tweet case Failure(e) => // ignore if the conversion failed } }.foreachRDD(rdd => { rdd.foreachPartition(partitionOfRecords => { val p = producerPool.value.borrowObject() partitionOfRecords.foreach { case tweet: Tweet => // Convert pojo back into Avro binary format val bytes = converter.value.apply(tweet) // Send the bytes to Kafka p.send(bytes) numOutputMessages += 1 } producerPool.value.returnObject(p) }) }) // Run the streaming job ssc.start() ssc.awaitTermination()
就我自己而言，我非常喜欢 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和深挖源代码。这里，我不得不说，维护帮助文档的同学做的实在是太棒了。
你可能已经发现在Spark中仍然有一些尚未解决的问题，下面我描述一些我的发现：
一方面，在对Kafka进行读写上仍然存在一些含糊不清的问题，你可以在类似 Multiple Kafka Receivers and Union和 How to scale more consumer to Kafka stream mailing list的讨论中发现。
另一方面，Spark Streaming中一些问题是因为Spark本身的固有问题导致，特别是故障发生时的数据丢失问题。换句话说，这些问题让你不想在生产环境中使用Spark。
在这些故障处理策略和Kafka聚焦的问题之外之外，扩展性和稳定性上的关注同样不可忽视。再一次，仔细观看 Bobby和Tom的视频以获得更多细节。在Spark使用经验上，他们都永远比我更丰富。
当然，我也有我的 评论，在 G1 garbage（在Java 1.7.0u4+中） 上可能也会存在问题。但是，我从来都没碰到这个问题。
在我实现这个示例的代码时，我做了一些重要的笔记。虽然这不是一个全面的指南，但是在你开始Kafka整合时可能发挥一定的作用。它包含了 Spark Streaming programming guide中的一些信息，也有一些是来自Spark用户的mailing list。
// Required to gain access to RDD transformations via implicits. import org.apache.spark.SparkContext._ // Required when working on `PairDStreams` to gain access to e.g. `DStream.reduceByKey` // (versus `DStream.transform(rddBatch => rddBatch.reduceByKey()`) via implicits. // // See also https://spark.apache.org/docs/1.1.0/programming-guide.html#working-with-key-value-pairs import org.apache.spark.streaming.StreamingContext.toPairDStreamFunctions
如果你是 Twitter Algebird的爱好者，你将会喜欢使用Count-Min Sketch和Spark中的一些特性，代表性的，你会使用reduce或者reduceByWindow这样的操作（比如， DStreams上的转换）。Spark项目包含了 Count-Min Sketch和 HyperLogLog的示例介绍。
如果你需要确定Algebird数据结构的内存介绍，比如Count-Min Sketch、HyperLogLog或者Bloom Filters，你可以使用SparkContext日志进行查看，更多细节参见 Determining Memory Consumption。
Kafka整合
我前文所述的一些增补：
测试
完整的Spark Streaming示例代码可以在 kafka-storm-starter查看。这个应用包含了Kafka、Zookeeper、Spark，以及上文我讲述的示例。
总体来说，我对我的初次Spark Streaming体验非常满意。当然，在Spark/Spark Streaming也存在一些需要特别指明的问题，但是我肯定Spark社区终将解决这些问题。在这个过程中，得到了Spark社区积极和热情的帮助，同时我也非常期待Spark 1.2版本的新特性。
在大型生产环境中，基于Spark还需要一些TLC才能达到Storm能力，这种情况我可能将它投入生产环境中么？大部分情况下应该不会，更准确的说应该是现在不会。那么在当下，我又会使用Spark Streaming做什么样的处理？这里有两个想法，我认为肯定存在更多：
感谢Spark社区对大数据领域所作出的贡献！
Mailing list discussions：
更多《问底》内容
《问底》是ImapBox云计算频道新建栏目，以实践为本，分享个人对于新时代软件架构与研发的深刻见解。在含有“【问底】”字样标题的文章中，你会看到某个国外IT巨头的架构分享，会看到国内资深工程师对某个技术的实践总结，更会看到一系列关于某个新技术的探索。《问底》邀请对技术具有独特/深刻见解的你一起打造一片只属于技术的天空，详情可邮件至zhonghao@csdn.net。
原文链接： Integrating Kafka and Spark Streaming: Code Examples and State of the Game（翻译/童阳 责编/仲浩）
本文为ImapBox原创文章，未经允许不得转载，如需转载请联系market#csdn.net(#换成@)
本网页所有文字内容由 imapbox邮箱云存储,邮箱网盘, iurlBox网页地址收藏管理器 下载并得到。
ImapBox 邮箱网盘 工具地址: https://www.imapbox.com/download/ImapBox.5.5.1_Build20141205_CHS_Bit32.exe
PC6下载站地址：PC6下载站分流下载
本网页所有视频内容由 imoviebox边看边下-网页视频下载, iurlBox网页地址收藏管理器 下载并得到。
ImovieBox 网页视频 工具地址: https://www.imapbox.com/download/ImovieBox4.7.0_Build20141115_CHS.exe
本文章由: imapbox邮箱云存储,邮箱网盘,ImageBox 图片批量下载器,网页图片批量下载专家,网页图片批量下载器,获取到文章图片,imoviebox网页视频批量下载器,下载视频内容,为您提供.
阅读和此文章类似的: 全球云计算
官方软件产品操作指南 (170)