Silver_Layer_Notebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2e815116-2c78-4393-9401-616e0c2b9796",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "# This Notebook is Made to Create Silver Layer Tables from Bronze Layer Tables as Cleaned and Normalized Tables "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3245a496-e577-4752-a278-2f0d8e7b952b",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Importing Packages and Config Files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5351b1d2-b3c8-4969-99c5-8e0c8069b5a0",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "import logging\n",
    "import time\n",
    "import datetime\n",
    "import json\n",
    "import re\n",
    "import requests\n",
    "from pyspark.sql import SQLContext\n",
    "from collections import defaultdict\n",
    "from pyspark.sql.functions import countDistinct\n",
    "from pyspark.sql.types import IntegerType,BooleanType,DateType,StringType,VarcharType,LongType,MapType,ArrayType,StructType,TimestampType,TimestampNTZType,DataType\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql import functions as F, Window,udf\n",
    "from pyspark.sql.functions import concat,when,md5\n",
    "from pyspark.sql.functions import explode, col\n",
    "from pyspark.sql import Row\n",
    "import pyspark.sql.types as T\n",
    "from pyspark.sql import DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b08524c3-4da4-4b49-b8d0-a74d029f09f3",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Getting Variables and Dataframes from config notebook\n",
    "%run /Workspace/Users/cihangiray.oner@gmail.com/congif.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "d5b9cd86-0166-41d9-b4a5-d77ed601d1fe",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Reading Tables from Bronze Layer (Only Getting Needed Columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e00beb48-51dc-4b62-bf4d-5fe8c595fe53",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "def check_if_table_exists(read_table_name,read_schema_name):\n",
    "    \"\"\"\n",
    "    Checks if a table exists in the spark catalog.\n",
    "\n",
    "    Args:\n",
    "        table_name (str): The name of the table to check.\n",
    "\n",
    "    Returns:\n",
    "        bool: True if the table exists, False otherwise.\n",
    "    \"\"\"\n",
    "    return spark.catalog.tableExists(f\"hive_metastore.{read_schema_name}.{read_table_name}\")\n",
    "\n",
    "\n",
    "def read_table(dataframe_name,schema_name,table_name,columns_to_read,filter_value = \"N/A\",fill_value = \"N/A\"):\n",
    "    \"\"\"\n",
    "    Reads table and assign it to a dataframe from databricks schemas.\n",
    "\n",
    "    Args:\n",
    "        schema_name (str): The name of the schema to read table from.\n",
    "        table_name (str): The name of the table to read data from.\n",
    "        columns_to_read (list): List containing column names from table to be read\n",
    "        fill_value (str): String value to be used in null value conversion\n",
    "        filter_value (str): Filter parameter for reading table with filtering by spesific values\n",
    "\n",
    "    Returns:\n",
    "        spark.DataFrame: DataFrame after reading table gets into DataFrame.\n",
    "    \"\"\"\n",
    "    #check if table exists\n",
    "    if check_if_table_exists(table_name,schema_name):\n",
    "        dataframe_name = spark.table(f\"hive_metastore.{schema_name}.{table_name}\").select(columns_to_read).na.fill(f'{fill_value}')\n",
    "    else:\n",
    "        print(f\"Read table not found hive_metastore.{schema_name}.{table_name}\")\n",
    "    return dataframe_name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "35d20d2d-3c97-4304-bd79-53082b382bb5",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "stream",
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "in_app_purchase_df\nlogin_df\nmultiplayer_battle_df\nnew_user_df\nsession_started_df\nship_transaction_df\n"
     ]
    }
   ],
   "source": [
    "for key,value in table_names_list.items():\n",
    "    print(key)\n",
    "    # Creating Condition to apply deduplication operations for new_user and login tables on read\n",
    "    if key == \"new_user_df\" or key == \"login_df\":\n",
    "        vars() [key] = read_table(key,first_layer_schema_name,f\"{key}_bronze_layer_managed_table\",silver_layer_col_names[key])\n",
    "        vars() [key] = vars() [key].dropDuplicates(['USER_ID'])\n",
    "    # Creating dataframes by create_dataframe function\n",
    "    else:\n",
    "        vars() [key] = read_table(key,first_layer_schema_name,f\"{key}_bronze_layer_managed_table\",silver_layer_col_names[key])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "0ce8f3ee-c9fc-4950-8821-3f3fa4c868d3",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Creating NEW_USER Dim Table (Deduplicating USER_ID)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e1cf2df3-d9a9-4b94-8eff-972ee6e89b90",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "### Creating USER_ID Dim Table (Getting Unique User Id`s from Session Started Table to get all Users does not exists on New User Table)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3f186b72-6a19-4d5b-9254-7a0d1e1f8a63",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# silver_layer_col_names_1 dict from config.py holding table name and column names\n",
    "for key,value in silver_layer_col_names_1.items():\n",
    "    # Selecting needed columns from Session Table\n",
    "    vars()[key] = session_started_df.select(value)\n",
    "    # Renaming USER_ID column for smooth write/join operation\n",
    "    vars()[key] = vars()[key].withColumnRenamed('USER_ID','SESSION_USER_ID').withColumnRenamed('Join_Key','SESSION_Join_Key')\n",
    "    # Removing USER_IDs belongs to new users and selecting only needed columns\n",
    "    vars()[key] = vars()[key].join(new_user_df,vars()[key].SESSION_USER_ID == new_user_df.USER_ID,\"left_anti\").select('SESSION_USER_ID','USER_IS_SPENDER','USER_GEO_LOCATION','SESSION_Join_Key')\n",
    "    # Getting Unique USER_ID`s by Dropping Duplicates on USER_ID column\n",
    "    vars()[key] = vars()[key].dropDuplicates(['SESSION_USER_ID'])\n",
    "    # Getting back to proper naming for further table naming operations\n",
    "    vars()[key] = vars()[key].withColumnRenamed('SESSION_USER_ID','USER_ID').withColumnRenamed('SESSION_Join_Key','Join_Key')\n",
    "    \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f0821b74-1db5-4c0b-815c-fb4f040ad09f",
     "showTitle": false,
     "title": ""
    }
   },
   "source": [
    "## Writing Dataframes into Hive Metastore Silver Layer Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a281f839-62e0-4d3d-aac2-15f0aab057f0",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "data": {
      "text/plain": [
       ""
      ]
     },
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "arguments": {},
       "data": "",
       "errorSummary": "Command skipped",
       "errorTraceType": "ansi",
       "metadata": {},
       "type": "ipynbError"
      }
     },
     "output_type": "display_data"
    }
   ],
   "source": [
    "def create_schema(second_layer_schema_name,location_name):\n",
    "    \"\"\"\n",
    "    Checks if a schema exists in the spark catalog.\n",
    "\n",
    "    Args:\n",
    "        location_name (str): catalog name to save schema on\n",
    "        second_layer_schema_name (str): The name of the schema to check.\n",
    "\n",
    "    Returns:\n",
    "        DataFrame: Empty DF creates schema on defined catalog if not exits\n",
    "    \"\"\"\n",
    "    return spark.sql(f\"CREATE SCHEMA IF NOT EXISTS  {location_name}.{second_layer_schema_name}\")\n",
    "\n",
    "\n",
    "def check_if_table_exists(second_layer_schema_name, table_name):\n",
    "    \"\"\"\n",
    "    Checks if a table exists in the spark catalog.\n",
    "\n",
    "    Args:\n",
    "        table_name (str): The name of the table to check.\n",
    "\n",
    "    Returns:\n",
    "        bool: True if the table exists, False otherwise.\n",
    "    \"\"\"\n",
    "    return spark.catalog.tableExists(f\"hive_metastore.{second_layer_schema_name}.{table_name}_silver_layer_managed_table\")\n",
    "\n",
    "def write_to_managed_table(df, table_name, second_layer_schema_name, location_name, mode = \"overwrite\"):\n",
    "    \"\"\"\n",
    "    Writes a DataFrame to a managed table in Delta Lake.\n",
    "\n",
    "    If the table exists and mode is overwrite, it performs an overwrite operation.\n",
    "    Otherwise, it either creates a new table or appends transactions to table based on the `mode` parameter.\n",
    "\n",
    "    Args:\n",
    "        df (pyspark.sql.DataFrame): The DataFrame to write to the table.\n",
    "        table_name (str): The name of the target table.\n",
    "        second_layer_schema_name (str): The schema name of target table.\n",
    "        location_name (str): Catalog name to save schema on\n",
    "        mode (str, optional): The write mode.\n",
    "    \"\"\"\n",
    "    #create schema if not exists\n",
    "    create_schema(second_layer_schema_name,location_name)\n",
    "    # check if the table exists\n",
    "    if check_if_table_exists(second_layer_schema_name, table_name):\n",
    "        print(f\"Table exists on hive_metastore.{second_layer_schema_name}.{table_name}_silver_layer_managed_table\")\n",
    "        if mode == \"overwrite\":\n",
    "            print(f\"Overwriting all transactions on managed table hive_metastore.{second_layer_schema_name}.{table_name}_silver_layer_managed_table\")\n",
    "            df.write.format(\"delta\").option(\"delta.columnMapping.mode\", \"name\").mode(mode).saveAsTable(f\"hive_metastore.{second_layer_schema_name}.{table_name}_silver_layer_managed_table\")\n",
    "        else:\n",
    "            print(f\"Appending all transactions on managed table hive_metastore.{second_layer_schema_name}.{table_name}_silver_layer_managed_table\")\n",
    "            df.write.format(\"delta\").option(\"delta.columnMapping.mode\", \"name\").mode(mode).saveAsTable(f\"hive_metastore.{second_layer_schema_name}.{table_name}_silver_layer_managed_table\")\n",
    "    else:\n",
    "        print(f\"Writing to managed table hive_metastore.{second_layer_schema_name}.{table_name}_silver_layer_managed_table\")\n",
    "        df.write.format(\"delta\").option(\"delta.columnMapping.mode\", \"name\").saveAsTable(f\"hive_metastore.{second_layer_schema_name}.{table_name}_silver_layer_managed_table\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5f89f09f-1e14-491e-9e06-4d39e27a476d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [
    {
     "output_type": "display_data",
     "data": {
      "text/plain": [
       ""
      ]
     },
     "metadata": {
      "application/vnd.databricks.v1+output": {
       "arguments": {},
       "data": "",
       "errorSummary": "Command skipped",
       "errorTraceType": "ansi",
       "metadata": {},
       "type": "ipynbError"
      }
     },
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Getting active dataframe names from ongoing session (One-time cache)\n",
    "def list_dataframes():\n",
    "    \"\"\"\n",
    "    Lists of dataframes names as list.\n",
    "\n",
    "    Returns:\n",
    "        list: list consist of names.\n",
    "    \"\"\"\n",
    "    return [k for (k, v) in globals().items() if isinstance(v, DataFrame)]\n",
    "\n",
    "\n",
    "for df_name in list_dataframes():\n",
    "    print(df_name)\n",
    "    # vars() [df_name] = vars() [df_name].drop('SESSION_ID').drop('USER_ID')\n",
    "    # Writing dataframes to 2nd layer on defined schema and table\n",
    "    write_to_managed_table(vars()[df_name], df_name,  second_layer_schema_name, location_name)"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookMetadata": {
    "mostRecentlyExecutedCommandWithImplicitDF": {
     "commandId": 2570834028859983,
     "dataframes": [
      "_sqldf"
     ]
    },
    "pythonIndentUnit": 4
   },
   "notebookName": "Silver_Layer_Notebook",
   "widgets": {}
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}