import configparser
import os
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col
from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, dayofweek, monotonically_increasing_id
from pyspark.sql.types import IntegerType, StringType, DoubleType


config = configparser.ConfigParser()
config.read('dl.cfg')

os.environ['AWS_ACCESS_KEY_ID']=config['KEYS']['AWS_ACCESS_KEY_ID']
os.environ['AWS_SECRET_ACCESS_KEY']=config['KEYS']['AWS_SECRET_ACCESS_KEY']


def create_spark_session():
    spark = SparkSession \
        .builder \
        .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:2.7.0") \
        .getOrCreate()
    return spark


def process_song_data(spark, input_data, output_data):
    # get filepath to song data file
    song_data ="{input_data}song-data/*/*/*/*.json".format(input_data=input_data)

    # read song data file
    df = spark.read.json(song_data).dropDuplicates().cache()

    df.createOrReplaceTempView('song_df_table')

    df = df.withColumn("song_id", df["song_id"].cast(StringType()))
    df = df.withColumn("title", df["title"].cast(StringType()))
    df = df.withColumn("artist_id", df["artist_id"].cast(StringType()))
    df = df.withColumn("year", df["year"].cast(IntegerType()))
    df = df.withColumn("duration", df["duration"].cast(DoubleType()))
    df = df.withColumn("artist_name", df["artist_name"].cast(StringType()))
    df = df.withColumn("artist_location", df["artist_location"].cast(StringType()))
    df = df.withColumn("artist_latitude", df["artist_latitude"].cast(DoubleType()))
    df = df.withColumn("artist_longitude", df["artist_longitude"].cast(DoubleType()))

    # extract columns to create songs table
    songs_table = df.select(
        col('song_id'),
        col('title'),
        col('artist_id'),
        col('year'),
        col('duration')
    ).distinct()

    # write songs table to parquet files partitioned by year and artist
    songs_table.write.partitionBy('year', 'artist_id') \
        .parquet("{output_data}songs/songs_table.parquet".format(
        output_data=output_data))

    # extract columns to create artists table
    artists_table = df.select(
        col('artist_id'),
        col('artist_name'),
        col('artist_location'),
        col('artist_latitude'),
        col('artist_longitude')
    ).distinct()

    # write artists table to parquet files
    artists_table.write.parquet("{output_data}artists/artists_table.parquet".format(
        output_data=output_data))


def process_log_data(spark, input_data, output_data):
    # get filepath to log data file
    log_data ="{input_data}log-data/*events.json".format(input_data=input_data)

    # read log data file
    df = spark.read.json(log_data).dropDuplicates().cache()

    # filter by actions for song plays
    df = df.filter(df.page == 'NextSong').cache()

    # extract columns for users table
    user_table = df.select(
        col('firstName'),
        col('lastName'),
        col('gender'),
        col('level'),
        col('userId')
    ).distinct()

    # write users table to parquet files
    user_table.write.parquet("{output_data}users/users_table.parquet".format(
        output_data=output_data))

    get_timestamp = udf(lambda x: str(int(int(x)/1000)))
    # create timestamp column from original timestamp column
    df = df.withColumn('timestamp', get_timestamp(col('ts')))

    # create datetime column from original timestamp column
    df = df.withColumn('start_time', get_timestamp(col('ts')))

    # extract columns to create time table
    df = df.withColumn('hour', hour('timestamp'))
    df = df.withColumn('day', dayofmonth('timestamp'))
    df = df.withColumn('month', month('timestamp'))
    df = df.withColumn('year', year('timestamp'))
    df = df.withColumn('week', weekofyear('timestamp'))
    df = df.withColumn('weekday', dayofweek('timestamp'))

    time_table = df.select(
        col('start_time'),
        col('hour'),
        col('day'),
        col('week'),
        col('month'),
        col('year'),
        col('weekday')
    ).distinct()

    # write time table to parquet files partitioned by year and month
    time_table.write.partitionBy('year', 'month') \
        .parquet("{output_data}time/time_table.parquet".format(
        output_data=output_data))

    # read in song data to use for songplays table
    song_df = spark.sql(
        'SELECT DISTINCT song_id, artist_id, artist_name FROM song_df_table'
    )

    # extract columns from joined song and log datasets to create songplays table
    songplays_table = df.join(
        song_df,
        song_df.artist_name == df.artist,
        'inner'
    ) \
        .distinct() \
        .select(
        col('start_time'),
        col('userId'),
        col('level'),
        col('sessionId'),
        col('location'),
        col('userAgent'),
        col('song_id'),
        col('artist_id'),
        col('year'),
        col('month')) \
        .withColumn('songplay_id', monotonically_increasing_id())

    # write songplays table to parquet files partitioned by year and month
    songplays_table.write.partitionBy('year', 'month') \
        .parquet("{output_data}songplays/songplays_table.parquet".format(
        output_data=output_data))


def main():
    spark = create_spark_session()
    input_data = 's3a://udacity-dend/'
    output_data = ""

    process_song_data(spark, input_data, output_data)
    process_log_data(spark, input_data, output_data)


if __name__ == "__main__":
    main()

import configparser
import os
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col
from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, dayofweek, monotonically_increasing_id
from pyspark.sql.types import IntegerType, StringType, DoubleType


config = configparser.ConfigParser()
config.read('dl.cfg')

os.environ['AWS_ACCESS_KEY_ID']=config['KEYS']['AWS_ACCESS_KEY_ID']
os.environ['AWS_SECRET_ACCESS_KEY']=config['KEYS']['AWS_SECRET_ACCESS_KEY']


def create_spark_session():
    spark = SparkSession \
        .builder \
        .config("spark.jars.packages", "org.apache.hadoop:hadoop-aws:2.7.0") \
        .getOrCreate()
    return spark


def process_song_data(spark, input_data, output_data):
    # get filepath to song data file
    song_data ="{input_data}song-data/*/*/*/*.json".format(input_data=input_data)

    # read song data file
    df = spark.read.json(song_data).dropDuplicates().cache()

    df.createOrReplaceTempView('song_df_table')

    df = df.withColumn("song_id", df["song_id"].cast(StringType()))
    df = df.withColumn("title", df["title"].cast(StringType()))
    df = df.withColumn("artist_id", df["artist_id"].cast(StringType()))
    df = df.withColumn("year", df["year"].cast(IntegerType()))
    df = df.withColumn("duration", df["duration"].cast(DoubleType()))
    df = df.withColumn("artist_name", df["artist_name"].cast(StringType()))
    df = df.withColumn("artist_location", df["artist_location"].cast(StringType()))
    df = df.withColumn("artist_latitude", df["artist_latitude"].cast(DoubleType()))
    df = df.withColumn("artist_longitude", df["artist_longitude"].cast(DoubleType()))

    # extract columns to create songs table
    songs_table = df.select(
        col('song_id'),
        col('title'),
        col('artist_id'),
        col('year'),
        col('duration')
    ).distinct()

    # write songs table to parquet files partitioned by year and artist
    songs_table.write.partitionBy('year', 'artist_id') \
        .parquet("{output_data}songs/songs_table.parquet".format(
        output_data=output_data))

    # extract columns to create artists table
    artists_table = df.select(
        col('artist_id'),
        col('artist_name'),
        col('artist_location'),
        col('artist_latitude'),
        col('artist_longitude')
    ).distinct()

    # write artists table to parquet files
    artists_table.write.parquet("{output_data}artists/artists_table.parquet".format(
        output_data=output_data))


def process_log_data(spark, input_data, output_data):
    # get filepath to log data file
    log_data ="{input_data}log-data/*events.json".format(input_data=input_data)

    # read log data file
    df = spark.read.json(log_data).dropDuplicates().cache()

    # filter by actions for song plays
    df = df.filter(df.page == 'NextSong').cache()

    # extract columns for users table
    user_table = df.select(
        col('firstName'),
        col('lastName'),
        col('gender'),
        col('level'),
        col('userId')
    ).distinct()

    # write users table to parquet files
    user_table.write.parquet("{output_data}users/users_table.parquet".format(
        output_data=output_data))

    get_timestamp = udf(lambda x: str(int(int(x)/1000)))
    # create timestamp column from original timestamp column
    df = df.withColumn('timestamp', get_timestamp(col('ts')))

    # create datetime column from original timestamp column
    df = df.withColumn('start_time', get_timestamp(col('ts')))

    # extract columns to create time table
    df = df.withColumn('hour', hour('timestamp'))
    df = df.withColumn('day', dayofmonth('timestamp'))
    df = df.withColumn('month', month('timestamp'))
    df = df.withColumn('year', year('timestamp'))
    df = df.withColumn('week', weekofyear('timestamp'))
    df = df.withColumn('weekday', dayofweek('timestamp'))

    time_table = df.select(
        col('start_time'),
        col('hour'),
        col('day'),
        col('week'),
        col('month'),
        col('year'),
        col('weekday')
    ).distinct()

    # write time table to parquet files partitioned by year and month
    time_table.write.partitionBy('year', 'month') \
        .parquet("{output_data}time/time_table.parquet".format(
        output_data=output_data))

    # read in song data to use for songplays table
    song_df = spark.sql(
        'SELECT DISTINCT song_id, artist_id, artist_name FROM song_df_table'
    )

    # extract columns from joined song and log datasets to create songplays table
    songplays_table = df.join(
        song_df,
        song_df.artist_name == df.artist,
        'inner'
    ) \
        .distinct() \
        .select(
        col('start_time'),
        col('userId'),
        col('level'),
        col('sessionId'),
        col('location'),
        col('userAgent'),
        col('song_id'),
        col('artist_id'),
        col('year'),
        col('month')) \
        .withColumn('songplay_id', monotonically_increasing_id())

    # write songplays table to parquet files partitioned by year and month
    songplays_table.write.partitionBy('year', 'month') \
        .parquet("{output_data}songplays/songplays_table.parquet".format(
        output_data=output_data))


def main():
    spark = create_spark_session()
    input_data = 's3a://udacity-dend/'
    output_data = ""

    process_song_data(spark, input_data, output_data)
    process_log_data(spark, input_data, output_data)


if __name__ == "__main__":
    main()

# get filepath to song data file # read song data file # extract columns to create songs table # write songs table to parquet files partitioned by year and artist # extract columns to create artists table # write artists table to parquet files # get filepath to log data file # read log data file # filter by actions for song plays # extract columns for users table # write users table to parquet files # create timestamp column from original timestamp column # create datetime column from original timestamp column # extract columns to create time table # write time table to parquet files partitioned by year and month # read in song data to use for songplays table # extract columns from joined song and log datasets to create songplays table # write songplays table to parquet files partitioned by year and month