Spark整合kafka0.10.0新特性（一）

Spark整合kafka0.10.0新特性(一)

Spark Streaming + Kafka Integration Guide (Kafka broker version 0.10.0 or higher)整合kafka0.10.0新特性（API都在实验中）。

The Spark Streaming integration for Kafka 0.10和kafka0.8的Direct Stream approach非常相似，并行度Kafka分区和Spark分区的比例1:1，并且可以访问Kafka的偏移和元数据。然而，新的整合方案使用的是new Kafka consumer API 而不是 simple API，所以在使用过程中需要注意区别，这个版本的整合现在正处于experimental，因此API可能随着时间会有变化。 Linking

For Scala/Java applications using SBT/Maven project definitions, link your streaming application with the following artifact (see Linking sectionin the main programming guide for further information).

groupId = org.apache.spark

artifactId = spark-streaming-kafka-0-10_2.11 version = 2.1.0

无论使用sbt还是maven，都需引入上面的坐标。

Creating a Direct Stream

注意导入包的路径是org.apache.spark.streaming.kafka010，切勿倒错包。

import org.apache.kafka.clients.consumer.ConsumerRecord import org.apache.kafka.common.serialization.StringDeserializer import org.apache.spark.streaming.kafka010._

import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

val kafkaParams = Map[String, Object](

\ \ \

\ \

\)

val topics = Array(\

val stream = KafkaUtils.createDirectStream[String, String](

streamingContext,//create entry point for all streaming functionality

PreferConsistent,//important feature：preferConsistent是一个方法，是consumer调度分区的位置策略

Subscribe[String, String](topics, kafkaParams)//is also import feature ：Subscribe是consumer的消费策略

)

stream.map(record => (record.key, record.value))

重点解释一下 PreferConsistent方法，首先我们还是看一下PreferConsistent方法实现，源码如下：

[java] view plain copy 在CODE上查看代码片派生到我的代码片 package org.apache.spark.streaming.kafka010

import java.{util => ju}

import scala.collection.JavaConverters._

import org.apache.kafka.common.TopicPartition import org.apache.spark.annotation.Experimental

import org.apache.spark.streaming.kafka010.{LocationStrategies, LocationStrategy, PreferFixed} /**

* :: Experimental ::

* Choice of how to schedule consumers for a given TopicPartition on an executor. * See [[LocationStrategies]] to obtain instances.

* Kafka 0.10 consumers prefetch messages, so it's important for performance

* to keep cached consumers on appropriate executors, not recreate them for every partition. * Choice of location is only a preference, not an absolute; partitions may be scheduled elsewhere. * 实验性API:

* 在executor上consumer如何调度给定的TopicPartition，使用LocationStrategies获取调度策略实例

* Kafka 0.10的消费者可以预取消息，因此对于性能来说在适合的executors上缓存

consumers是比较重要的，而不是对每一个分区

* 进行重新创建。对于分区位置的选择只是一个偏好，并非是绝对的。分区可能被调度到其他位置 * * */

@Experimental

sealed abstract class LocationStrategy

/**

* 使用PreferBrokers策略，必须是你的executors和kafka brokers在相同节点上。 */

private case object PreferBrokers extends LocationStrategy /**

* 大多数情况下使用PreferConsistent需要一贯的将kafka的分区分布到所有的executors上 */

private case object PreferConsistent extends LocationStrategy /**

* Use this to place particular TopicPartitions on particular hosts if your load is uneven.

* Any TopicPartition not specified in the map will use a consistent location.

* 默认情况如果分区加载的不均衡的话，可以使用这个策略：放置特定的分区到特定的主机上

* 任何TopicPartition没有和hosts映射的TopicPartition将会使用consistent location（就是安置到所有executor） *

* 参数Map：就是TopicPartition和主机地址的映射 */

private case class PreferFixed(hostMap: ju.Map[TopicPartition, String]) extends LocationStrategy /**

* :: Experimental :: object to obtain instances of [[LocationStrategy]] * */

@Experimental

object LocationStrategies { /**

* :: Experimental ::

* Use this only if your executors are on the same nodes as your Kafka brokers. */

@Experimental

def PreferBrokers: LocationStrategy =

org.apache.spark.streaming.kafka010.PreferBrokers /**

* :: Experimental ::

* Use this in most cases, it will consistently distribute partitions across all executors. */

@Experimental

def PreferConsistent: LocationStrategy =

org.apache.spark.streaming.kafka010.PreferConsistent /**

* :: Experimental ::

* Use this to place particular TopicPartitions on particular hosts if your load is uneven.

* Any TopicPartition not specified in the map will use a consistent location.

* 两个方法就是Map类型不一样而已，一个是Scala Map 另一个是Java Map * */

@Experimental

def PreferFixed(hostMap: collection.Map[TopicPartition, String]): LocationStrategy = new PreferFixed(new ju.HashMap[TopicPartition, String](hostMap.asJava)) /**

* :: Experimental ::

* Use this to place particular TopicPartitions on particular hosts if your load is uneven. * Any TopicPartition not specified in the …… 此处隐藏：4937字，全部文档内容请下载后查看。喜欢就下载吧 ……