aboutsummaryrefslogtreecommitdiff
path: root/examples/src/main/python/streaming/recoverable_network_wordcount.py
diff options
context:
space:
mode:
Diffstat (limited to 'examples/src/main/python/streaming/recoverable_network_wordcount.py')
-rw-r--r--examples/src/main/python/streaming/recoverable_network_wordcount.py11
1 files changed, 6 insertions, 5 deletions
diff --git a/examples/src/main/python/streaming/recoverable_network_wordcount.py b/examples/src/main/python/streaming/recoverable_network_wordcount.py
index fc6827c82b..ac91f0a06b 100644
--- a/examples/src/main/python/streaming/recoverable_network_wordcount.py
+++ b/examples/src/main/python/streaming/recoverable_network_wordcount.py
@@ -35,6 +35,7 @@
checkpoint data exists in ~/checkpoint/, then it will create StreamingContext from
the checkpoint data.
"""
+from __future__ import print_function
import os
import sys
@@ -46,7 +47,7 @@ from pyspark.streaming import StreamingContext
def createContext(host, port, outputPath):
# If you do not see this printed, that means the StreamingContext has been loaded
# from the new checkpoint
- print "Creating new context"
+ print("Creating new context")
if os.path.exists(outputPath):
os.remove(outputPath)
sc = SparkContext(appName="PythonStreamingRecoverableNetworkWordCount")
@@ -60,8 +61,8 @@ def createContext(host, port, outputPath):
def echo(time, rdd):
counts = "Counts at time %s %s" % (time, rdd.collect())
- print counts
- print "Appending to " + os.path.abspath(outputPath)
+ print(counts)
+ print("Appending to " + os.path.abspath(outputPath))
with open(outputPath, 'a') as f:
f.write(counts + "\n")
@@ -70,8 +71,8 @@ def createContext(host, port, outputPath):
if __name__ == "__main__":
if len(sys.argv) != 5:
- print >> sys.stderr, "Usage: recoverable_network_wordcount.py <hostname> <port> "\
- "<checkpoint-directory> <output-file>"
+ print("Usage: recoverable_network_wordcount.py <hostname> <port> "
+ "<checkpoint-directory> <output-file>", file=sys.stderr)
exit(-1)
host, port, checkpoint, output = sys.argv[1:]
ssc = StreamingContext.getOrCreate(checkpoint,