标签：val partition Kafka topic Streaming offset path Spark
来源:大数据技术与架构 作者:王知无By 大数据技术与架构 场景描述： Kafka配合Spark Streaming是大数据领域常见的黄金搭档之一，主要是用于数据实时入库或分析。 为了应对可能出现的引起Streaming程序崩溃的异常情况，我们一般都需要手动管理好Kafka的offset，而不是让它自动提交，即需要将enable.auto.commit设为false。 只有管理好offset，才能使整个流式系统最大限度地接近exactly once语义。 关键词：offset Spark Streaming
Kafka+Spark Streaming主要用于实时流处理。 到目前为止，在大数据领域中是一种非常常见的架构。 Kafka在其中主要起着一个缓冲的作用，所有的实时数据都会经过kafka。 所以对kafka offset的管理是其中至关重要的一环。
我们一般都需要手动管理好Kafka的offset，而不是让它自动提交，即需要将
enable.auto.commit
设为false。
一但管理不善，就会到导致数据丢失或重复消费。
offset的管理方式
一个简单的流程如下：在Kafka DirectStream初始化时，取得当前所有partition的存量offset，以让DirectStream能够从正确的位置开始读取数据。
读取消息数据，处理并存储结果。
提交offset，并将其持久化在可靠的外部存储中。
保存offset的方式
Checkpoint：Spark Streaming的checkpoints是最基本的存储状态信息的方式，一般是保存在HDFS中。但是最大的问题是如果streaming程序升级的话，checkpoints的数据无法使用，所以几乎没人使用。
offset的三种管理方式：
自动提交offset：
-
enable.auto.commit=true。
一但consumer挂掉，就会导致数据丢失或重复消费。
offset不可控。
Kafka自身的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语义的）
自定义offset：
-
（推荐，采用这种方式，可以做到At-least-once语义）：
可以将offset存放在第三方储中，包括RDBMS、Redis、ZK、ES等。
若消费数据存储在带事务的组件上，则强烈推荐将offset存储在一起，借助事务实现 Exactly-once 语义。
示例
Kafka自身管理offset：在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
在Spark Streaming连接Kafka应用中使用Zookeeper来存储offsets也是一种比较可靠的方式。在这个方案中，Spark Streaming任务在启动时会去Zookeeper中读取每个分区的offsets。如果有新的分区出现，那么他的offset将会设置在最开始的位置。在每批数据处理完之后，用户需要可以选择存储已处理数据的一个offset或者最后一个offset。此外，新消费者将使用跟旧的Kafka 消费者API一样的格式将offset保存在ZooKeeper中。因此，任何追踪或监控Zookeeper中Kafka Offset的工具仍然生效的。
一个典型的工具类：
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)
)
})
}
}其他介质
Hbase、Redis甚至Mysql也经常被用作进行offset的存储。方式和上面类似，代码可以去网上搜一搜。
需要注意的点
特别需要注意，在转换过程中不能破坏RDD分区与Kafka分区之间的映射关系。亦即像map()/mapPartitions()这样的算子是安全的，而会引起shuffle或者repartition的算子，如reduceByKey()/join()/coalesce()等等都是不安全的。
对Dstream进行窗口操作后就不能手动提交offset。 因为保存offset需要HasOffsetRanges这个类。而
HasOffsetRanges
是
KafkaRDD
的一个trait，而
CanCommitOffsets
是
DirectKafkaInputDStream
的一个trait。
Scala Trait(特征) 相当于 Java 的接口，实际上它比接口还功能强大。
与接口不同的是，它还可以定义属性和方法的实现。
如下：
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才行。
欢迎点赞+收藏+转发朋友圈素质三连文章不错？点个【在看】吧！ ?
标签：val,partition,Kafka,topic,Streaming,offset,path,Spark 来源： https://blog.51cto.com/u_14222592/2893373
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。