欢迎关注Hadoop、Spark、Flink、Hive、Hbase、Flume等大数据资料分享微信公共账号:iteblog_hadoop
  1. 文章总数:961
  2. 浏览总数:11,479,756
  3. 评论:3873
  4. 分类目录:103 个
  5. 注册用户数:5841
  6. 最后更新:2018年10月17日
过往记忆博客公众号iteblog_hadoop
欢迎关注微信公众号:
iteblog_hadoop
大数据技术博客公众号bigdata_ai
大数据猿:
bigdata_ai

Spark 从 Kafka 读数并发问题

经常使用 Apache SparkKafka 读数的同学肯定会遇到这样的问题:某些 Spark 分区已经处理完数据了,另一部分分区还在处理数据,从而导致这个批次的作业总消耗时间变长;甚至导致 Spark 作业无法及时消费 Kafka 中的数据。为了简便起见,本文讨论的 Spark Direct 方式读取 Kafka 中的数据,这种情况下 Spark RDD 中分区和 Kafka 分区是一一对应的,更多的细节请参见官方文档,这里就不介绍。

那么有没有办法解决这个问题呢?我们先来看看社区是咋解决这个问题。

SPARK-22056 这个 issue 正好提出了这个问题,并给出了一种解决方案。


如果想及时了解Spark、Hadoop或者Hbase相关的文章,欢迎关注微信公共帐号:iteblog_hadoop

也就是修改了 KafkaRDD 类的 getPartitions 方法:
原实现:

override def getPartitions: Array[Partition] = {
  offsetRanges.zipWithIndex.map { case (o, i) =>
      val (host, port) = leaders(TopicAndPartition(o.topic, o.partition))
      new KafkaRDDPartition(i, o.topic, o.partition, o.fromOffset, o.untilOffset, host, port)
  }.toArray
}

修改后的实现:

override def getPartitions: Array[Partition] = {

  val subconcurrency = if (kafkaParams.contains("topic.partition.subconcurrency"))
    kafkaParams.getOrElse("topic.partition.subconcurrency","1").toInt
  else 1
  val numPartitions = offsetRanges.length

  val subOffsetRanges: Array[OffsetRange] = new Array[OffsetRange](subconcurrency * numPartitions)
  for (i <- 0 until numPartitions) {
    val offsetRange = offsetRanges(i)
    val step = (offsetRange.untilOffset - offsetRange.fromOffset) / subconcurrency

    var from = -1L
    var until = -1L

    for (j <- 0 until subconcurrency) {
      from = offsetRange.fromOffset + j * step
      until = offsetRange.fromOffset + (j + 1) * step -1
      if (j == subconcurrency) {
        until = offsetRange.untilOffset
      }
      subOffsetRanges(i * subconcurrency + j) = OffsetRange.create(offsetRange.topic, offsetRange.partition, from, until)
    }
  }

  subOffsetRanges.zipWithIndex.map{ case (o, i) =>
    val (host, port) = leaders(TopicAndPartition(o.topic, o.partition))
    new KafkaRDDPartition(i, o.topic, o.partition, o.fromOffset, o.untilOffset, host, port)
  }.toArray

}

这个方法的实现思想还是很简单的,就是通过设置 topic.partition.subconcurrency 参数,如果这个参数等于1,整个函数的执行效果和之前一样。但是如果这个参数大于1,则之前一个 Kafka 分区由一个 Spark 分区消费的数据变成由 topic.partition.subconcurrency 个 Spark 分区去消费,每个 Spark 分区消费的数据量相等。这个无疑会加快 Kafka 数据的消费,但是这种方法的问题也很明显:

  • 如果数据的顺序很重要,这种方法会存在乱序的问题。
  • Spark 设计的 KafkaRDD 目的是让 Kafka Partition 和 Spark RDD Partition 一一对应,这样可以保证同一个分区里面的数据顺序,但是这种方法实现变成了 Kafka Partition 和 Spark RDD Partition 一对多的关系,无疑破坏了官方的原有设计。

基于这些问题,社区的 jerryshao 和 koeninger 大佬对此并不看好:(PR 19274

jerryshao: Hi @loneknightpy , think a bit on your PR, I think this can also be done in the user side. User could create several threads in one task (RDD#mapPartitions) to consume the records concurrently, so such feature may not be so necessary to land in Spark's code. What do you think?

koeninger: Search Jira and the mailing list, this idea has been brought up multiple times. I don't think breaking fundamental assumptions of Kafka (one consumer thread per group per partition) is a good idea.

到目前为止,上述 PR 被关闭,而且 SPARK-22056 一直处于 IN PROGRESS 状态,我猜这个最后可能也会被关闭掉。

那除了上面实现,我们还有其他实现吗?当然有,我们可以在处理数据之前通过 repartitioncoalease 对数据进行重分区:

val kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, 
  StringDecoder](streamingContext, kafkaParams, topics).repartition(xxx).mapPartition(xxx)

这种方法的好处是,对同一类型的数据,先后顺序是不会乱的,因为同一类型的数据经过重分区最后还是会分发到同一个分区里面的。

但是这个方法的使用前提是数据重分区+后续处理的时间比没有重分区直接处理数据的时间要短,否则重分区的开销过大导致总的处理时间过长那就没意义了。

当然,我们可以可以通过在 RDD#mapPartitions 里面创建多个线程来处理同一个 RDD 分区里面的数据。

但是上面两种方法无法解决 Kafka 端数据倾斜导致的数据处理过慢的问题(也就是有些分区数据量相比其他分区大很多,光是从这些分区消费数据的时间就比其他分区要长很多)。针对这种情况,我们需要考虑 Kafka 分区设置是否合理?是否能够通过修改 Kafka 分区的实现来解决数据倾斜的问题。

如果不是 Kafka 数据倾斜导致的数据处理过慢的问题,而是所有 Kafka 分区的整体数据量就比较大,那这种情况我们可以考虑是否可以增加 Kafka 分区数?是否需要增加 Spark 的处理资源等。建议最好还是别使用多个线程处理同一个 Kafka 分区里面的数据。

本博客文章除特别声明,全部都是原创!
转载本文请加上:转载自过往记忆(https://www.iteblog.com/)
本文链接: 【Spark 从 Kafka 读数并发问题】(https://www.iteblog.com/archives/2419.html)
喜欢 (7)
分享 (0)
发表我的评论
取消评论

表情
本博客评论系统带有自动识别垃圾评论功能,请写一些有意义的评论,谢谢!