如何使用 Pyspark Streaming 模块实现 RabbitMQ 消费者？

Question

我有一个 Apache Spark 集群和一个 RabbitMQ 代理，我想使用 pyspark.streaming 模块来使用消息并计算一些指标。

问题是我只找到this package，但是用Java和Scala实现的。除此之外，我没有在 Python 中找到任何示例或桥接实现。

我有一个使用 Pika 实现的消费者，但我不知道如何将有效负载传递给我的 StreamingContext。

Answer 1

此解决方案使用 Spark Streaming 中的 pika asynchronous consumer example 和 socketTextStream 方法

1. 下载示例并将其保存为.py 文件
2. 修改文件以使用您自己的 RabbitMQ 凭据和连接参数。就我而言，我必须修改 Consumer 类

3. 在if __name__ == '__main__': 下，我们需要打开一个套接字，其中HOST 和PORT 对应于您与Spark Streaming 的TCP 连接。我们必须将方法 sendall 从套接字保存到一个变量中，并将其传递给 Consumer 类

   with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
     s.bind((HOST, PORT))
     s.listen(1)
     conn, addr = s.accept()
     dispatcher = conn.sendall #assigning sendall to dispatcher variable
   consumer = Consumer(dispatcher)
   try:
     consumer.run()
   except Exception as e:
     consumer.stop()
     s.close()
   

4. 修改Consumer中的__init__方法传递dispatcher

   def __init__(self,dispatcher):
     self._connection = None
     self._channel = None
     self._closing = False
     self._consumer_tag = None
     self._url = amqp_url
     #new code
     self._dispatcher = dispatcher
   

5. 在Consumer内部的方法on_message中我们调用self._dispatcher来发送AMQP消息的body

   def on_message(self, unused_channel, basic_deliver, properties, body):
     self._channel.basic_ack(basic_deliver.delivery_tag)
     try:
       # we need an '\n' at the each row Spark socketTextStream
       self._dispatcher(bytes(body.decode("utf-8")+'\n',"utf-8"))
     except Exception as e:
       raise
   

6. 在 Spark 中，将 ssc.socketTextStream(HOST, int(PORT)) 与 HOST 和 PORT 对应于我们的 TCP 套接字。 Spark 将管理连接

7. 先运行消费者，然后运行 ​​Spark 应用程序

最后的评论：

• 尝试在不同的机器上运行您的消费者，而不是您的 Spark 机器
• 任何超过 10000 的端口都可以。不要让内核随意打开一些端口
• 平台：Linux Debian 7 和 8，以及 Ubuntu 14.04 和 16.04
• 鼠兔版本 0.10.0
• Python 版本 3.5.2
• Spark 版本 1.6.1、1.6.2 和 2.0.0