diff options
Diffstat (limited to 'examples/src/main/python/streaming/mqtt_wordcount.py')
-rw-r--r-- | examples/src/main/python/streaming/mqtt_wordcount.py | 59 |
1 files changed, 0 insertions, 59 deletions
diff --git a/examples/src/main/python/streaming/mqtt_wordcount.py b/examples/src/main/python/streaming/mqtt_wordcount.py deleted file mode 100644 index abf9c0e21d..0000000000 --- a/examples/src/main/python/streaming/mqtt_wordcount.py +++ /dev/null @@ -1,59 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -""" - A sample wordcount with MqttStream stream - Usage: mqtt_wordcount.py <broker url> <topic> - - To run this in your local machine, you need to setup a MQTT broker and publisher first, - Mosquitto is one of the open source MQTT Brokers, see - http://mosquitto.org/ - Eclipse paho project provides number of clients and utilities for working with MQTT, see - http://www.eclipse.org/paho/#getting-started - - and then run the example - `$ bin/spark-submit --jars \ - external/mqtt-assembly/target/scala-*/spark-streaming-mqtt-assembly-*.jar \ - examples/src/main/python/streaming/mqtt_wordcount.py \ - tcp://localhost:1883 foo` -""" - -import sys - -from pyspark import SparkContext -from pyspark.streaming import StreamingContext -from pyspark.streaming.mqtt import MQTTUtils - -if __name__ == "__main__": - if len(sys.argv) != 3: - print >> sys.stderr, "Usage: mqtt_wordcount.py <broker url> <topic>" - exit(-1) - - sc = SparkContext(appName="PythonStreamingMQTTWordCount") - ssc = StreamingContext(sc, 1) - - brokerUrl = sys.argv[1] - topic = sys.argv[2] - - lines = MQTTUtils.createStream(ssc, brokerUrl, topic) - counts = lines.flatMap(lambda line: line.split(" ")) \ - .map(lambda word: (word, 1)) \ - .reduceByKey(lambda a, b: a+b) - counts.pprint() - - ssc.start() - ssc.awaitTermination() |