• 在Kafka DirectStream初始化时，取得当前所有partition的存量offset，以让DirectStream能够从正确的位置开始读取数据。

• 读取消息数据，处理并存储结果。

• 提交offset，并将其持久化在可靠的外部存储中。

• enable.auto.commit=true。

  一但consumer挂掉，就会导致数据丢失或重复消费。

  offset不可控。

• （属于At-least-once语义，如果做好了幂等性，可以使用这种方式）：

  在Kafka 0.10+版本中，offset的默认存储由ZooKeeper移动到了一个自带的topic中，名为__consumer_offsets。

  Spark Streaming也专门提供了commitAsync() API用于提交offset。

  需要将参数修改为enable.auto.commit=false。

  在我实际测试中发现，这种offset的管理方式，不会丢失数据，但会出现重复消费。

  停掉streaming应用程序再次启动后，会再次消费停掉前最后的一个批次数据，应该是由于offset是异步提交的方式导致，offset更新不及时引起的。

  因此需要做好数据的幂等性。

  （修改源码将异步改为同步，应该是可以做到Exactly-once语义的）

• （推荐，采用这种方式，可以做到At-least-once语义）：

  可以将offset存放在第三方储中，包括RDBMS、Redis、ZK、ES等。

  若消费数据存储在带事务的组件上，则强烈推荐将offset存储在一起，借助事务实现 Exactly-once 语义。

在Kafka 0.10+版本中，offset的默认存储由ZooKeeper移动到了一个自带的topic中，名为__consumer_offsets。所以我们读写offset的对象正是这个topic，Spark Streaming也专门提供了commitAsync() API用于提交offset。实际上，一切都已经封装好了，直接调用相关API即可。

stream.foreachRDD { rdd =>
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  // 确保结果都已经正确且幂等地输出了
  stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}

ZooKeeper

class ZkKafkaOffsetManager(zkUrl: String) {
    private val logger = LoggerFactory.getLogger(classOf[ZkKafkaOffsetManager])

    private val zkClientAndConn = ZkUtils.createZkClientAndConnection(zkUrl, 30000, 30000);
    private val zkUtils = new ZkUtils(zkClientAndConn._1, zkClientAndConn._2, false)

    def readOffsets(topics: Seq[String], groupId: String): Map[TopicPartition, Long] = {
        val offsets = mutable.HashMap.empty[TopicPartition, Long]
        val partitionsForTopics = zkUtils.getPartitionsForTopics(topics)

        // /consumers/<groupId>/offsets/<topic>/<partition>
        partitionsForTopics.foreach(partitions => {
            val topic = partitions._1
            val groupTopicDirs = new ZKGroupTopicDirs(groupId, topic)

            partitions._2.foreach(partition => {
                val path = groupTopicDirs.consumerOffsetDir + "/" + partition
                try {
                    val data = zkUtils.readData(path)
                    if (data != null) {
                        offsets.put(new TopicPartition(topic, partition), data._1.toLong)
                        logger.info(
                            "Read offset - topic={}, partition={}, offset={}, path={}",
                            Seq[AnyRef](topic, partition.toString, data._1, path)
                        )
                    }
                } catch {
                    case ex: Exception =>
                        offsets.put(new TopicPartition(topic, partition), 0L)
                        logger.info(
                            "Read offset - not exist: {}, topic={}, partition={}, path={}",
                            Seq[AnyRef](ex.getMessage, topic, partition.toString, path)
                        )
                }
            })
        })

        offsets.toMap
    }

    def saveOffsets(offsetRanges: Seq[OffsetRange], groupId: String): Unit = {
        offsetRanges.foreach(range => {
            val groupTopicDirs = new ZKGroupTopicDirs(groupId, range.topic)
            val path = groupTopicDirs.consumerOffsetDir + "/" + range.partition
            zkUtils.updatePersistentPath(path, range.untilOffset.toString)
            logger.info(
                "Save offset - topic={}, partition={}, offset={}, path={}",
                Seq[AnyRef](range.topic, range.partition.toString, range.untilOffset.toString, path)
            )
        })
    }
}

  与接口不同的是，它还可以定义属性和方法的实现。

private[spark] class KafkaRDD[K, V](
    sc: SparkContext,
    val kafkaParams: ju.Map[String, Object],
    val offsetRanges: Array[OffsetRange],
    val preferredHosts: ju.Map[TopicPartition, String],
    useConsumerCache: Boolean
) extends RDD[ConsumerRecord[K, V]](sc, Nil) with Logging with HasOffsetRanges

private[spark] class DirectKafkaInputDStream[K, V](
    _ssc: StreamingContext,
    locationStrategy: LocationStrategy,
    consumerStrategy: ConsumerStrategy[K, V],
    ppc: PerPartitionConfig
  ) extends InputDStream[ConsumerRecord[K, V]](_ssc) with Logging with CanCommitOffsets {

不能对stream对象做transformation操作之后的结果进行强制转换（会直接报ClassCastException），因为RDD与DStream的类型都改变了。只有RDD或DStream的包含类型为ConsumerRecord才行。