Spark Streaming和Kafka整合开发指南(二)

<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka_2.10</artifactId> <version>1.3.0</version> </dependency>

　　如果你是使用SBT，可以这么引入：

libraryDependencies += "org.apache.spark" % "spark-streaming-kafka_2.10" % "1.3.0"

　　2、编程

　　在Streaming应用程序代码中，引入KafkaUtils ，并创建DStream输入流：

 import org.apache.spark.streaming.kafka._

 val directKafkaStream = KafkaUtils.createDirectStream[
     [key class], [value class], [key decoder class], [value decoder class] ](
     streamingContext, [map of Kafka parameters], [set of topics to consume])

　　在 Kafka parameters参数中，你必须指定 metadata.broker.list或者bootstrap.servers参数。在默认情况下，Spark Streaming将会使用最大的偏移量来读取Kafka每个分区的数据。如果你配置了auto.offset.reset为smallest，那么它将会从最小的偏移量开始消费。

　　当然，你也可以使用KafkaUtils.createDirectStream的另一个版本从任意的位置消费数据。如果你想回去每个batch中Kafka的偏移量，你可以如下操作：

 directKafkaStream.foreachRDD { rdd => 
     val offsetRanges = rdd.asInstanceOf[HasOffsetRanges]
     // offsetRanges.length = # of Kafka partitions being consumed
     ...
 }

你可以通过这种方式来手动地更新Zookeeper里面的偏移量，使得基于Zookeeper偏移量的Kafka监控工具可以使用。

　　还有一点需要注意，因为这里介绍的方法没有使用到Receiver，所以Spark中关于spark.streaming.receiver.*相关的配置参数将不会对创建DStreams 有影响。我们可以使用spark.streaming.kafka.*参数进行配置。

　　3、部署

　　对应任何的Spark 应用，我们都是用spark-submit来启动你的应用程序，对于Scala和Java用户，如果你使用的是SBT或者是Maven，你可以将spark-streaming-kafka_2.10及其依赖打包进应用程序的Jar文件中，并确保spark-core_2.10和 spark-streaming_2.10标记为provided，因为它们在Spark 安装包中已经存在：

<dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming_2.10</artifactId>
          <version>1.3.0</version>
          <scope>provided</scope>
</dependency>

<dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_2.10</artifactId>
          <version>1.3.0</version>
          <scope>provided</scope>
</dependency>

然后使用spark-submit来启动你的应用程序。

　　本文翻译自：https://spark.apache.org/docs/latest/streaming-kafka-integration.html