# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

import pyspark
from pyspark import SparkContext
sc = SparkContext.getOrCreate();
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").getOrCreate()
spark.conf.set("spark.sql.repl.eagerEval.enabled", True) # Property used to format output tables better
spark

sc.stop()

from __future__ import print_function
from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc=SparkContext(appName="StreamingTwitterAnalysis")
sc.setLogLevel("ERROR")
ssc=StreamingContext(sc,10)

socket_stream = ssc.socketTextStream("127.0.0.1",5555)

lines=socket_stream.window(60)

hashtags = lines.flatMap(lambda text:text.split(" ")).filter(lambda word:word.lower().startswith('#')).map(lambda word:(word.lower(),1)).reduceByKey(lambda a,b:a+b)

author_counts_sorted_dstream=hashtags.transform(lambda foo:foo.sortBy(lambda x:x[0].lower()).sortBy(lambda x:x[1],ascending=False))

author_counts_sorted_dstream.pprint()

ssc.start()

ssc.awaitTermination()





StarcoderdataJupyterScriptsDedupFiltered