<jupyter_start><jupyter_text>create dstream from a port<jupyter_code># Create a local StreamingContext with two working thread and batch interval of 1 second
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, batchDuration=5)
# Create a DStream that will connect to hostname:port, like localhost:9999
lines = ssc.socketTextStream("localhost", 9999)
# Split each line into words
words = lines.flatMap(lambda line: line.split(" "))
# Count each word in each batch
pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)

# Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.pprint()
# install netcat
# !yum install nc

# send data to the port 
# nc -l 9999
ssc.start()             # Start the computation
ssc.awaitTermination()  # Wait for the computation to terminate<jupyter_output><empty_output><jupyter_text>listen to a folder Create an input stream that monitors a Hadoop-compatible file system for new files and reads them as text files. Files must be wrriten to the monitored directory by “moving” them from another location within the same file system. File names starting with . are ignored.<jupyter_code>!hadoop fs -mkdir /input
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, batchDuration=5)
lines = ssc.textFileStream("/input")
words = lines.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()<jupyter_output><empty_output><jupyter_text>Listen to Kafka<jupyter_code>from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, batchDuration=5)
#!kafka-topics --create --topic t1 --zookeeper 0.0.0.0:2181 --partitions 1 --replication-factor 1
#!kafka-console-producer --topic t1 --broker-list 0.0.0.0:9092
brokers, topic = "0.0.0.0:9092",  "t1" 
kvs = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})
lines = kvs.map(lambda x: x[1])
counts = lines.flatMap(lambda line: line.split(" ")) \
        .map(lambda word: (word, 1)) \
        .reduceByKey(lambda a, b: a+b)
counts.pprint()

 
ssc.start()
ssc.awaitTermination()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11603812681309573, 'fraction_numerical': 0.02154993783671778, 'mean_word_length': 5.080604534005038, 'pattern_counts': {'":': 1, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '605106', 'license': 'MIT', 'n_tokens_mistral': 828, 'n_tokens_neox': 791, 'n_words': 242, 'path': 'spark_streaming/1_create_dstream.ipynb', 'repo_name': 'bryanyang0528/spark_practice', 'size': '2413'}