在中使用“topics”参数的正确方法是什么kafkatils.createstream()?

Question

我在python中使用spark streaming和kafka，并松散地跟随this post，但我对kafkatils.createStream（）早期提到的功能。在

documentation无法明确解释主题词典的影响。但我有一种怀疑，我之所以这么认为，是因为我对卡夫卡如何运作的了解并不可靠，答案是显而易见的。在

我知道它应该是这样的字典：{"topic.name": 1}，我可以照搬文档，说这意味着创建的流将从单个分区消耗。在

所以我想我只是想澄清一下这个特殊函数的用法，还有我对卡夫卡概念的理解。我们将使用以下示例：

假设我定义了一个topic my.topic，它有3个分区，其传入消息在一个键上被拆分，比如说一个userid。在

如果我像这样初始化流：

from pyspark.streaming.kafka import KafkaUtils

kafkaStream = KafkaUtils.createStream(
    ssc, 
    'kafka:2181', 
    'consumer-group-name', 
    {'my.topic':1}
)  

我认为这个流将只从一个分区消费，因此不会看到进入my.topic的所有消息，这是正确的吗？只会看到从另一个用户标识发送到另一个用户标识的消息中的一个？在

我的问题是：

1. 如何正确设置此参数以使用发送到my.topic的所有消息？

   我的直觉是，我只需将topics参数设置为{'my.topic': 3}，那么我的问题就变成：

2. 为什么我会使用小于分区总数的数字？

   我的直觉告诉我，这是一个问题，你正在做的工作是如何“原子”的。例如，如果我只是转换数据（比如，从一个CSV到一个JSON文档列表或其他内容）然后将上述3个流（每个流的主题参数设置为{'my.topic': 1}）和同一个消费者组的所有部分，通过从每个分区启用并行消费将是有益的，因为不需要共享关于所消费的每条消息的信息。在

   同时，如果我计算的是整个主题的实时指标，即带有过滤器的时间窗口平均值，等等，我很难找到一种方法来实现这样的东西，而不设置{'my.topic': 3}，或者如果它是一个总和，然后对一个用户组内的每个分量信号进行稍微复杂的下游处理，即Sum1+Sum2+Sum3=TotalSum

   不过，我的知识又完全处于与卡夫卡和星火玩耍的“羽翼未丰”阶段。

3. 有没有办法让createStream（）在不提前知道有多少个分区的情况下使用所有分区？像{'my.topic': -1}之类的？

4. 单个流中是否可以指定多个主题？一、 E.{'my.topic': 1, 'my.other.topic': 1}

我真的很讨厌这个问题的答案仅仅是“是的，你的直觉是正确的”。最好的情况是有人告诉我我误解了一切，并让我直截了当。所以请…这样做！在

Answer 1

这是卡夫卡星火整合页面上提到的。在

val kafkaStream = KafkaUtils.createStream(streamingContext, [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

在kafkatils.createStream将创建一个接收器并使用Kafka主题。在

选项“每个主题要使用的Kafka分区数”表示这个接收器将并行读取多少个分区。在

例如，您有一个名为“Topic1”的主题，其中有两个分区，并且您提供了选项“Topic1”：1，那么Kafka接收器将一次读取一个分区[它最终将读取所有分区，但每次将读取一个分区]。这样做的原因是在分区中读取消息，并保留数据写入主题的顺序。在

例如，Topic1的partition1包含消息{1,11,21,31,41}，partition2包含消息{2,12,22,32,42}，那么使用上述设置进行读取将生成一个流，如{1,11,21,31,41,2,12,22,32,42}。每个分区中的消息是分开读取的，因此它不会混合在一起。在

如果您以'Topic1'：2提供选项，则接收器将一次读取2个分区，并且这些分区内的消息将混合在一起。对于上面开始的同一个示例，具有“Topic1”：2的接收器将生成类似于{1,2,11,12,21,22….}

可以将其视为接收器可以对给定主题分区执行的并行读取数。在

5。一个流中是否可以指定多个主题？ 是的，你可以。在

Answer 2

只需指定不带分区的主题，就可以得到该主题中的所有消息，无论主题中有多少个分区。在

您只需查看示例代码：https://github.com/apache/spark/blob/v2.2.1/examples/src/main/python/streaming/direct_kafka_wordcount.py#L48