spark-streaming-kafka-0-10：如何限制 Spark 分区的数量答案

【问题标题】：spark-streaming-kafka-0-10: How to limit number of Spark partitionsspark-streaming-kafka-0-10：如何限制 Spark 分区的数量
【发布时间】：2017-07-10 03:07:08
【问题描述】：

是否可以使用 spark-streaming-kafka-0-10 库配置 Spark 以使用单个任务读取多个 Kafka 分区或整个 Kafka 主题，而不是创建不同的 Spark 任务每个 Kafka 分区都可用？

请原谅我对这些技术的粗略理解；我想我对 Spark 和 Kafka 还是很陌生。架构和设置主要是为了探索和了解这些技术如何协同工作。

我有四个虚拟主机，一个带有一个 Spark master，每个带有一个 Spark worker。其中一台主机还运行基于Spotify's Docker image 的Kafka 代理。每台主机都有四个内核和大约 8 GB 的未使用 RAM。

Kafka broker有206个topic，每个topic有10个partition。因此，共有 2,060 个分区可供应用程序读取。

我正在使用 spark-streaming-kafka-0-10 库（目前处于试验阶段）从 Spark Streaming 作业订阅 Kafka 中的主题。我正在使用SubscribePattern 类订阅 Spark 的所有 206 个主题：

val stream = KafkaUtils.createDirectStream[String, String](
  ssc,
  PreferConsistent,
  SubscribePattern[String, String](Pattern.compile("(pid\\.)\\d+"),
  kafkaParams)
)

当我将此作业提交给 Spark 主服务器时，似乎启动了 16 个执行器，集群中的每个核心一个。看起来每个 Kafka 分区都有自己的任务，总共有 2,060 个任务。我认为我的 16 个执行器集群在处理这么多任务时遇到了麻烦，因为在完成 1500 到 1800 个任务之间的不同点上，这项工作不断失败。

我发现 a tutorial by Michael Noll from 2014 使用 spark-streaming-kafka-0-8 库来控制每个主题的消费者线程数：

val kafkaParams: Map[String, String] = Map("group.id" -> "terran", ...)

val consumerThreadsPerInputDstream = 3
val topics = Map("zerg.hydra" -> consumerThreadsPerInputDstream)
val stream = KafkaUtils.createStream(ssc, kafkaParams, topics, ...)

【问题讨论】：

标签： scala apache-spark apache-kafka spark-streaming kafka-consumer-api

【解决方案1】：

是否可以使用 spark-streaming-kafka-0-10 库配置 Spark，以通过单个任务读取多个 Kafka 分区或整个 Kafka 主题，而不是为每个可用的 Kafka 分区创建不同的 Spark 任务？

您可以通过在流上调用repartition 来更改生成的分区数量，但这样您就失去了 Kafka 和 RDD 分区之间的 1:1 对应关系。

Kafka 分区生成的任务数量与您有 16 个执行器这一事实无关。执行器的数量取决于您的设置和您使用的资源管理器。

Kafka 分区和 RDD 分区之间存在 1:1 映射，使用直接流 API，每个 executor 将从 Kafka 获取这些分区的子集以使用并处理每个分区独立且可以自己计算的过程。这与基于接收器的 API 不同，后者在任意执行器上创建单个接收器并通过节点上的线程使用数据本身。

如果您有 206 个主题和每个 10 个分区，那么您最好有一个大小合适的集群，它可以处理生成的任务的负载。您可以控制每个分区生成的最大消息数，但您可以更改分区数，除非您愿意调用repartition 转换的混洗效果。

【讨论】：

感谢您提供详细信息。你触及了一个我实际上还没有完全弄清楚的问题。你的意思是在新的 Direct API 中，对应于一个分区（Kafka Partition / RDD Partition）的任务，既读取数据，又处理数据。我一直想知道这是为了调整我的执行程序以及最终我的节点和集群的大小。
@Maatdeamon 是的。当执行程序被分配一个要操作的分区时，驱动程序会分配要读取的每个分区的偏移量。
我认为这个问题需要他自己的帖子。我还是有点疑惑。请您看看我的帖子stackoverflow.com/questions/45526554/… 并尝试在那里回答？
Spark-Streaming Kafka Direct Streaming API & Parallelism
@MaatDeamon 我现在不在我的电脑上。如果您还没有找到答案，明天再给它一个机会。

【解决方案2】：

第二种方法最适合您的要求。只有你必须设置 consumerThreadsPerInputDstream = 1。因此每次读取操作只会创建一个线程，因此每个集群将涉及单个机器。

【讨论】：