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

import os
from datetime import datetime
import pyspark.sql.functions as F
import pyspark.sql.types as T
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("sparkify_etl").config("spark.sql.session.timeZone", "UTC")\
                            .master("local").getOrCreate()

# ## First, Let us check the schema for log data files

# +
file_path = os.path.join("data","log-data")
data_log = spark.read.json(file_path)
data_log = data_log.where(F.col("page")=="NextSong")
data_log.printSchema()
data_log.limit(2).toPandas()

# Observation
# 1. itemInSession can be integer
# 2. timestamp column can be datetime
# -

# ## We will check schema and explore song data files next

# +
file_path = os.path.join("data","song_data","*","*","*")
data_song = spark.read.json(file_path)
data_song.printSchema()

# Since there's one record per song file, we don't need to use limit
data_song.limit(5).toPandas()

# Observation
#lat,long can be double
# -

# ## Let's first create the user table

# +
#user_id, first_name, last_name, gender, level

df_user = data_log.select("userId","firstName","lastName","gender","level")

# User sql expression to cast specific columns
df_user = df_user.withColumn("userId",F.expr("cast(userId as long) userId"))
df_user.printSchema()
df_user.limit(5).toPandas()
# -

# ## Next we will create songs table

# +
#song_id, title, artist_id, year, duration

df_song = data_song.select("song_id","title","artist_id","year","duration")
df_song = df_song.withColumn("year",F.col("year").cast(T.IntegerType()))
df_song.printSchema()
df_song.toPandas()
# -

# ## Artist Table will be created from song data as well

# +
# artist_id, name, location, lattitude, longitude

df_artist = data_song.select("artist_id","artist_name","artist_location","artist_latitude","artist_longitude")
df_artist = df_artist.withColumn("artist_latitude",F.col("artist_latitude").cast(T.DecimalType()))
df_artist = df_artist.withColumn("artist_longitude",F.col("artist_longitude").cast(T.DecimalType()))
df_artist.printSchema()
df_artist.toPandas()
# -

# ## Our next dimension Table would be of Time where we'd split "ts" timestamp col further to      granular level

# +
# start_time, hour, day, week, month, year, weekday

df_time = data_log.select("ts")
time_format = "yyyy-MM-dd' 'HH:mm:ss.SSS"


#func = F.udf("start_time")
df_time = df_time.withColumn("start_time", \
                             F.to_utc_timestamp(F.from_unixtime(F.col("ts")/1000,format=time_format),tz="UTC"))
df_time = df_time.withColumn("hour",F.hour(F.col("start_time")))
df_time = df_time.withColumn("day",F.dayofmonth(F.col("start_time")))
df_time = df_time.withColumn("week",F.weekofyear(F.col("start_time")))
df_time = df_time.withColumn("month",F.month(F.col("start_time")))
df_time = df_time.withColumn("year",F.year(F.col("start_time")))
df_time = df_time.withColumn("weekday",F.dayofweek(F.col("start_time")))

df_time.printSchema()
df_time.limit(2).toPandas()
# -

# ## Now that we've created all DIMENSIONAL tables let us proceed for the FACTS table creation
#
# ### In order to create facts table, we have to perform joins
#
# #### SQL syntax is better for longer join queries, but the same can be replicated using spark dataframe operations

# +
# songplay_id, start_time, user_id, level, song_id, artist_id, session_id, location, user_agent
#TODO : Partition by specific keys before uploading to S3 as parquet

df_song_play = data_song.join(data_log,data_song.title==data_log.song, how="inner").\
                    select("userId","level","song_id","artist_id","sessionId","location","userAgent")
df_song_play.printSchema()
df_song_play.limit(2).toPandas()

# +
# First let us define some views
data_log.createOrReplaceTempView("t_log")
data_song.createOrReplaceTempView("t_song")
df_time.createOrReplaceTempView("t_time")

df_song_play = spark.sql("select t_time.start_time, t_log.userId, t_log.level, \
                          t_song.song_id, t_song.artist_id, t_log.sessionId, \
                          t_log.location, t_log.userAgent \
                          from t_log  \
                          inner join t_song \
                          on t_log.song=t_song.title \
                          inner join t_time \
                          on t_time.ts = t_log.ts \
                          where t_log.artist = t_song.artist_name \
                          and song_id is not null \
                          ")
df_song_play.limit(2).toPandas()
# -

df_song_play.write.mode("overwrite").parquet("data/output.parquet")



df = spark.read.parquet('data/time_parquet.parquet')
df.toPandas()



StarcoderdataJupyterScriptsDedupFiltered