# ---
# 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
# ---

# +
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext

spark = SparkSession.builder.appName('MostViewedURL').master('local[*]').getOrCreate()
# -

ssc = StreamingContext(spark.sparkContext, 1) # 1s

# +
from pyspark.sql.functions import regexp_extract
# Monitor the logs directory for new log data, and read in the raw lines as accessLines
accessLines = spark.readStream.text("../SparkCourse/logs")

# Parse out the common log format to a DataFrame
contentSizeExp = r'\s(\d+)$'
statusExp = r'\s(\d{3})\s'
generalExp = r'\"(\S+)\s(\S+)\s*(\S*)\"'
timeExp = r'\[(\d{2}/\w{3}/\d{4}:\d{2}:\d{2}:\d{2} -\d{4})]'
hostExp = r'(^\S+\.[\S+\.]+\S+)\s'

logsDF = accessLines.select(regexp_extract('value', hostExp, 1).alias('host'),
                         regexp_extract('value', timeExp, 1).alias('timestamp'),
                         regexp_extract('value', generalExp, 1).alias('method'),
                         regexp_extract('value', generalExp, 2).alias('endpoint'),
                         regexp_extract('value', generalExp, 3).alias('protocol'),
                         regexp_extract('value', statusExp, 1).cast('integer').alias('status'),
                         regexp_extract('value', contentSizeExp, 1).cast('integer').alias('content_size'))
# -

from pyspark.sql import functions as F

# update timestamp, os valores no teste são antigos!
logsDF = logsDF.withColumn('timestampNew', F.current_timestamp())

# 30s window with a 10s slide
endpointsCount = logsDF.groupby(F.window(F.col('timestampNew'),\
                                         windowDuration='30 seconds', \
                                         slideDuration='10 seconds'),\
                                F.col('endpoint')).count()

query = ( endpointsCount.orderBy(F.col('count').desc()).writeStream.outputMode("complete").format("console").queryName("counts").start() )


# Run forever until terminated
query.awaitTermination()


# Cleanly shut down the session
spark.stop()



StarcoderdataJupyterScriptsDedupFiltered