<jupyter_start><jupyter_text><jupyter_code>!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget https://downloads.apache.org/spark/spark-2.4.8/spark-2.4.8-bin-hadoop2.7.tgz
!tar -xvf spark-2.4.8-bin-hadoop2.7.tgz
!pip install findspark
!wget "https://repo1.maven.org/maven2/org/apache/spark/spark-streaming-kafka-0-8-assembly_2.11/2.4.8/spark-streaming-kafka-0-8-assembly_2.11-2.4.8.jar"
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.4.8-bin-hadoop2.7"
os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars /content/spark-streaming-kafka-0-8-assembly_2.11-2.4.8.jar pyspark-shell'
import findspark
findspark.init()
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.ml.feature import Normalizer, StandardScaler
import random
import pyspark
import sys
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from uuid import uuid1
import time

kafka_topic_name = "my-first-topic"
kafka_bootstrap_servers = 'ec2-13-229-46-113.ap-southeast-1.compute.amazonaws.com:9092'
sc = pyspark.SparkContext()
ssc = StreamingContext(sc,1)
kvs = KafkaUtils.createStream(ssc, kafka_bootstrap_servers, 'spark-streaming-consumer', {kafka_topic_name:1}) 
kvs = KafkaUtils.createDirectStream(ssc, [kafka_topic_name], {"metadata.broker.list": kafka_bootstrap_servers})
kvs = KafkaUtils.createDirectStream(ssc, [kafka_topic_name], {
                        'bootstrap.servers':kafka_bootstrap_servers,
                        'group.id':'test-group',
                        'auto.offset.reset':'largest'})
lines = kvs.map(lambda x: x[1])
counts1 = lines.flatMap(lambda line: line.split(' ')).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a+b).filter(lambda y: y[1]>5)
print(type(counts1))
counts1.pprint()
ssc.start()
# stream will run for 30 sec
ssc.awaitTerminationOrTimeout(30)
ssc.stop()
sc.stop()
ssc.stop()
sc.stop()<jupyter_output>-------------------------------------------
Time: 2021-09-14 05:01:46
-------------------------------------------

-------------------------------------------
Time: 2021-09-14 05:01:47
-------------------------------------------

-------------------------------------------
Time: 2021-09-14 05:01:48
-------------------------------------------

-------------------------------------------
Time: 2021-09-14 05:01:49
-------------------------------------------

-------------------------------------------
Time: 2021-09-14 05:01:50
-------------------------------------------

-------------------------------------------
Time: 2021-09-14 05:01:51
-------------------------------------------

-------------------------------------------
Time: 2021-09-14 05:01:52
-------------------------------------------

-------------------------------------------
Time: 2021-09-14 05:01:53
-------------------------------------------

-------------------------------------------
Time: 2021-09-14 05:01:54
----------[...]
{'fraction_non_alphanumeric': 0.360530679933665, 'fraction_numerical': 0.06766169154228856, 'mean_word_length': 1.5888412017167381, 'pattern_counts': {'":': 1, '<': 4, '<?xml version=': 0, '>': 6, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '91479', 'license': 'Apache-2.0', 'n_tokens_mistral': 1062, 'n_tokens_neox': 934, 'n_words': 147, 'path': 'Week06/Week6_spark.ipynb', 'repo_name': 'k-sukhantharat/Data_Streaming_and_Realtime_Analytics', 'size': '3015'}