labs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# AWS Glue Notebook for Serverless Data Lake Workshop\n",
    "This notebook contains the PySpark scripts run in AWS Glue to transform the data in the data lake. Each section refers a section in the lab. \n",
    "\n",
    "## Initialization\n",
    "The first two sections initialize the Spark environment and only need to be run once. The first block may take a few seconds as it negotiates the Spark session."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## @ Import the AWS Glue libraries, pySpark we'll need \n",
    "import sys\n",
    "from awsglue.transforms import *\n",
    "from awsglue.utils import getResolvedOptions\n",
    "from pyspark.context import SparkContext\n",
    "from awsglue.context import GlueContext\n",
    "from awsglue.job import Job\n",
    "from pyspark.sql.functions import *\n",
    "from awsglue.dynamicframe import DynamicFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## @ set up a single GlueContext.\n",
    "sc = SparkContext.getOrCreate()\n",
    "\n",
    "glueContext = GlueContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lab - Transform / Decode data with AWS Glue\n",
    "The first script is a straight forward transformation of the user activity table. The request column contains a number of fields embedded within it\n",
    "\n",
    "```\n",
    "GET /petstore/Cats/Treats\n",
    "```\n",
    "The column is split out to get the request type, top level, top page, and sub page.\n",
    "\n",
    "The timestamp field is also parsed out to extract the date components: date, time, month, and year.\n",
    "\n",
    "\n",
    "17/Jan/2018:10:43:54 \n",
    "Date: 01/17/2018\n",
    "Time: 10:43:54\n",
    "Year: 2018\n",
    "Month: 10\n",
    "\n",
    "| ip_address | username | timestamp | request | http | bytes | requesttype | topdomain | subpage | date | time | year | month | toppage |\n",
    "|------|------|------|------|------|------|------|------|------|------|------|------|------|------|\n",
    "|   0.32.193.205  | grldmccfdm8 | 11/Oct/2018:23:36:54 | DELETE | /petstore/Bird/Treats | 500 | 927 | DELETE | petstore | Treats | 10/11/2018 | 23:36:54 | 2018 | 10 | Bird |\n",
    "|   0.32.193.205  | grldmccfdm8 | 6/Jun/2017:13:03:54 | PUT | /petstore/Bird/Food | 500 | 927 | DELETE | petstore | Food | 06/06/2017 | 13:03:54 | 2017 | 6 | Bird |\n",
    "\n",
    "The data is then written out in a partitioned parquet format.\n",
    "\n",
    "*This process takes 3-5 minutes.*\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = glueContext.spark_session\n",
    "job = Job(glueContext)\n",
    "job.init('^stackname^-exercise1')\n",
    "\n",
    "## @ create the Glue DynamicFrame from table schema. A DynamicFrame is similar to a DataFrame, except that each record is \n",
    "## @ self-describing, so no schema is required initially.\n",
    "useractivity = glueContext.create_dynamic_frame.from_catalog(database = \"weblogs\", table_name = \"useractivity\", transformation_ctx = \"useractivity\")\n",
    "\n",
    "## @ ApplyMapping is one of the built in transforms that maps source columns and data types from a DynamicFrame to target columns \n",
    "## @ and data types in a returned DynamicFrame. You specify the mapping argument, which is a list of tuples that contain source column,\n",
    "## @ source type, target column, and target type.\n",
    "useractivityApplyMapping = ApplyMapping.apply(frame = useractivity, mappings = [(\"ip_address\", \"string\", \"ip_address\", \"string\"), (\"username\", \"string\", \"username\", \"string\"), (\"timestamp\", \"string\", \"timestamp\", \"string\"), (\"request\", \"string\", \"request\", \"string\"), (\"http\", \"long\", \"http\", \"long\"), (\"bytes\", \"long\", \"bytes\", \"long\")], transformation_ctx = \"applymapping1\")\n",
    "\n",
    "## @ ResolveChoice is another built in transform that you can use to specify how a column should be handled when it contains values of \n",
    "## @ multiple types. You can choose to either cast the column to a single data type, discard one or more of the types, or retain all \n",
    "## @ types in either separate columns or a structure. You can select a different resolution policy for each column or specify a global \n",
    "## @ policy that is applied to all columns.\n",
    "resolvechoice2 = ResolveChoice.apply(frame = useractivityApplyMapping, choice = \"make_struct\", transformation_ctx = \"resolvechoice2\")\n",
    "\n",
    "## @ DropNullFields transform removes null fields from a DynamicFrame. The output DynamicFrame does not contain fields of the null type\n",
    "## @ in the schema.\n",
    "useractivity = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = \"dropnullfields3\")\n",
    "\n",
    "## @ We will leverage PySpark functions to manipulate our data, starting with converting glue DynamicFrame to DataFrame\n",
    "dataframe0 = DynamicFrame.toDF(useractivity)\n",
    "\n",
    "## @ Use PySpark functions to split request columns on '/' \n",
    "split_column = split(dataframe0['request'], '/')\n",
    "\n",
    "dataframe0 = dataframe0.withColumn('requesttype', split_column.getItem(0))\n",
    "\n",
    "dataframe0 = dataframe0.withColumn('topdomain', split_column.getItem(1))\n",
    "dataframe0 = dataframe0.withColumn('toppage', split_column.getItem(2))\n",
    "dataframe0 = dataframe0.withColumn('subpage', split_column.getItem(3))\n",
    "\n",
    "## @ split timestamp column into date, time, year and month\n",
    "dataframe0 = dataframe0.withColumn('date',date_format(from_unixtime(unix_timestamp('timestamp', 'd/MMM/yyyy:HH:mm:ss')), 'MM/dd/yyy'))\n",
    "dataframe0 = dataframe0.withColumn('time',date_format(from_unixtime(unix_timestamp('timestamp', 'd/MMM/yyyy:HH:mm:ss')), 'HH:mm:ss'))\n",
    "\n",
    "dataframe0 = dataframe0.withColumn('year', year(from_unixtime(unix_timestamp('timestamp', 'd/MMM/yyyy:HH:mm:ss'))))\n",
    "dataframe0 = dataframe0.withColumn('month', month(from_unixtime(unix_timestamp('timestamp', 'd/MMM/yyyy:HH:mm:ss'))))\n",
    "\n",
    "\n",
    "## @ convert dataframe to glue DynamicFrame and write the output in Parquet format partitioned on toppage column\n",
    "useractivity = DynamicFrame.fromDF(dataframe0, glueContext, \"name1\")\n",
    "\n",
    "writeUseractivityToS3 = glueContext.write_dynamic_frame.from_options(frame = useractivity, connection_type = \"s3\", connection_options = {\"path\": 's3://^ingestionbucket^/weblogs/useractivityconverted', \"partitionKeys\" :[\"toppage\"]}, format = \"parquet\", transformation_ctx = \"writeUseractivityToS3\")\n",
    "\n",
    "job.commit()\n",
    "\n",
    "dataframe0.show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Results\n",
    "To to the S3 bucket to view the results of the transformation and continue with the lab instructions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lab - Join and relationalize data with AWS Glue\n",
    "In this lab we will take two different datasets from different source systems and merge them to prepare a table that combines both useractivity and user profile datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "job = Job(glueContext)\n",
    "job.init('^stackname^-exercise2')\n",
    "\n",
    "## @ useractivity dynamicframe\n",
    "useractivity = glueContext.create_dynamic_frame.from_catalog(database = \"weblogs\", table_name = \"useractivity\", transformation_ctx = \"useractivity\")\n",
    "\n",
    "## @ applymappings to the dynamicframe to make sure we have the correct data types and column names\n",
    "applymapping1 = ApplyMapping.apply(frame = useractivity, mappings = [(\"ip_address\", \"string\", \"ip_address\", \"string\"), (\"username\", \"string\", \"username\", \"string\"), (\"timestamp\", \"string\", \"timestamp\", \"string\"), (\"request\", \"string\", \"request\", \"string\"), (\"http\", \"long\", \"http\", \"long\"), (\"bytes\", \"long\", \"bytes\", \"long\")], transformation_ctx = \"applymapping1\")\n",
    "\n",
    "## @ resolve any issues with column data types\n",
    "resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = \"make_struct\", transformation_ctx = \"resolvechoice2\")\n",
    "\n",
    "## @ drop any null fields\n",
    "useractivity = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = \"useractivity\")\n",
    "\n",
    "## @ create userprofile dynamicframe\n",
    "userprofile = glueContext.create_dynamic_frame.from_catalog(database=\"weblogs\", table_name=\"userprofile\")\n",
    "\n",
    "## @ we will only keep the fields that we want and drop the rest and rename username to dy_username\n",
    "userprofile = userprofile.drop_fields(['cc', 'password', 'ssn', 'email', 'phone','ip_address'])\n",
    "userprofile = userprofile.rename_field('username','dy_username')\n",
    "\n",
    "## @ as the data types in different datasets are different we are going to convert all column to string\n",
    "## @ The Glue build in transform ApplyMapping, Maps source columns and data types from a DynamicFrame to target columns and data types \n",
    "## @ in a returned DynamicFrame. You specify the mapping argument, which is a list of tuples that contain source column, source type, \n",
    "## @ target column, and target type. In the below case we are converting the data types for zip and age to string and updating the column\n",
    "## @ names for first_name & last_name\n",
    "userprofile = ApplyMapping.apply(frame = userprofile, \n",
    "mappings = [(\"first_name\", \"string\", \"firstname\", \"string\"), \n",
    "(\"dy_username\", \"string\", \"dy_username\", \"string\"), \n",
    "(\"zip\", \"bigint\", \"zip\", \"string\"), \n",
    "(\"age\", \"bigint\", \"age\", \"string\"), \n",
    "(\"gender\", \"string\", \"gender\", \"long\"),\n",
    "(\"last_name\", \"string\", \"lastname\", \"long\")\n",
    "], transformation_ctx = \"userprofile\")\n",
    "\n",
    "## @join useractivity and userprofile datasets to create one file and drop the duplicate column dy_username\n",
    "joined = Join.apply(userprofile, useractivity, 'dy_username', 'username').drop_fields(['dy_username'])\n",
    "\n",
    "glueContext.write_dynamic_frame.from_options(frame = joined,\n",
    "          connection_type = \"s3\",\n",
    "          connection_options = {\"path\": 's3://^ingestionbucket^/weblogs/joindatasets'},\n",
    "          format = \"parquet\")\n",
    "\n",
    "job.commit()\n",
    "\n",
    "print 'Job Complete'\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = DynamicFrame.toDF(joined)\n",
    "\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Results\n",
    "Continue with the lab to explore the resulting table in the data lake."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a UDF to simplify apply a hash function to columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import udf\n",
    "from pyspark.sql.types import StringType\n",
    "import hashlib\n",
    "from dateutil.parser import parse\n",
    "\n",
    "def hash_cc(s):\n",
    "    return hashlib.sha256(s).hexdigest()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "job = Job(glueContext)\n",
    "job.init('^stackname^-exercise3')\n",
    "\n",
    "datasource0 = glueContext.create_dynamic_frame.from_catalog(database = \"weblogs\", table_name = \"userprofile\", transformation_ctx = \"datasource0\")\n",
    "\n",
    "\n",
    "## @convert glue DynamicFrame to DataFrame to manipulate the columns\n",
    "dataframe0 = DynamicFrame.toDF(datasource0)\n",
    "\n",
    "hash_cc_f = udf(lambda x: hash_cc(x), StringType())\n",
    "\n",
    "dataframe0 = dataframe0.withColumn(\"hash_cc\", hash_cc_f(dataframe0[\"cc\"])).withColumn(\"hash_ssn\", hash_cc_f(dataframe0[\"ssn\"]))\n",
    "dataframe0 = dataframe0.drop('cc').drop('ssn').drop('password')\n",
    "\n",
    "## @convert dataframe to glue DynamicFrame and write the output in Parquet format\n",
    "datasource1 = DynamicFrame.fromDF(dataframe0, glueContext, \"name1\")\n",
    "\n",
    "\n",
    "datasink4 = glueContext.write_dynamic_frame.from_options(frame = datasource1, connection_type = \"s3\", connection_options = {\"path\": 's3://^ingestionbucket^/weblogs/userprofile-secure'}, format = \"parquet\", transformation_ctx = \"datasink4\")\n",
    "\n",
    "job.commit()\n",
    "\n",
    "print 'Job Complete'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dataframe0.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise 4 - Lookup\n",
    "\n",
    "You are on your own now! Using you pyspark skills, create a simple UDF that performs a lookup on the data.\n",
    "\n",
    "In this case, we're going create a lookup for the geocode of the IP Address. Instead of calling out to a geocoding service, we'll just return US if the first value is less than 100 and UK if the value is 100 or greater.\n",
    "\n",
    "Using this function, create a lookup that creates a copy of the useractivity table that includes the country."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def geocode(ip):\n",
    "    if int(ip.split('.')[0]) < 100:\n",
    "        return \"US\"\n",
    "    else:\n",
    "        return \"UK\"\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "job = Job(glueContext)\n",
    "job.init('Job4')\n",
    "\n",
    "# Write Transformation Code here\n",
    "\n",
    "job.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Sparkmagic (PySpark)",
   "language": "",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 2
   },
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}