kafka direct方式获取数据解析

Receiver方式

处理流程

实际上作kafka receiver的时候，经过receiver来获取数据，这个时候，kafka receiver是使用的kafka高层次的comsumer api来实现的。receiver会从kafka中获取数据，而后把它存储到咱们具体的Executor内存中。而后Spark streaming也就是driver中，会根据这获取到的数据，启动job去处理。

receiver缺点

1. 已经拉取的数据消费失败后，会致使数据丢失。此问题虽然能够经过WAL方式或者Memory_and_Disc2解决，可是存在耗时等问题
2. 使用了kafka consumer的高阶API，KafkaInputDStream的实现和咱们经常使用的consumer实现相似，须要zk额外的记录偏移量

Direct方式

实现

在使用kafka接收消息时，都是调用了KafkaUtils里面createStream的不一样实现。

receiver方式的实现方式以下。

/**
   * 建立一个inputStream，从kafkaBrokers上拉去消息，须要传入zk集群信息，默认会复制到另外一个excutor
   */
  def createStream(
      ssc: StreamingContext,// spark上下文
      zkQuorum: String,// zk集群信息（hostname:port,hostname:port...）
      groupId: String,// 当前consumer所属分组
      topics: Map[String, Int],// Map[topic_name,numPartitions],topic消费对应的分区
      storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
    ): ReceiverInputDStream[(String, String)] = {
    val kafkaParams = Map[String, String](
      "zookeeper.connect" -> zkQuorum, "group.id" -> groupId,
      "zookeeper.connection.timeout.ms" -> "10000")
      // 写日志
	  val walEnabled = WriteAheadLogUtils.enableReceiverLog(ssc.conf)
      // 组装成KafkaInputDStream
      new KafkaInputDStream[K, V, U, T](
            ssc, kafkaParams, topics, walEnabled, 	storageLevel)
  }

direct方式实现消费

/**
   * 摒弃了高阶的kafkaConsumerAPI直接从kafkaBrokers获取信息，能够保证每条消息只被消费一次
   * 特色：
   * - No receivers：没有receivers,直接从kafka拉取数据
   * - Offsets：不用zookeeper存储offsets,偏移量是经过stream本身跟踪记录的，能够经过HasOffsetRanges获取offset
   * - Failure Recovery故障恢复：须要开启sparkContext的checkpoint功能
   * - End-to-end semantics最终一致性：保证消息被消费且只消费一次
   * @return DStream of (Kafka message key, Kafka message value)
   */
  def createDirectStream[
    K: ClassTag,
    V: ClassTag,
    KD <: Decoder[K]: ClassTag,
    VD <: Decoder[V]: ClassTag] (
      ssc: StreamingContext,
      // brokers列表，Map("metadata.broker.list" -> brokers)
      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)
  }

特色

1. Direct的方式是会直接操做kafka底层的元数据信息，这样若是计算失败了，能够把数据从新读一下，从新处理。即数据必定会被处理。拉数据，是RDD在执行的时候直接去拉数据。
2. 因为直接操做的是kafka，kafka就至关于你底层的文件系统。这个时候能保证严格的事务一致性，即必定会被处理，并且只会被处理一次。而Receiver的方式则不能保证，由于Receiver和ZK中的数据可能不一样步，Spark Streaming可能会重复消费数据。而Direct api直接是操做kafka的，spark streaming本身负责追踪消费这个数据的偏移量或者offset，而且本身保存到checkpoint，因此它的数据必定是同步的，必定不会被重复。
3. 底层是直接读数据，没有所谓的Receiver，直接是周期性(Batch Intervel)的查询kafka，处理数据的时候，咱们会使用基于kafka原生的Consumer api来获取kafka中特定范围(offset范围)中的数据，这个时候，Direct Api访问kafka带来的一个显而易见的性能上的好处就是，若是你要读取多个partition，Spark也会建立RDD的partition，这个时候RDD的partition和kafka的partition是一致的。因此增长kafka中的topic的partition数量能够提升并行度。
4. 偏移量：默认从最新偏移量(largest)开始消费。若是设置了auto.offset.reset参数值为smallest将从最小偏移处开始消费。

checkpoint恢复后，若是数据累积太多处理不过来，怎么办?

1）限速，经过spark.streaming.kafka.maxRatePerPartition参数配置

2）加强机器的处理能力

3）放到数据缓冲池中。

模式比对

1. .简化并行性：无需建立多个输入Kafka流而且结合它们。 使用directStream，Spark Streaming将建立与要消费的Kafkatopic中partition分区同样多的RDD分区，这将从Kafka并行读取数据。 所以，在Kafka和RDD分区之间存在一对一映射，这更容易理解和调整。
2. 效率：在第一种方法中实现零数据丢失须要将数据存储在预写日志中，该日志进一步复制数据。 这其实是低效的，由于数据有效地被复制两次 - 一次是Kafka，另外一次是写入提早日志。 第二种方法消除了问题，由于没有接收器(zookeeper)，所以不须要预写日志。 将元数据信息直接保存在kafka中，能够从Kafka恢复消息。
3. Exactly-once语义：第一种方法使用Kafka的高级API在Zookeeper中存储消耗的偏移量。这是传统上消费Kafka数据的方式。虽然这种方法（与预写日志结合）能够确保零数据丢失（即至少一次语义），可是一些记录在一些故障下可能被消耗两次。这是由于Spark Streaming可靠接收的数据与Zookeeper跟踪的偏移之间存在不一致。所以，在第二种方法中，咱们使用简单的Kafka API,不使用Zookeeper的。偏移由Spark Streaming在其检查点内跟踪。这消除了Spark Streaming和Zookeeper / Kafka之间的不一致，因此每一个记录被Spark Streaming有效地精确接收一次，尽管失败了。为了实现输出结果的一次性语义，将数据保存到外部数据存储的输出操做必须是幂等的，或者是保存结果和偏移量的原子事务。

direct源码

获取offset集合，而后建立DirectKafkaInputDStream对象

//  class KafkaUtils  
private[kafka] def getFromOffsets(
      kc: KafkaCluster,
      kafkaParams: Map[String, String],
      topics: Set[String]
    ): Map[TopicAndPartition, Long] = {
    // createDirectStream方法kafkaParams入参：消费起始位置
    val reset = kafkaParams.get("auto.offset.reset").map(_.toLowerCase(Locale.ROOT))
    val result = for {
      topicPartitions <- kc.getPartitions(topics).right
      leaderOffsets <- (if (reset == Some("smallest")) {
        // smallest表示最小offset,即从topic的开始位置消费全部消息.
        kc.getEarliestLeaderOffsets(topicPartitions)
      } else {
        // largest表示接受接收最大的offset(即最新消息),
        kc.getLatestLeaderOffsets(topicPartitions)
      }).right
      // for循环中的 yield 会把当前的元素记下来，保存在集合中，循环结束后将返回该集合。Scala中for循环是有返回值的。若是被循环的是Map，返回的就是Map，被循环的是List，返回的就是List，以此类推。
    } yield {
      // 存放for循环的计算结果：map[TopicAndPartition, LeaderOffset]
      leaderOffsets.map { case (tp, lo) =>
          (tp, lo.offset)
      }
    }
    KafkaCluster.checkErrors(result)
  }

def createDirectStream{
    new DirectKafkaInputDStream[K, V, KD, VD, (K, V)](
      ssc, kafkaParams, fromOffsets, messageHandler)
}

DirectKafkaInputDStream.compute中建立KafkaRDD，并将offsets信息发送给inputStreamTracker.

override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {
    // Map[TopicAndPartition, LeaderOffset] topic的partiton对应偏移量集合
    val untilOffsets = clamp(latestLeaderOffsets(maxRetries))
    // 消息处理函数val messageHandler = (mmd: MessageAndMetadata[K, V]) => (mmd.key, mmd.message)
    // 建立KafkaRDD
    val rdd = KafkaRDD[K, V, U, T, R](
      context.sparkContext, kafkaParams, currentOffsets, untilOffsets, messageHandler)

    // 将topic和partition信息包装成OffsetRange对象中
    val offsetRanges = currentOffsets.map { case (tp, fo) =>
      val uo = untilOffsets(tp)
      OffsetRange(tp.topic, tp.partition, fo, uo.offset)
    }
   
 	// 将OffsetRange报告给InputInfoTracker记录
    ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)
    currentOffsets = untilOffsets.map(kv => kv._1 -> kv._2.offset)
    Some(rdd)
  }

KafkaRDD计算时直接从kafka上拉取数据

override def compute(thePart: Partition, context: TaskContext): Iterator[R] = {
    val part = thePart.asInstanceOf[KafkaRDDPartition]
    new KafkaRDDIterator(part, context)
}

private class KafkaRDDIterator(
      part: KafkaRDDPartition,
      context: TaskContext) extends NextIterator[R] {
	// 根据metadata.broker.list初始化KafkaCluster，用来链接到kafka
    val kc = new KafkaCluster(kafkaParams)
    
    var requestOffset = part.fromOffset
    var iter: Iterator[MessageAndOffset] = null

    // 提供一个最优的host优先访问，最大化的减小重试次数
    val consumer:SimpleConsumer = {
        // 重试次数大于0
      if (context.attemptNumber > 0) {
        kc.connectLeader(part.topic, part.partition).fold(
          errs => throw new SparkException(
            s"Couldn't connect to leader for topic ${part.topic} ${part.partition}: " +
              errs.mkString("\n")),
          consumer => consumer
        )
      } else {
          // 不用重试，直接链接
        kc.connect(part.host, part.port)
      }
    }

    // 建立请求拉取数据
    private def fetchBatch: Iterator[MessageAndOffset] = {
      val req = new FetchRequestBuilder()
        .addFetch(part.topic, part.partition, requestOffset, kc.config.fetchMessageMaxBytes)
        .build()
      val resp = consumer.fetch(req)
      // 失败重试 
      handleFetchErr(resp)
      // kafka may return a batch that starts before the requested offset
      resp.messageSet(part.topic, part.partition)
        .iterator
        .dropWhile(_.offset < requestOffset)
    }
    
    // 拉取失败，通知另外一个rdd从新尝试
    private def handleFetchErr(resp: FetchResponse) {
      if (resp.hasError) {
   		// Let normal rdd retry sort out reconnect attempts
        throw ErrorMapping.exceptionFor(err)
      }
    }

    override def getNext(): R = {
      if (iter == null || !iter.hasNext) {
          // 拉取数据
        iter = fetchBatch
      }
      if (!iter.hasNext) {
        assert(requestOffset == part.untilOffset, errRanOutBeforeEnd(part))
        finished = true
        null.asInstanceOf[R]
      } else {
          // 遍历拉取到的数据
        val item = iter.next()
        if (item.offset >= part.untilOffset) {
		  // 若是当前item的偏移量大于须要拉取的最大偏移量则结束
          finished = true
          null.asInstanceOf[R]
        } else {
          requestOffset = item.nextOffset
            // 将拉取到的数据交由messageHandler处理
          messageHandler(new MessageAndMetadata(
            part.topic, part.partition, item.message, item.offset, keyDecoder, valueDecoder))
        }
      }
    }
  }
}

经过chekpoint的方式保存offset

// DStream中定义checkpoint的实现类
class DirectKafkaInputDStream extends InputDStream{
   override val checkpointData =new DirectKafkaInputDStreamCheckpointData
}
class DirectKafkaInputDStreamCheckpointData extends DStreamCheckpointData(this) {
    def batchForTime: mutable.HashMap[Time, Array[(String, Int, Long, Long)]] = {
        // 定义一个不可变数组保存offset信息
      data.asInstanceOf[mutable.HashMap[Time, Array[OffsetRange.OffsetRangeTuple]]]
    }
}