问题描述

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

Just to make things tricky, I'd like to consume messages from the rabbitMQ queue. Now I know there is a plugin for MQTT on rabbit (https://www.rabbitmq.com/mqtt.html).

但是，我似乎无法举例说明 Spark 使用从 pika 生成的消息.

However I cannot seem to make an example work where Spark consumes a message that has been produced from pika.

例如，我在这里使用简单的 wordcount.py 程序(https://spark.apache.org/docs/1.2.0/streaming-programming-guide.html) 查看是否可以通过以下方式看到消息producer:

For example I am using the simple wordcount.py program here (https://spark.apache.org/docs/1.2.0/streaming-programming-guide.html) to see if I can I see a message producer in the following way:

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())

火花流消费者如下:

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()

但与简单的字数统计示例不同，我无法使其正常工作并出现以下错误:

However unlike the simple wordcount example, I cannot get this to work and get the following error:

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上.

So my questions are, what should be the settings in terms of MQTTUtils.createStream(ssc, brokerUrl, topic) to listen into the queue and whether there are any more fuller examples and how these map onto those of rabbitMQ.

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

I am running my consumer code with: ./bin/spark-submit ../../bb/code/skunkworks/sparkMQTTRabbit.py

我已按照一条评论的建议使用 TCP 参数更新了生产者代码如下:

I have updated the producer code as follows with TCP parameters as suggested by one comment:

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()

推荐答案

你好像使用了错误的端口号.假设:

It looks like you are using wrong port number. Assuming that:

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

• you have a local instance of RabbitMQ running with default settings and you've enabled MQTT plugin (rabbitmq-plugins enable rabbitmq_mqtt) and restarted RabbitMQ server
• included spark-streaming-mqtt when executing spark-submit / pyspark (either with packages or jars / driver-class-path)

您可以通过 tcp://localhost:1883 使用 TCP 进行连接.您还必须记住 MQTT 使用的是 amq.topic.

you can connect using TCP with tcp://localhost:1883. You have to also remember that MQTT is using amq.topic.

快速入门:

• 使用以下内容创建 Dockerfile:

FROM rabbitmq:3-management

RUN rabbitmq-plugins enable rabbitmq_mqtt

docker build -t rabbit_mqtt .

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

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

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()

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

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

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

If you followed all the steps you should see Hello world messages in the Spark log.

这篇关于使用 pika 在 python 中的 SparkStreaming、RabbitMQ 和 MQTT的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持跟版网！