title: SparkStreaming, RabbitMQ and MQTT in python using pika
body: <p>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 (<a href="https://www.rabbitmq.com/mqtt.html">https://www.rabbitmq.com/mqtt.html</a>). </p>
  
  <p>However I cannot seem to make an example work where Spark consumes a message that has been produced from pika.</p>
  
  <p>For example I am using the simple wordcount.py program here (<a href="https://spark.apache.org/docs/1.2.0/streaming-programming-guide.html">https://spark.apache.org/docs/1.2.0/streaming-programming-guide.html</a>) to see if I can I see a message <strong>producer</strong> in the following way:</p>
  
  <pre><code>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())
  </code></pre>
  
  <p>The sparkstreaming <strong>consumer</strong> is the following:</p>
  
  <pre><code>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()
  </code></pre>
  
  <p>However unlike the simple wordcount example, I cannot get this to work and get the following error:</p>
  
  <pre><code>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.&lt;init&gt;(MqttAsyncClient.java:273)
  </code></pre>
  
  <p>So my questions are, what should be the settings in terms of <code>MQTTUtils.createStream(ssc, brokerUrl, topic)</code> to listen into the queue and whether there are any more fuller examples and how these map onto those of rabbitMQ.</p>
  
  <p>I am running my consumer code with: <code>./bin/spark-submit ../../bb/code/skunkworks/sparkMQTTRabbit.py</code></p>
  
  <p>I have updated the producer code as follows with TCP parameters as suggested by one comment:</p>
  
  <pre><code>url_location = 'tcp://localhost'
  url = os.environ.get('', url_location)
  params = pika.URLParameters(url)
  connection = pika.BlockingConnection(params)
  </code></pre>
  
  <p>and the spark streaming as:</p>
  
  <pre><code>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()
  </code></pre>