First commit

Spark+Data+Lakes/Project+Data+Lake/README.md

@@ -0,0 +1,58 @@
+# Project: Data Lake
+-------------------------
+
+### Introduction
+
+In this project, we will help Sparkifly music streaming startup to move their data warehouse to a data lake. Their data resides in S3, then we will provide an ETL pipeline that extracts their data from S3, processes them using Spark, and loads the data back into S3 as a set of dimensional tables.
+
+### Project Datasets
+
+We'll be working with two datasets that resides in S3. Here are the S3 links for each:
+
++ **Song Dataset** - The first dataset is a subset of real data from the Million Song Dataset. Each file is in JSON format and contains metadata about a song and the artist of that song. The files are partitioned by the first three letters of each song's track ID. For example, here are file paths to two files in this dataset.
+
+```
+song_data/A/B/C/TRABCEI128F424C983.json
+song_data/A/A/B/TRAABJL12903CDCF1A.json
+```
+
+And below is an example of what a single song file, TRAABJL12903CDCF1A.json, looks like.
+
+```
+{"num_songs": 1, "artist_id": "ARJIE2Y1187B994AB7", "artist_latitude": null, "artist_longitude": null, "artist_location": "", "artist_name": "Line Renaud", "song_id": "SOUPIRU12A6D4FA1E1", "title": "Der Kleine Dompfaff", "duration": 152.92036, "year": 0}
+
+```
+
++ **Log Dataset** - The second dataset consists of log files in JSON format generated by this event simulator based on the songs in the dataset above. These simulate app activity logs from an imaginary music streaming app based on configuration settings.
+
+The log files in the dataset you'll be working with are partitioned by year and month. For example, here are file paths to two files in this dataset.
+
+```
+log_data/2018/11/2018-11-12-events.json
+log_data/2018/11/2018-11-13-events.json
+```
+
+And below is an example of what the data in a log file, 2018-11-12-events.json, looks like.
+
+![log-data](./log-data.PNG)
+
+### Schema for Song Play Analysis
+
+The database schema is shown as follows
+
+![schema](./fact_dimensional_tables.jpg)
+
+
+### Data processing
+
+You will find out all processing steps into `etl.py` file.
+
+To create `songs_table` we used the command **select()** in order to get only the columns required: 'song_id', 'title', 'artist_id', 'year', 'duration'.
+Also removed eventually duplicated values, with the command **dropDuplicates()**. The sames procedure was used to create other tables.
+
+As Sparkify mentioned that their user base and song database is growing, make sense we use **PySpark partitions**. This way we will access the data faster and provides the ability to perform an operation on a smaller dataset. To do so, we partioned the songs_table by 'year', 'artist_id'.
+The same happens with the table `time_table` and `songplays_table` partioned by 'year', 'month'.
+
+
+
+

Spark+Data+Lakes/Project+Data+Lake/dl.cfg

@@ -0,0 +1,2 @@
+AWS_ACCESS_KEY_ID=''
+AWS_SECRET_ACCESS_KEY=''

Spark+Data+Lakes/Project+Data+Lake/etl.ipynb

@@ -0,0 +1,304 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 61,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": [
+    "import configparser\n",
+    "from datetime import datetime\n",
+    "from pyspark.sql.functions import dayofweek\n",
+    "import os\n",
+    "from pyspark.sql import SparkSession\n",
+    "from pyspark.sql.functions import udf, col\n",
+    "from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 2,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": [
+    "def create_spark_session():\n",
+    "    spark = SparkSession \\\n",
+    "        .builder \\\n",
+    "        .config(\"spark.jars.packages\", \"org.apache.hadoop:hadoop-aws:2.7.0\") \\\n",
+    "        .getOrCreate()\n",
+    "    return spark"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 3,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": [
+    "def process_song_data(spark, input_data, output_data):\n",
+    "    \"\"\"\n",
+    "    This function reads the song data from S3, processes that data using Spark, and writes them back to S3\n",
+    "\n",
+    "    Parameters\n",
+    "    ----------\n",
+    "        spark: This is the spark session\n",
+    "        input_date: This is the path to song_date into S3 bucket\n",
+    "        output_data: This is the path where parquet files will be written\n",
+    "    \n",
+    "    \"\"\"\n",
+    "    \n",
+    "    # get filepath to song data file\n",
+    "    # e.g. from S3 file location: song_data/A/B/C/TRABCEI128F424C983.json\n",
+    "    song_data = input_data + 'song_data/*/*/*/*.json'\n",
+    "    \n",
+    "    # read song data file\n",
+    "    df = spark.read.json(song_data)\n",
+    "\n",
+    "    # extract columns to create songs table\n",
+    "    songs_table = df.select('song_id', 'title', 'artist_id', 'year', 'duration').dropDuplicates()\n",
+    "    songs_table.createOrReplaceTempView('songs')\n",
+    "    \n",
+    "    # write songs table to parquet files partitioned by year and artist\n",
+    "    songs_table.write.partitionBy('year', 'artist_id').parquet(os.path.join(output_data, 'songs/songs.parquet'), 'overwrite')\n",
+    "\n",
+    "    # extract columns to create artists table\n",
+    "    artists_table = df.select('artist_id', 'artist_name', 'artist_location', 'artist_latitude', 'artist_longitude') \\\n",
+    "                      .withColumnRenamed('artist_name', 'name') \\\n",
+    "                      .withColumnRenamed('artist_location', 'location') \\\n",
+    "                      .withColumnRenamed('artist_latitude', 'latitude') \\\n",
+    "                      .withColumnRenamed('artist_longitude', 'longitude').dropDuplicates()\n",
+    "    \n",
+    "    artists_table.createOrReplaceTempView('artists')\n",
+    "    \n",
+    "    \n",
+    "    # write artists table to parquet files\n",
+    "    artists_table.write.parquet(os.path.join(output_data, 'artists/artists.parquet'), 'overwrite')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 64,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": [
+    "def process_log_data(spark, input_data, output_data):\n",
+    "    \"\"\"\n",
+    "    This function reads the log data from S3, processes that data using Spark, and writes them back to S3\n",
+    "\n",
+    "    Parameters\n",
+    "    ----------\n",
+    "        spark: This is the spark session\n",
+    "        input_date: This is the path to log_date into S3 bucket\n",
+    "        output_data: This is the path where parquet files will be written\n",
+    "    \n",
+    "    \"\"\"\n",
+    "    \n",
+    "    \n",
+    "    # get filepath to log data file\n",
+    "    # e.g. from S3 file location: log_data/2018/11/2018-11-12-events.json\n",
+    "    log_data = input_data + 'log_data/*.json'\n",
+    "\n",
+    "    # read log data file\n",
+    "    df = spark.read.json(log_data)\n",
+    "    \n",
+    "    # filter by actions for song plays\n",
+    "    # df_actions = df.filter(df.page == \"NextSong\").select('ts', 'userId', 'level', 'song', 'artist', 'sessionId', 'location', 'userAgent')\n",
+    "    df_actions = df.where(df.page == 'NextSong')\n",
+    "    df_actions.select('ts', 'userId', 'level', 'song', 'artist', 'sessionId', 'location', 'userAgent')\n",
+    "    \n",
+    "    # extract columns for users table    \n",
+    "    users_table = df.select('userId', 'firstName', 'lastName', 'gender', 'level').dropDuplicates()\n",
+    "    users_table.createOrReplaceTempView('users')\n",
+    "    \n",
+    "    # write users table to parquet files\n",
+    "    users_table.write.parquet(os.path.join(output_data, 'users/users.parquet'), 'overwrite')\n",
+    "\n",
+    "    # create timestamp column from original timestamp column\n",
+    "    get_timestamp = udf(lambda x: str(int(int(x)/1000)))\n",
+    "    df = df.withColumn('timestamp', get_timestamp(df_actions.ts))\n",
+    "    \n",
+    "    # create datetime column from original timestamp column\n",
+    "    get_datetime = udf(lambda x: str(datetime.fromtimestamp(int(x) / 1000)))\n",
+    "    df = df.withColumn('datetime', get_datetime(df_actions.ts))\n",
+    "    \n",
+    "    # extract columns to create time table\n",
+    "    time_table = df.select('datetime') \\\n",
+    "                           .withColumn('start_time', df.datetime) \\\n",
+    "                           .withColumn('hour', hour('datetime')) \\\n",
+    "                           .withColumn('day', dayofmonth('datetime')) \\\n",
+    "                           .withColumn('week', weekofyear('datetime')) \\\n",
+    "                           .withColumn('month', month('datetime')) \\\n",
+    "                           .withColumn('year', year('datetime')) \\\n",
+    "                           .withColumn('weekday', dayofweek('datetime')) \\\n",
+    "                           .dropDuplicates()\n",
+    "    \n",
+    "    # write time table to parquet files partitioned by year and month\n",
+    "    time_table.write.partitionBy('year', 'month').parquet(os.path.join(output_data,'time/time.parquet'), 'overwrite')\n",
+    "\n",
+    "    # read in song data to use for songplays table\n",
+    "    song_df = spark.read.json(input_data + 'song_data/*/*/*/*.json')\n",
+    "\n",
+    "    # extract columns from joined song and log datasets to create songplays table\n",
+    "    df = df.alias('log_df')\n",
+    "    song_df = song_df.alias('song_df')\n",
+    "    log_and_songs = df.join(song_df, col('log_df.artist') == col('song_df.artist_name'), 'inner')\n",
+    "    \n",
+    "    \n",
+    "    songplays_table = log_and_songs.select(\n",
+    "        col('log_df.datetime').alias('start_time'),\n",
+    "        col('log_df.userId').alias('user_id'),\n",
+    "        col('log_df.level').alias('level'),\n",
+    "        col('song_df.song_id').alias('song_id'),\n",
+    "        col('song_df.artist_id').alias('artist_id'),\n",
+    "        col('log_df.sessionId').alias('session_id'),\n",
+    "        col('log_df.location').alias('location'), \n",
+    "        col('log_df.userAgent').alias('user_agent'),\n",
+    "        year('log_df.datetime').alias('year'),\n",
+    "        month('log_df.datetime').alias('month'))\n",
+    "\n",
+    "    songplays_table.createOrReplaceTempView('songplays')\n",
+    "    \n",
+    "    # write songplays table to parquet files partitioned by year and month\n",
+    "    time_table = time_table.alias('timetable')\n",
+    "\n",
+    "    songplays_table.write.partitionBy('year', 'month').parquet(os.path.join(output_data,'songplays/songplays.parquet'),'overwrite')\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 5,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": [
+    "    '''Start spark session and define S3 locations where the files will be readed and written'''\n",
+    "    spark = create_spark_session()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 6,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": [
+    "    '''Paths for local testing'''\n",
+    "    input_data = \"data/\"\n",
+    "    output_data = \"data/outputs/\""
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 65,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": [
+    "    '''Run above functions according to the previous parameters '''\n",
+    "    process_song_data(spark, input_data, output_data)    \n",
+    "    process_log_data(spark, input_data, output_data)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 59,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": [
+    "from pyspark.sql.functions import dayofweek\n",
+    "df = spark.read.json(\"data/log_data/*.json\")\n",
+    "df_actions = df.where(df.page == 'NextSong')\n",
+    "df_actions.select('ts', 'userId', 'level', 'song', 'artist', 'sessionId', 'location', 'userAgent')\n",
+    "\n",
+    "# create timestamp column from original timestamp column\n",
+    "get_timestamp = udf(lambda x: str(int(int(x)/1000)))\n",
+    "df = df.withColumn('timestamp', get_timestamp(df_actions.ts))\n",
+    "\n",
+    "# create datetime column from original timestamp column\n",
+    "get_datetime = udf(lambda x: str(datetime.fromtimestamp(int(x) / 1000)))\n",
+    "df = df.withColumn('datetime', get_datetime(df_actions.ts))\n",
+    "\n",
+    "# extract columns to create time table\n",
+    "time_table = df.select('datetime') \\\n",
+    "                           .withColumn('start_time', df.datetime) \\\n",
+    "                           .withColumn('hour', hour('datetime')) \\\n",
+    "                           .withColumn('day', dayofmonth('datetime')) \\\n",
+    "                           .withColumn('week', weekofyear('datetime')) \\\n",
+    "                           .withColumn('month', month('datetime')) \\\n",
+    "                           .withColumn('year', year('datetime')) \\\n",
+    "                           .withColumn('weekday', dayofweek('datetime')) \\\n",
+    "                           .dropDuplicates()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 60,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "+--------------------+--------------------+----+---+----+-----+----+-------+\n",
+      "|            datetime|          start_time|hour|day|week|month|year|weekday|\n",
+      "+--------------------+--------------------+----+---+----+-----+----+-------+\n",
+      "|2018-11-01 21:05:...|2018-11-01 21:05:...|  21|  1|  44|   11|2018|      5|\n",
+      "|2018-11-01 21:42:...|2018-11-01 21:42:...|  21|  1|  44|   11|2018|      5|\n",
+      "|2018-11-01 21:17:...|2018-11-01 21:17:...|  21|  1|  44|   11|2018|      5|\n",
+      "+--------------------+--------------------+----+---+----+-----+----+-------+\n",
+      "only showing top 3 rows\n",
+      "\n"
+     ]
+    }
+   ],
+   "source": [
+    "time_table.show(3)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "editable": true
+   },
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.6.3"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 4
+}

Spark+Data+Lakes/Project+Data+Lake/etl.py

@@ -0,0 +1,169 @@
+import configparser
+from datetime import datetime
+from pyspark.sql.functions import dayofweek
+import os
+from pyspark.sql import SparkSession
+from pyspark.sql.functions import udf, col
+from pyspark.sql.functions import year, month, dayofmonth, hour, weekofyear, date_format
+
+# In case you want to run this code locally, please set these next lines as comment
+
+config = configparser.ConfigParser()
+config.read('dl.cfg')
+
+os.environ['AWS_ACCESS_KEY_ID']=config['AWS_ACCESS_KEY_ID']
+os.environ['AWS_SECRET_ACCESS_KEY']=config['AWS_SECRET_ACCESS_KEY']
+
+# In case you want to run this code locally, please set these above lines as comment
+
+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):
+    """
+    This function reads the song data from S3, processes that data using Spark, and writes them back to S3
+
+    Parameters
+    ----------
+        spark: This is the spark session
+        input_date: This is the path to song_date into S3 bucket
+        output_data: This is the path where parquet files will be written
+    
+    """
+    
+    # get filepath to song data file
+    # e.g. from S3 file location: song_data/A/B/C/TRABCEI128F424C983.json
+    song_data = input_data + 'song_data/*/*/*/*.json'
+    
+    # read song data file
+    df = spark.read.json(song_data)
+
+    # extract columns to create songs table
+    songs_table = df.select('song_id', 'title', 'artist_id', 'year', 'duration').dropDuplicates()
+    songs_table.createOrReplaceTempView('songs')
+    
+    # write songs table to parquet files partitioned by year and artist
+    songs_table.write.partitionBy('year', 'artist_id').parquet(os.path.join(output_data, 'songs/songs.parquet'), 'overwrite')
+
+    # extract columns to create artists table
+    artists_table = df.select('artist_id', 'artist_name', 'artist_location', 'artist_latitude', 'artist_longitude') \
+                      .withColumnRenamed('artist_name', 'name') \
+                      .withColumnRenamed('artist_location', 'location') \
+                      .withColumnRenamed('artist_latitude', 'latitude') \
+                      .withColumnRenamed('artist_longitude', 'longitude').dropDuplicates()
+    
+    artists_table.createOrReplaceTempView('artists')
+    
+    
+    # write artists table to parquet files
+    artists_table.write.parquet(os.path.join(output_data, 'artists/artists.parquet'), 'overwrite')
+
+
+def process_log_data(spark, input_data, output_data):
+    """
+    This function reads the log data from S3, processes that data using Spark, and writes them back to S3
+
+    Parameters
+    ----------
+        spark: This is the spark session
+        input_date: This is the path to log_date into S3 bucket
+        output_data: This is the path where parquet files will be written
+    
+    """
+    
+    
+    # get filepath to log data file
+    # e.g. from S3 file location: log_data/2018/11/2018-11-12-events.json
+    log_data = input_data + 'log_data/*.json'
+
+    # read log data file
+    df = spark.read.json(log_data)
+    
+    # filter by actions for song plays
+    # df_actions = df.filter(df.page == "NextSong").select('ts', 'userId', 'level', 'song', 'artist', 'sessionId', 'location', 'userAgent')
+    df_actions = df.where(df.page == 'NextSong')
+    df_actions.select('ts', 'userId', 'level', 'song', 'artist', 'sessionId', 'location', 'userAgent')
+    
+    # extract columns for users table    
+    users_table = df.select('userId', 'firstName', 'lastName', 'gender', 'level').dropDuplicates()
+    users_table.createOrReplaceTempView('users')
+    
+    # write users table to parquet files
+    users_table.write.parquet(os.path.join(output_data, 'users/users.parquet'), 'overwrite')
+
+    # create timestamp column from original timestamp column
+    get_timestamp = udf(lambda x: str(int(int(x)/1000)))
+    df = df.withColumn('timestamp', get_timestamp(df_actions.ts))
+    
+    # create datetime column from original timestamp column
+    get_datetime = udf(lambda x: str(datetime.fromtimestamp(int(x) / 1000)))
+    df = df.withColumn('datetime', get_datetime(df_actions.ts))
+    
+    # extract columns to create time table
+    time_table = df.select('datetime') \
+                           .withColumn('start_time', df.datetime) \
+                           .withColumn('hour', hour('datetime')) \
+                           .withColumn('day', dayofmonth('datetime')) \
+                           .withColumn('week', weekofyear('datetime')) \
+                           .withColumn('month', month('datetime')) \
+                           .withColumn('year', year('datetime')) \
+                           .withColumn('weekday', dayofweek('datetime')) \
+                           .dropDuplicates()
+    
+    # write time table to parquet files partitioned by year and month
+    time_table.write.partitionBy('year', 'month').parquet(os.path.join(output_data,'time/time.parquet'), 'overwrite')
+
+    # read in song data to use for songplays table
+    song_df = spark.read.json(input_data + 'song_data/*/*/*/*.json')
+
+    # extract columns from joined song and log datasets to create songplays table
+    df = df.alias('log_df')
+    song_df = song_df.alias('song_df')
+    log_and_songs = df.join(song_df, col('log_df.artist') == col('song_df.artist_name'), 'inner')
+    
+    
+    songplays_table = log_and_songs.select(
+        col('log_df.datetime').alias('start_time'),
+        col('log_df.userId').alias('user_id'),
+        col('log_df.level').alias('level'),
+        col('song_df.song_id').alias('song_id'),
+        col('song_df.artist_id').alias('artist_id'),
+        col('log_df.sessionId').alias('session_id'),
+        col('log_df.location').alias('location'), 
+        col('log_df.userAgent').alias('user_agent'),
+        year('log_df.datetime').alias('year'),
+        month('log_df.datetime').alias('month'))
+
+    songplays_table.createOrReplaceTempView('songplays')
+    
+    # write songplays table to parquet files partitioned by year and month
+    time_table = time_table.alias('timetable')
+
+    songplays_table.write.partitionBy('year', 'month').parquet(os.path.join(output_data,'songplays/songplays.parquet'),'overwrite')
+
+
+def main():
+    '''Start spark session and define S3 locations where the files will be readed and written'''
+    spark = create_spark_session()
+    
+    '''Paths to run on S3 buckets'''
+    input_data = "s3a://udacity-dend/"
+    output_data = "s3a://gp-output-data"
+    
+    '''Paths for local testing'''
+#     input_data = "data/"
+#     output_data = "data/outputs/"
+    
+    
+    '''Run above functions according to the previous parameters '''
+    process_song_data(spark, input_data, output_data)    
+    process_log_data(spark, input_data, output_data)
+
+
+if __name__ == "__main__":
+    main()