使用pika的python中的SparkStreaming,RabbitMQ和MQTT

Question

只是为了让事情变得棘手,我想使用来自rabbitMQ队列的消息.现在我知道有一个针对兔子的MQTT插件(https://www.rabbitmq.com/mqtt.html).

但是,我似乎无法在Spark消耗由pika生成的消息的情况下进行示例工作.

例如,我在这里使用简单的wordcount.py程序(https://spark.apache.org/docs/1.2.0/streaming-programming-guide.html),看看我是否可以在下面看到一个消息生产者办法:

import sys
import pika
import json
import future
import pprofile

def sendJson(json):

  connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
  channel = connection.channel()

  channel.queue_declare(queue='analytics', durable=True)
  channel.queue_bind(exchange='analytics_exchange',
                       queue='analytics')

  channel.basic_publish(exchange='analytics_exchange', routing_key='analytics',body=json)
  connection.close()

if __name__ == "__main__":
  with open(sys.argv[1],'r') as json_file:
    sendJson(json_file.read())

Sparkstreaming 消费者如下:

import sys
import operator

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.mqtt import MQTTUtils

sc = SparkContext(appName="SS")
sc.setLogLevel("ERROR")
ssc = StreamingContext(sc, 1)
ssc.checkpoint("checkpoint")
#ssc.setLogLevel("ERROR")


#RabbitMQ

"""EXCHANGE = 'analytics_exchange'
EXCHANGE_TYPE = 'direct'
QUEUE = 'analytics'
ROUTING_KEY = 'analytics'
RESPONSE_ROUTING_KEY = 'analytics-response'
"""


brokerUrl = "localhost:5672" # "tcp://iot.eclipse.org:1883"
topic = "analytics"

mqttStream = MQTTUtils.createStream(ssc, brokerUrl, topic)
#dummy functions - nothing interesting...
words = mqttStream.flatMap(lambda line: line.split(" "))
pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)

wordCounts.pprint()
ssc.start()
ssc.awaitTermination()

但是,与简单的wordcount示例不同,我无法使其工作并出现以下错误:

16/06/16 17:41:35 ERROR Executor: Exception in task 0.0 in stage 7.0 (TID 8)
java.lang.NullPointerException
    at org.eclipse.paho.client.mqttv3.MqttConnectOptions.validateURI(MqttConnectOptions.java:457)
    at org.eclipse.paho.client.mqttv3.MqttAsyncClient.<init>(MqttAsyncClient.java:273)

所以我的问题是,MQTTUtils.createStream(ssc, brokerUrl, topic)听取队列的设置应该是什么,以及是否有更全面的例子以及这些例子如何映射到rabbitMQ.

我正在运行我的消费者代码: ./bin/spark-submit ../../bb/code/skunkworks/sparkMQTTRabbit.py

我已按照以下注释建议的TCP参数更新了生产者代码:

url_location = 'tcp://localhost'
url = os.environ.get('', url_location)
params = pika.URLParameters(url)
connection = pika.BlockingConnection(params)

和火花流:

brokerUrl = "tcp://127.0.0.1:5672"
topic = "#" #all messages

mqttStream = MQTTUtils.createStream(ssc, brokerUrl, topic)
records = mqttStream.flatMap(lambda line: json.loads(line))
count = records.map(lambda rec: len(rec))
total = count.reduce(lambda a, b: a + b)
total.pprint()

Answer 1

看起来您使用了错误的端口号。假如说：

• 您有一个使用默认设置运行的 RabbitMQ 本地实例，并且您已启用 MQTT 插件 ( rabbitmq-plugins enable rabbitmq_mqtt) 并重新启动了 RabbitMQ 服务器
• spark-streaming-mqtt执行spark-submit/时包含pyspark（使用packages或jars/ driver-class-path）

您可以使用 TCP 与tcp://localhost:1883. 您还必须记住 MQTT 正在使用amq.topic.

快速启动：

• Dockerfile使用以下内容创建：

  FROM rabbitmq:3-management
  
  RUN rabbitmq-plugins enable rabbitmq_mqtt
  

• 构建 Docker 镜像：

  docker build -t rabbit_mqtt .
  

• 启动映像并等待服务器准备就绪：

  docker run -p 15672:15672 -p 5672:5672 -p 1883:1883 rabbit_mqtt 
  

• producer.py使用以下内容创建：

  import pika
  import time 
  
  
  connection = pika.BlockingConnection(pika.ConnectionParameters(
      host='localhost'))
  channel = connection.channel()
  channel.exchange_declare(exchange='amq.topic',
                   type='topic', durable=True)
  
  for i in range(1000):
      channel.basic_publish(
          exchange='amq.topic',  # amq.topic as exchange
          routing_key='hello',   # Routing key used by producer
          body='Hello World {0}'.format(i)
      )
      time.sleep(3)
  
  connection.close()
  

• 开始制作人

  python producer.py
  

  并访问管理控制台http://127.0.0.1:15672/#/exchanges/%2F/amq.topic

  查看收到的消息。

• consumer.py使用以下内容创建：

  from pyspark import SparkContext
  from pyspark.streaming import StreamingContext
  from pyspark.streaming.mqtt import MQTTUtils
  
  sc = SparkContext()
  ssc = StreamingContext(sc, 10)
  
  mqttStream = MQTTUtils.createStream(
      ssc, 
      "tcp://localhost:1883",  # Note both port number and protocol
      "hello"                  # The same routing key as used by producer
  )
  mqttStream.count().pprint()
  ssc.start()
  ssc.awaitTermination()
  ssc.stop()
  

• 下载依赖项（将 Scala 版本调整为用于构建 Spark 和 Spark 版本的版本）：

  mvn dependency:get -Dartifact=org.apache.spark:spark-streaming-mqtt_2.11:1.6.1
  

• 确保SPARK_HOME并PYTHONPATH指向正确的目录。

• 提交consumer.py（像以前一样调整版本）：

  spark-submit --packages org.apache.spark:spark-streaming-mqtt_2.11:1.6.1 consumer.py
  

如果您按照所有步骤操作，您应该会在 Spark 日志中看到 Hello world 消息。