From 6a6c1fc5c807ba4e8aba3e260537aa527ff5d46a Mon Sep 17 00:00:00 2001 From: Bryan Cutler Date: Thu, 10 Dec 2015 14:21:15 -0800 Subject: [SPARK-11713] [PYSPARK] [STREAMING] Initial RDD updateStateByKey for PySpark Adding ability to define an initial state RDD for use with updateStateByKey PySpark. Added unit test and changed stateful_network_wordcount example to use initial RDD. Author: Bryan Cutler Closes #10082 from BryanCutler/initial-rdd-updateStateByKey-SPARK-11713. --- examples/src/main/python/streaming/stateful_network_wordcount.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) (limited to 'examples/src/main/python/streaming/stateful_network_wordcount.py') diff --git a/examples/src/main/python/streaming/stateful_network_wordcount.py b/examples/src/main/python/streaming/stateful_network_wordcount.py index 16ef646b7c..f8bbc659c2 100644 --- a/examples/src/main/python/streaming/stateful_network_wordcount.py +++ b/examples/src/main/python/streaming/stateful_network_wordcount.py @@ -44,13 +44,16 @@ if __name__ == "__main__": ssc = StreamingContext(sc, 1) ssc.checkpoint("checkpoint") + # RDD with initial state (key, value) pairs + initialStateRDD = sc.parallelize([(u'hello', 1), (u'world', 1)]) + def updateFunc(new_values, last_sum): return sum(new_values) + (last_sum or 0) lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2])) running_counts = lines.flatMap(lambda line: line.split(" "))\ .map(lambda word: (word, 1))\ - .updateStateByKey(updateFunc) + .updateStateByKey(updateFunc, initialRDD=initialStateRDD) running_counts.pprint() -- cgit v1.2.3