如何使用火花流区分在 kafka 中收到的主题答案

【问题标题】：How to differentiate topics received in kafka using spark streaming如何使用火花流区分在 kafka 中收到的主题
【发布时间】：2016-01-11 15:45:25
【问题描述】：

我正在使用以下代码从 kafka 获取消息

scala 代码：

val lines: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc,
 zookeeperQuorum, consumerGroup, topicMap)
lines.print(10)

这是我的示例生产者代码。

    from kafka import SimpleProducer, KafkaClient
    import time
    # To send messages synchronously
    kafka = KafkaClient(serverip+':'+port)
    producer = SimpleProducer(kafka)
    kafka.ensure_topic_exists('test')

    kafka.ensure_topic_exists('test1')

    while(1):
     print "sending message "
     producer.send_messages(b'test', 'test,msg')
     time.sleep(2)
     producer.send_messages(b'test1', 'test1,msg')
     time.sleep(2)

我的流媒体接收器打印

(null,'test,msg')
(null,'test1,msg')

问题：

1) How can I differentiate msg per topic level without actually
decoding the message ?

2) Why it is giving me null in the output ? From the documentation
it says key,value tuple. How can I create key,value tuple kind of
message ?

编辑：使用 keyedProducer

kafka = KafkaClient(serverip+':'+port)
producer = KeyedProducer(kafka)

kafka.ensure_topic_exists('test2')

while(1):
   print "sending msg "
   producer.send_messages(b'test2',b'key1','msg')
   time.sleep(2)

这是给我的错误

raise PartitionUnavailableError("%s not available" % str(key))                                                                                                                            
kafka.common.PartitionUnavailableError: TopicAndPartition(topic='test2', partition='key1') not available

【问题讨论】：

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

【解决方案1】：

对于 #1，最简单的方法是为每个主题设置单独的流，如果在任何时候您需要将它们组合在一起并且它们具有相同的结构 - 您可以将它们合并

对于#2，您是否尝试过使用KeyedProducer？

来自以上链接的片段：

producer = KeyedProducer(kafka)
producer.send_messages(b'my-topic', b'key1', b'some message')
producer.send_messages(b'my-topic', b'key2', b'this methode')

【讨论】：

我已经尝试过 keyproducer ，但是我得到了这个错误 kafka.common.PartitionUnavailableError: TopicAndPartition(topic='test2', partition='test1') not available 如何创建分区？
对于#1，那么我需要为接收器分配那么多核心对吗？
在创建每个主题时至少创建一个（或多个分区）。但是默认情况下，您应该可以使用所有默认值进行测试（在创建主题期间它将创建 1 个分区），并且 api 应该默认使用 hashPartitioner。异常看起来像其他地方出了问题。你有创建 KeyedProducer 和发送消息的代码的 sn-p 吗？
对于核心数量 - 是的，对于每个额外的接收器，您都需要额外的核心。一般建议是核心数量多于接收器数量
我尝试了您的 KeyedProducer 代码，它在我的本地 kafka 实例上运行良好，带有 python 2.7 和最新版本的 python-kafka。您确定该主题存在吗？你真的在卡夫卡看到这个话题吗？如果它不存在并且主题的自动创建被禁用 - 它可能会产生类似的错误。

【解决方案2】：

对于问题号。 1 你可以使用这个签名

def
createDirectStream[K, V, KD <: Decoder[K], VD <: Decoder[V], R]
(ssc: StreamingContext, kafkaParams: Map[String, String], fromOffsets: Map[TopicAndPartition, Long],
 messageHandler: (MessageAndMetadata[K, V]) ⇒ R): InputDStream[R]

这将使您能够访问 MessageAndMetadata 类，该类包含主题名称以及一些其他元数据，例如分区号和消息偏移量。例如

KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, Map[String, String]](
  ssc,
  Map("metadata.broker.list" -> "localhost:9092"),
  topics,
  (mm: MessageAndMetadata[String, String]) => Map(mm.topic -> mm.message))

然后你可以在地图键上进行模式匹配来做任何你想做的事情

【讨论】：