
!pip install findspark
!pip install pyspark


# vytvorenie spark aplikacie
import findspark
findspark.init()

import pyspark


from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# Vytvorime objekt Spark kontext a vytvorime Streaming kontext. Ten (okrem parametra Spark kontextu) definuje aj parameter, ktory specifikuje velkost okna pre mikro-davky. V tomto pripade
# V tomto pripade StreamingContext(sc, 1) vytvori Streaming kontext zo Spark kontextu sc a s intervalom mikro-davok 1 sekunda  

batchIntervalSeconds = 1

sc = SparkContext(appName = "NetworkWordCount")
ssc = StreamingContext(sc, batchIntervalSeconds)

# Teraz vytvorime DStream - diskretizovany stream - ktory nacitava data 
# použijeme operaciu socketTextStream ktora bude ocakavat vstupne data na localhoste, na porte 9999

lines = ssc.socketTextStream("localhost", 9999)

# Tento DStream potom obsahuje textove data prijate z localhostu na porte 9999. Kazdy zaznam obsahuje 
# riadok textu. Tie potom spracujeme operaciami podobne, ako pri skripte wordcount
# zo statickych dat

# Najprv teda odseparujeme v retazci slova oddelene medzerou a spocitame jednotlive vyskyty slov

wordCounts = lines.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda x,y: x + y)

# Vypiseme ich na obrazovku pomocou pprint()

wordCounts.pprint()


ssc.start()
ssc.awaitTermination(10)
ssc.stop()


# Najprv nacitame potrebne kniznice
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split

# Pre pouzivanie Structured Streamingu budeme pouzivat objekt Spark Session
spark = SparkSession.builder.appName("StructuredNetworkWordCount").getOrCreate()

# Podobne ako v prvom priklade budeme nacitavat data z localhostu na porte 9999
# Rozdiel je v tom, ze tentokrat nebudeme vytvarat 'DStream' (RDD) ale datovy ramec
# Pomocou funkcie 'readStream' vieme specifikovat zdroj a typ prudu ('socket'), volitelne parametre ('host' a 'port'), funkcia load() prud nacita

lines = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()

# Datovy ramec s nacitanymi datami transformujeme na iny datovy ramec - rozdelenim podla oddelovaca
# Funkcia explode vytvori novy zaznam pre kazdy element pre dany stlpec datoveho ramca - transformuje datovy ramec
# pozostavajuci z riadkov na datovy ramec slov
 
words = lines.select(
   explode(
       split(lines.value, " ")
   ).alias("word")
)

# Nasledne mozme pouzit operacie pre datove ramce a spocitat pocet vyskytov slov (zoskupenim podla slov a spocitanim)

wordCounts = words.groupBy("word").count()

# Pri Structured Streamingu je ale aj vysledok operacie strukturovany prud. Ten teda vytvorime a bude 
# kontinualne vykonavat na danom prude specifikovany vypocet

query = wordCounts.writeStream.outputMode("complete").queryName("counts").format("memory").start()
query.awaitTermination(20)
query.stop()

spark.table("counts").show()

# Po vykonani tohoto kodu sa na pozadi spusti vypocet a bude sa ukladat do konzoly.
# V tomto pripade je nastaveny sposob vystupu "complete".


# Importujeme potrebne moduly 
from pyspark import SparkContext, Row, SQLContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import *

# Vytvorime Spark Context a Spark Session objekty
#sc = SparkContext(appName="pubnub_dir_streaming_app")
#spark = SparkSession.builder.appName("StreamingPubNub").getOrCreate()

# Specifikujeme schemu JSON objektov, ktore budeme spracovavat. Pre atributy (okrem 'timestamp' atributu) typ String, kedze prijate JSON objekty obsahuju hodnoty velicin ako retazce

schema = StructType([
    StructField("timestamp", TimestampType(), True),
    StructField('ambient_temperature', StringType(), True),
    StructField('humidity', StringType(), True),
    StructField('photosensor', StringType(), True),
    StructField('sensor_uuid', StringType(), True),
    StructField('radiation_level', StringType(), True)
])

# Vytvorime Streaming Data Frame zo zdroja, ktory obsahuje pribudajuce JSON subory. Ako parameter pri vytvarani datoveho ramca definujeme schemu JSON objektov, ktore budeme spracovavat
# ako parameter nastavime atribut 'maxFilesPerTrigger', ktory specifikuje, kolko suborov sa bude naraz spracovavat a zdroj JSON streamovanych objektov

df = spark.readStream.schema(schema).option("maxFilesPerTrigger",1).json("stream/")

# V dalsom kroku transformujeme atributy datoveho ramca na pozadovane hodnty (hodnoty senzorickych velicin na z retazcov na numericke)

df = df.withColumn("ambient_temperature", df["ambient_temperature"].cast(FloatType()))
df = df.withColumn("humidity", df["humidity"].cast(FloatType()))
df = df.withColumn("photosensor", df["photosensor"].cast(FloatType()))
df = df.withColumn("radiation_level", df["radiation_level"].cast(FloatType()))


# Ak chceme skontrolovat streamovany datovy ramec, musime specifikovat streamovany dopyt, ktory ho bude kontinualne po monitorovat
# Vytvorime teda 'df_stream' z datoveho ramca, specifikujeme sposob vypisu do konzoly ("console") a spustime

df_stream = df.writeStream.format("memory").queryName("stream").start()

# Tato query bude aktivna 10 sekund
df_stream.awaitTermination(10)
df_stream.stop()

spark.table("stream").show()


df_stream = df.writeStream.format("memory").queryName("stream").start()

query = df.select(df["sensor_uuid"]).where(df["humidity"] >= 10.0)
query_stream = query.writeStream.format("memory").queryName("humidity").start()
df_stream.awaitTermination(10)
df_stream.stop()
query_stream.stop()

spark.table("humidity").show()


df_stream = df.writeStream.format("memory").queryName("stream").start()

query = df.filter(df['radiation_level'] >= 200.0)
query_stream = query.writeStream.format("memory").queryName("radiation").start()
df_stream.awaitTermination(10)
df_stream.stop()
query_stream.stop()

spark.table("radiation").show()


df_stream = df.writeStream.format("memory").queryName("stream").start()

query = df.groupBy("sensor_uuid").count()
query_stream = query.writeStream.outputMode("complete").format("memory").queryName("uuid_counts").start()
df_stream.awaitTermination(10)
df_stream.stop()
query_stream.stop()

spark.table("uuid_counts").show()


df_stream = df.writeStream.format("memory").queryName("stream").start()

query = df.groupBy(df['sensor_uuid'], window(df['timestamp'], "5 seconds")).count()
query_stream = query.writeStream.format("memory").outputMode("complete").queryName("windows").start()
df_stream.awaitTermination(10)
df_stream.stop()
query_stream.stop()

spark.table("windows").show()

Cvičenie 10 - Spracovanie prúdov dát pomocou Apache Spark Streaming¶

Práca so štruktúrovanými prúdmi¶

Spracovanie streamu senzorických dát¶

Úloha 10.1¶

Úloha 10.2¶

Úloha 10.3¶