Spark Streaming 消费kafka到HDFS

先说下程序功能，使用Spark Streaming 实时消费kafka，并将message写入HDFS上指定的topic目录中。
消费kafka使用的是Spark提供的Direct Approach方法，然后利用HDFS API将不同topic下的message写到各自topic目录下。

Spark Streaming简介

Spark Streaming是Spark的扩展，能够扩展的。高吞吐、容错的处理实时数据流。Spark Streaming的工作流程是将接受到实时的数据划分到不同的batch中，然后由Spark Engine处理并生成结果batch。如下图：

Spark Streaming提供了一个高级的抽象模型，叫做discretized stream或者叫做DStream,它代表了一个持续的数据流。

Saprk Streaming中的Context是StreamingContext，StreamingContext是所有功能的主入口，可以通过两种方法create，

// first
val conf = new SparkConf().setAppName(appName).setMaster(master)
val ssc = new StreamingContext(conf, Seconds(1))
// second
val sc = ...                // existing SparkContext
val ssc = new StreamingContext(sc, Seconds(1))

然后由ssc(StreamingContext)创建一个DStream，DStream可以指定数据输入源（DStream支持很多数据源，如kakfa、flume、twitter、TCP socketd），这里来个简单的TCP socket例子

// Create a DStream that will connect to hostname:port, like localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)

lines的类型是DStream，代表从数据服务器上接受到的数据流。lines中的每一条记录是一行文本。可以对其进行一些高级操作如map、reduce、join和window，_需要注意的是有些RDD操作并没有对DStream开放，如果想使用那些API，则需要进行Transform Operation_，transform Operation就是讲DStream转换为RDD，方法调用为dStream.transform，代码示例

val dStreamToRdd = dStream.transform(rdd => {
  ...
  rdd
})

Spark Streaming仅仅设置这些计算， 它并没有马上被执行。当所有的计算设置完后，我们可以调用下面的代码启动处理

ssc.start()             // Start the computation
ssc.awaitTermination()  // Wait for the computation to terminate

Spark Streaming Demo

来个简单的Demo看看，此Demo可以在${SPARK_HOME}/examples/src/main/scala/org/apache/spark/examples/streaming中找到

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.storage.StorageLevel

/**
 * Counts words in UTF8 encoded, '\n' delimited text received from the network every second.
 *
 * Usage: NetworkWordCount <hostname> <port>
 * <hostname> and <port> describe the TCP server that Spark Streaming would connect to receive data.
 *
 * To run this on your local machine, you need to first run a Netcat server
 *    `$ nc -lk 9999`
 * and then run the example
 *    `$ bin/run-example org.apache.spark.examples.streaming.NetworkWordCount localhost 9999`
 */
object NetworkWordCount {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println("Usage: NetworkWordCount <hostname> <port>")
      System.exit(1)
    }

    StreamingExamples.setStreamingLogLevels()

    // Create the context with a 1 second batch size
    val sparkConf = new SparkConf().setAppName("NetworkWordCount")
    // 得到一个spark streaming的context，为spark streaming的主入口
    val ssc = new StreamingContext(sparkConf, Seconds(1))

    // Create a socket stream on target ip:port and count the
    // words in input stream of \n delimited text (eg. generated by 'nc')
    // Note that no duplication in storage level only for running locally.
    // Replication necessary in distributed scenario for fault tolerance.
    // 由ssc创建一个TCP socket的DStream，DStream存放的就是接收的数据流
    val lines = ssc.socketTextStream(args(0), args(1).toInt, StorageLevel.MEMORY_AND_DISK_SER)
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
    wordCounts.print()
    // 调用start之后，才能进行真正的计算
    ssc.start()
    ssc.awaitTermination()
  }
}

打开两个终端AB，在A输入nc -lk 9999命令，在B进入spark的home目录运行./bin/run-example streaming.NetworkWordCount localhost 9999命令。在A中输入一些测试数据，就可以在B查看计算结果。

Spark Streaming 消费Kafka

Spark消费kafka有两种方式，这里主要介绍第二种_Direct Approach (No Receivers)_

Direct Approach将kafka数据源包裹成了一个KafkaRDD，RDD里的partition 对应的数据源为kafka的partition（有利于并行的读取kafka message）。
kafka message并不是立马被读入spark内存，而是在Kafka存着呢，直到有实际的Action被触发，才会去kafka主动拉数据。
Direct Approach使用的是kafka simple consumer api，这样可以指定从某个offset处进行读取，有利于故障恢复。

Spark Streaming consumer Kafka to HDFS

本篇文章主要将kafka中的message通过spark streaming根据不同的topic写到不同的hdfs文件中，并且能够记录消费message的offset，以支持故障恢复。

offset存储方案选择

• 利用checkpoint将offset存储在hdfs
  简单容易实现，根据需求有一定的局限，无法更好的满足需求
• 将offset存储在HDFS
  一开始为了尽量减少依赖的组件，减少组件原因造成应用故障，使用选择将offset存储在hdfs上，但在开发中考虑到offset文件需要频繁的读写操作，可能会在性能上有所影响
• 将offset存储在zk
  跟Kafka high-level consumer API一样将offset存储在zk上，代码逻辑图如下：
  
  由Spark driver计算下个batch的offsets，指导executor消费对应的topics和partitions。使消费Kafka消息，就像消费文件系统文件一样。

代码实现

Spark Streaming通过Direct Approach接收数据的入口自然是KafkaUtils.createDirectStream 了。在调用该方法时，会先创建KafkaCluster，得到offset

def createDirectStream[
  K: ClassTag,
  V: ClassTag,
  KD <: Decoder[K]: ClassTag,
  VD <: Decoder[V]: ClassTag] (
    ssc: StreamingContext,
    kafkaParams: Map[String, String],
    topics: Set[String]
): InputDStream[(K, V)] = {
  val messageHandler = (mmd: MessageAndMetadata[K, V]) => (mmd.key, mmd.message)
  val kc = new KafkaCluster(kafkaParams)
  val fromOffsets = getFromOffsets(kc, kafkaParams, topics)
  new DirectKafkaInputDStream[K, V, KD, VD, (K, V)](
    ssc, kafkaParams, fromOffsets, messageHandler)
}

KafkaCluster是操作kafka的一个关键类，但由于其是private的，要想在自己的代码中使用该类得重写该类。

创建KafkaManager类，其中主要包含三个方法，分别是根据offset创建一个DStream、得到offset、更新zk上的offset，代码如下：

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.Decoder
import org.apache.spark.SparkException
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka.KafkaCluster.LeaderOffset
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaCluster, KafkaUtils}

import scala.reflect.ClassTag

/**
  * Created by hunhun on 2016/5/27.
  */
class KafkaManager(val kafkaParams: Map[String, String]) {

  // KafkaCluster in Spark is overwrited by myself
  private val kc = new KafkaCluster(kafkaParams)

  // 根据offset创建一个DStream
  // return key message topic
  def createDirectStream[K: ClassTag, V: ClassTag, KD <: Decoder[K]: ClassTag, VD <: Decoder[V]: ClassTag]
    (ssc: StreamingContext, kafkaParams: Map[String, String], topics: Set[String]): InputDStream[(K, V, String)] = {
    val groupId = kafkaParams.get("group.id").get
    // 在zookeeper上读取offsets前先根据实际情况更新offsets
    setOrUpdateOffsets(topics, groupId)

    //从zookeeper上读取offset开始消费message
    val messages = {
      // Either 类型
      val partitionsE = kc.getPartitions(topics)
      if (partitionsE.isLeft)
        // s"xx ${}" 字符串插值
        throw new SparkException(s"get kafka partition failed: ${partitionsE.left.get}")
      val partitions = partitionsE.right.get
      val consumerOffsetsE = kc.getConsumerOffsets(groupId, partitions)
      if (consumerOffsetsE.isLeft)
        throw new SparkException(s"get kafka consumer offsets failed: ${consumerOffsetsE.left.get}")
      val consumerOffsets = consumerOffsetsE.right.get
      // 从指定offsets处消费kafka
      // messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key(), mmd.message())
      // MessageAndMetadata里包含message的topic message 等等信息
      KafkaUtils.createDirectStream[K, V, KD, VD, (K, V, String)](
        ssc, kafkaParams, consumerOffsets, (mmd: MessageAndMetadata[K, V]) => ( mmd.key, mmd.message, mmd.topic))
    }
    messages
  }

  private def setOrUpdateOffsets(topics: Set[String], groupId: String): Unit = {
    topics.foreach(topic => {
      var hasConsumed = true
      val partitionsE = kc.getPartitions(Set(topic))
      if (partitionsE.isLeft)
        throw new SparkException(s"get kafka partition failed: ${partitionsE.left.get}")
      val partitions = partitionsE.right.get
      val consumerOffsetsE = kc.getConsumerOffsets(groupId, partitions)
      if (consumerOffsetsE.isLeft) hasConsumed = false
      // 某个groupid首次没有offset信息，会报错，从头开始读
      if (hasConsumed) {// 消费过
        /**
          * 如果streaming程序执行的时候出现kafka.common.OffsetOutOfRangeException，
          * 说明zk上保存的offsets已经过时了，即kafka的定时清理策略已经将包含该offsets的文件删除。
          * 针对这种情况，只要判断一下zk上的consumerOffsets和earliestLeaderOffsets的大小，
          * 如果consumerOffsets比earliestLeaderOffsets还小的话，说明consumerOffsets已过时,
          * 这时把consumerOffsets更新为earliestLeaderOffsets
          */
        val earliestLeaderOffsetsE = kc.getEarliestLeaderOffsets(partitions)
        if (earliestLeaderOffsetsE.isLeft)
          throw new SparkException(s"get earliest leader offsets failed: ${earliestLeaderOffsetsE.left.get}")
        val earliestLeaderOffsets = earliestLeaderOffsetsE.right.get
        val consumerOffsets = consumerOffsetsE.right.get

        // 可能只是存在部分分区consumerOffsets过时，所以只更新过时分区的consumerOffsets为earliestLeaderOffsets
        var offsets: Map[TopicAndPartition, Long] = Map()
        consumerOffsets.foreach({ case(tp, n) =>
          val earliestLeaderOffset = earliestLeaderOffsets(tp).offset
          if (n < earliestLeaderOffset) {
            println("consumer group:" + groupId + ",topic:" + tp.topic + ",partition:" + tp.partition +
              " offsets已经过时，更新为" + earliestLeaderOffset)
            offsets += (tp -> earliestLeaderOffset)
          }
        })
        if (!offsets.isEmpty) {
          kc.setConsumerOffsets(groupId, offsets)
        }
      } else {// 没有消费过
      val reset = kafkaParams.get("auto.offset.reset").map(_.toLowerCase)
        var leaderOffsets: Map[TopicAndPartition, LeaderOffset] = null
        if (reset == Some("smallest")) {// 从头消费
          val leaderOffsetsE = kc.getEarliestLeaderOffsets(partitions)
          if (leaderOffsetsE.isLeft)
            throw new SparkException(s"get earliest leader offsets failed: ${leaderOffsetsE.left.get}")
          leaderOffsets = leaderOffsetsE.right.get
        } else { // 从最新offset处消费
          val leaderOffsetsE = kc.getLatestLeaderOffsets(partitions)
          if (leaderOffsetsE.isLeft)
            throw new SparkException(s"get latest leader offsets failed: ${leaderOffsetsE.left.get}")
          leaderOffsets = leaderOffsetsE.right.get
        }
        val offsets = leaderOffsets.map {
          case (tp, offset) => (tp, offset.offset)
        }
        kc.setConsumerOffsets(groupId, offsets)
      }
    })
  }

  def updateZKOffsets(rdd: RDD[(String, String, String)]) : Unit = {
    val groupId = kafkaParams.get("group.id").get
    val offsetsList = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

    for (offsets <- offsetsList) {
      val topicAndPartition = TopicAndPartition(offsets.topic, offsets.partition)
      val o = kc.setConsumerOffsets(groupId, Map((topicAndPartition, offsets.untilOffset)))
      if (o.isLeft) {
        println(s"Error updating the offset to Kafka cluster: ${o.left.get}")
      }
    }
  }
}

创建主类SparkConsumerKafka

import kafka.serializer.StringDecoder
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FSDataOutputStream, Path, FileSystem}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer
import scala.util.Random

/**
  * Created by hunhun on 2016/5/23.
  */
object SparkConsumerKafka {
  def main(args: Array[String]) {
    if (args.length < 4) {
      System.err.println( s"""
                             |Usage: DirectKafkaWordCount <brokers> <topics> <groupid>
                             |  <brokers> is a list of one or more Kafka brokers
                             |  <topics> is a list of one or more kafka topics to consume from
                             |  <groupid> is a consume group
                             |  <hdfspath> is a HDFS Path, like /user/admin/scalapath
                             |
        """.stripMargin)
      System.exit(1)
    }

    val Array(brokers, topics, groupId, hdfsPath) = args
    val sparkConf = new SparkConf().setAppName("SparkConsumerKafka")
    // spark.streaming.kafka.maxRatePerPartition 限速
    val ssc = new StreamingContext(sparkConf, Seconds(60))

    // Create direct kafka stream with brokers and topics
    val topicsSet = topics.split(",").toSet
    val kafkaParams = Map[String, String](
      "metadata.broker.list" -> brokers,
      "group.id" -> groupId,
      "auto.offset.reset" -> "smallest"
    )

    val km = new KafkaManager(kafkaParams)

    val messages = km.createDirectStream[String, String, StringDecoder, StringDecoder](ssc,
      kafkaParams, topicsSet)
    // foreachRDD是DStream的output操作
    messages.foreachRDD( rdd => {
      if (!rdd.isEmpty()){
        rdd.foreachPartition{ partitionOfRecords =>
          // 得到HDFS的操作client
          // 此代码必须放在worker中创建，如果在driver中创建，则将会被序列化到worker中
          // 在worker中操作时会报错。
          val conf = new Configuration()
          val fs = FileSystem.get(conf)

          var messageList = new ListBuffer[String]
          var topic = ""
          // 此处代码只为得到topic
          // 之所以将message放入messageList中是因为partitionOfRecords变为list之后
          // 拿到一个record的topic之后，partitionOfRecords中的内容将消失，具体原因不知道
          // 代码如下：
          // val topic = partitionOfRecords.toList(0)._3
          // 然后写入hdfs时调用
          // partitionOfRecords.foreach(record => outputStream.write((record._2 + "\n").getBytes("UTF-8")))
          // 此时写入hdfs的内容为null，不知道为什么为null
          // 所以只好在得到topic的同时把message先存入messageList
          partitionOfRecords.foreach(record => {
            messageList += record._2
            if (topic == ""){
              topic = record._3
            }
          })

          if (topic != ""){
            // 拼出各个topic message的文件地址
            val path = new Path(hdfsPath + "/" + topic + "/" + Random.nextInt(100) + topic
                          + System.currentTimeMillis())
            // 创建一个HDFS outputStream流
            val outputStream = if (fs.exists(path)){
              fs.append(path)
            }else{
              fs.create(path)
            }
            // 将message逐条写入
            messageList.foreach(message => outputStream.write((message + "\n").getBytes("UTF-8")))
            outputStream.close()
          }

        }
        // 更新zk上的offset
        km.updateZKOffsets(rdd)
      }
    })

    ssc.start()
    ssc.awaitTermination()
  }
}

附加

将offset写入checkpoint中

import kafka.serializer.StringDecoder
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FSDataOutputStream, Path, FileSystem}
import org.apache.spark.SparkConf
import org.apache.spark.sql.catalyst.expressions.Second
import org.apache.spark.streaming.kafka.{HasOffsetRanges, OffsetRange, KafkaUtils}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by hunhun on 2016/5/26.
  */
object WriteToHdfsWithCheckpoint {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println(s"""
                            |Usage: DirectKafkaWordCount <brokers> <topics>
                            |  <brokers> is a list of one or more Kafka brokers
                            |  <topics> is a list of one or more kafka topics to consume from
                            |
        """.stripMargin)
      System.exit(1)
    }

    val Array(brokers, topics) = args

    // Create context with 2 second batch interval
    val sparkConf = new SparkConf().setAppName("ConsumerKafkaToHdfsWithCheckPoint")

    // Create direct kafka stream with brokers and topics
    val topicsSet = topics.split(",").toSet
    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers,
      "auto.offset.reset" -> "smallest")
    val checkpointPath = "hdfs://127.0.0.1:8020/spark_checkpoint"
    def functionToCreateContext(): StreamingContext = {
      val ssc = new StreamingContext(sparkConf, Seconds(2))
      val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)
      // checkpoint 在哪保存？ 这样会不会出现offset保存了但没有成功写入hdfs
      // checkpoint调用的位置会有影响吗？
      ssc.checkpoint(checkpointPath)

      messages.map(_._2).foreachRDD(rdd => {
        rdd.foreachPartition{ partitionOfRecords =>
          val conf = new Configuration()
          val fs = FileSystem.get(conf)
          val path = new Path("/user/admin/scalapath/test" + System.currentTimeMillis())
          val outputStream : FSDataOutputStream = if (fs.exists(path)){
            fs.append(path)
          }else{
            fs.create(path)
          }
          partitionOfRecords.foreach(record => outputStream.write((record + "\n").getBytes("UTF-8")))
          outputStream.close()
        }
      })
      ssc
    }
    // 决定否创建新的Context
    // 没有checkpoint就创建一个新的StreamingContext即初次启动应用
    // 如果有checkpoint则checkpoint中记录的信息恢复StreamingContext
    val context = StreamingContext.getOrCreate(checkpointPath, functionToCreateContext _)

    context.start()
    context.awaitTermination()
  }
}