为 spark kafka 消费者动态更新主题列表答案

【问题标题】：Dynamically update topics list for spark kafka consumer为 spark kafka 消费者动态更新主题列表
【发布时间】：2017-01-28 18:39:39
【问题描述】：

是否可以在 spark-kafka 消费者中动态更新主题列表？

我有一个使用 spark-kafka 消费者的 Spark Streaming 应用程序。最初说我有 spark-kakfa 消费者监听主题：[“test”]，过了一段时间我的主题列表更新为 [“test”，“testNew”]。现在有办法更新 spark-kafka 消费者主题列表并要求 spark-kafka 消费者在不停止 sparkStreaming 应用程序或 sparkStreaming 上下文的情况下使用更新的主题列表的数据

【问题讨论】：

标签： apache-spark apache-kafka spark-streaming

【解决方案1】：

是否可以在 spark-kafka 消费者中动态更新主题列表

没有。使用KafkaUtils 初始化 kafka 流后，接收器和无接收器方法都是固定的。由于 DAG 是固定的，因此您无法随时传递新主题。

如果你想动态读取，或许可以考虑批量 k 迭代调度的作业，可以动态读取主题并从中创建RDD。

另一种解决方案是使用一种技术，让您可以灵活地控制消费，例如 Akka Streams。

【讨论】：

【解决方案2】：

正如 Yuval 所说，这是不可能的，但如果您知道您从 Kafka 处理的数据的结构/格式是什么，那么可能会有一种解决方法。

例如，

如果您的流应用程序正在侦听主题 ["test","testNew"]
在您想要添加名为 [test4] 的新主题的行下，作为一种解决方法，您可以简单地向其中包含的主题添加一个唯一键并将其传递给现有主题。
设计您的流应用程序，以便根据您添加到该 test2 数据的键来识别/过滤数据

【讨论】：

【解决方案3】：

您可以使用基于线程的方法
1. 使用包含主题列表的任何数据结构定义缓存
2. 在此缓存中添加元素的方法
3. 你必须对 A 和 B 进行分类，其中 B 具有所有与 spark 相关的逻辑
4 A 类是长期运行的工作，从 A 你调用 B ，每当有新主题时，你只是用 B 产生新线程

【讨论】：

我目前正在使用类似的方法，但这有很多复杂性，例如每次我需要更新主题列表时都需要优雅地停止流式传输上下文。这是一个异步过程，就停止所需的时间而言是不可预测的。所有这一切，而我将无法处理传入的数据流。流式上下文需要停止、启动和恢复计算。
@rohith-yeravothula 你有没有找到任何替代解决方案，我认为只能使用带有演员系统的 Akka 流。我尝试了 SubscribePattern 但这只是启动期间的过滤器主题，而不是在 DAG 期间添加主题并安排流。

【解决方案4】：

我建议从最新的Spark-Kafka integration (0.10) API 版本尝试ConsumerStrategies.SubscribePattern。

看起来像：

KafkaUtils.createDirectStream(
mySparkStreamingContext,
PreferConsistent,
SubscribePattern("test.*".r.pattern, myKafkaParamsMap))

【讨论】：

我尝试了同样的方法，它不会动态选择主题意味着是的，当您启动流时，它将使用正则表达式匹配所有主题（就像过滤器一样）并为这些主题创建流。我们正在寻找一个解决方案，当流已经运行时，我可以动态添加新主题。看起来这是不可能的，因为 Spark 集群的工作方式可以重新安排作业和流式传输 DAG。