ETL_Pipeline_Bidnamic_task-checkpoint.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 143,
   "metadata": {},
   "outputs": [],
   "source": [
    "import psycopg2 as pg2\n",
    "import json\n",
    "import pyspark\n",
    "import pyspark.sql.functions as f\n",
    "from pyspark.sql.functions import col, lit, current_date,current_timestamp,monotonically_increasing_id,desc,expr,split\n",
    "from pyspark.sql.types import StringType, BooleanType, IntegerType, FloatType, DateType\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 144,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark=SparkSession.builder.appName('ETL Pipeline').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# #this sparksession used for S3 bucket configuring\n",
    "# spark=SparkSession.builder\\\n",
    "#             .appName(\"ETL\")\\\n",
    "#             .config(\"spark.hadoop.fs.s3a.access.key\", \"****************\")\\\n",
    "#             .config(\"spark.hadoop.fs.s3a.secret.key\", '*****************')\\\n",
    "#             .config(\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")\\\n",
    "#             .config(\"spark.jars\", 'hadoop-aws-3.3.1.jar,aws-java-sdk-bundle-1.11.901.jar','spark-postgres-jar')\\\n",
    "#             .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Extraction\n",
    "\n",
    "<h4>These files are extracted from AWS S3 bucket and also given the local input file path as an optional in the case of exception due to the ****HADOOP JAR VERSION CAMPATIBILTY ISSUES**** by this we can choose any inpupath either s3 or local, where these are considred as RAW csv data files loaded to S3 from given URL </h4>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---------------+-------+--------------------+----------+--------------------+\n",
      "|campaign_id|structure_value| status|   created_timestamp|created_by|  modified_timestamp|\n",
      "+-----------+---------------+-------+--------------------+----------+--------------------+\n",
      "| 1578451881|          venum|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578451584|        ellesse|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578451386|       converse|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578412457|         wilson|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 9872103720|         wham-o|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 9872103720|         wham-o|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578451386|       converse|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578451623|       spalding|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578451386|       converse|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 9872103720|         wham-o|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578411800|           nike|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 9872103720|         wham-o|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 8677963439|           net1|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578629887|         adidas|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578630373|         reebok|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1589980762|          kappa|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578411800|           nike|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1589980762|          kappa|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1589980762|          kappa|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1589980762|          kappa|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "+-----------+---------------+-------+--------------------+----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "root\n",
      " |-- campaign_id: long (nullable = true)\n",
      " |-- structure_value: string (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- created_timestamp: timestamp (nullable = false)\n",
      " |-- created_by: string (nullable = false)\n",
      " |-- modified_timestamp: timestamp (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "campaigns_df = spark.read.option('header',True).csv(aws_config['inputpath']['local_campaigns'])\n",
    "campaigns_df1 = campaigns_df.withColumn(\"created_timestamp\", current_timestamp())\\\n",
    "    .withColumn(\"campaign_id\",col('campaign_id').cast('bigint'))\\\n",
    "    .withColumn(\"created_by\",lit('abspark'))\\\n",
    "    .withColumn(\"modified_timestamp\",current_timestamp())\\\n",
    "    .withColumn(\"created_by\",lit('abspark'))\\\n",
    "    .withColumn(\"modified_timestamp\",current_timestamp())\n",
    "campaigns_df1.show()\n",
    "campaigns_df1.printSchema()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----------+--------------------+-------+--------------------+----------+--------------------+\n",
      "| ad_group_id|campaign_id|               alias| status|   created_timestamp|created_by|  modified_timestamp|\n",
      "+------------+-----------+--------------------+-------+--------------------+----------+--------------------+\n",
      "| 66372665454| 1578451881|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 84481260174| 1578451584|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 59624654596| 1578451386|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 59977520149| 1578412457|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|102171970298| 9872103720|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|102171970298| 9872103720|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 58309319903| 1578451386|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|119487436958| 1578451623|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 59617309655| 1578451386|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|102171970298| 9872103720|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 61228310066| 1578411800|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|102171970298| 9872103720|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 89733540689| 8677963439|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 81471926123| 1578629887|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|104471629643| 1578630373|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|102171967178| 1589980762|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 81713176441| 1578411800|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|102171967178| 1589980762|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|102171967178| 1589980762|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|102171967178| 1589980762|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "+------------+-----------+--------------------+-------+--------------------+----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "root\n",
      " |-- ad_group_id: long (nullable = true)\n",
      " |-- campaign_id: long (nullable = true)\n",
      " |-- alias: string (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- created_timestamp: timestamp (nullable = false)\n",
      " |-- created_by: string (nullable = false)\n",
      " |-- modified_timestamp: timestamp (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "adgroups_df = spark.read.option('header',True).csv(aws_config['inputpath']['local_adgroups'])\n",
    "adgroups_df1 = adgroups_df.withColumn(\"created_timestamp\", current_timestamp())\\\n",
    "    .withColumn(\"created_by\",lit('abspark'))\\\n",
    "    .withColumn(\"ad_group_id\",col('ad_group_id').cast('bigint'))\\\n",
    "    .withColumn(\"campaign_id\",col('campaign_id').cast('bigint'))\\\n",
    "    .withColumn(\"modified_timestamp\",current_timestamp())\\\n",
    "    .withColumn(\"created_by\",lit('abspark'))\\\n",
    "    .withColumn(\"modified_timestamp\",current_timestamp())\n",
    "adgroups_df1.show()\n",
    "adgroups_df1.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 148,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- date: date (nullable = true)\n",
      " |-- ad_group_id: long (nullable = true)\n",
      " |-- campaign_id: long (nullable = true)\n",
      " |-- clicks: integer (nullable = true)\n",
      " |-- cost: double (nullable = true)\n",
      " |-- conversion_value: double (nullable = true)\n",
      " |-- conversions: integer (nullable = true)\n",
      " |-- search_term: string (nullable = true)\n",
      " |-- created_timestamp: timestamp (nullable = false)\n",
      " |-- search_terms_id: long (nullable = false)\n",
      " |-- modified_timestamp: timestamp (nullable = false)\n",
      " |-- created_by: string (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "search_terms_df = spark.read.option('header',True).csv(aws_config['inputpath']['local_search_terms'])\n",
    "search_terms_df1 = search_terms_df.withColumn(\"created_timestamp\", current_timestamp())\\\n",
    "    .withColumn(\"search_terms_id\",monotonically_increasing_id())\\\n",
    "    .withColumn(\"ad_group_id\",col('ad_group_id').cast('bigint'))\\\n",
    "    .withColumn(\"campaign_id\",col('campaign_id').cast('bigint'))\\\n",
    "    .withColumn('date',col('date').cast('date'))\\\n",
    "    .withColumn(\"clicks\",col('clicks').cast('int'))\\\n",
    "    .withColumn(\"cost\",col('cost').cast('double'))\\\n",
    "    .withColumn(\"conversion_value\",col('conversion_value').cast('double'))\\\n",
    "    .withColumn(\"conversions\",col('conversions').cast('int'))\\\n",
    "    .withColumn(\"modified_timestamp\",current_timestamp())\\\n",
    "    .withColumn(\"created_by\",lit('abspark'))\\\n",
    "    .withColumn(\"modified_timestamp\",current_timestamp())\n",
    "# search_terms_df1.show()\n",
    "search_terms_df1.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data cleaning (Removing Duplicates) for loading data into STORE data  schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###############################################################\n",
    "### \"DISTINCT data of Campaigns\"\n",
    "###############################################################"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The DUPLICATE and DISTINCT data of Campaigns\n",
      "+-----------+---------------+-------+--------------------+----------+--------------------+\n",
      "|campaign_id|structure_value| status|   created_timestamp|created_by|  modified_timestamp|\n",
      "+-----------+---------------+-------+--------------------+----------+--------------------+\n",
      "| 9903945998|           nerf|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 8677993568|      smellwell|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1585315601|        arsenal|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578451362|           beco|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578451356|       converse|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578411749|     cartasport|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 9564222913|  england rugby|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578629923|    fitness mad|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1645615867|       berghaus|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|11156611436|  juicy couture|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1744538394|           vans|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578412481|       zoo york|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578411497|      golds gym|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|12583609321|          umbro|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578412217|    new balance|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1721712416|             d3|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578630121|        optimum|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578451665|     runderwear|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578630385|            ufe|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 1578629884|         adidas|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "+-----------+---------------+-------+--------------------+----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "Total number of ACTUAL rows in Search Terms DF: 221855\n",
      "Total number of DUPLICATE rows in Search Terms DF: 221503\n",
      "Total number of UNIQUE rows in Search Terms DF: 352\n"
     ]
    }
   ],
   "source": [
    "print('The DUPLICATE and DISTINCT data of Campaigns')\n",
    "campaigns_unique = campaigns_df1.dropDuplicates()\n",
    "campaigns_duplicate = campaigns_df1.exceptAll(campaigns_df1.dropDuplicates(['campaign_id','structure_value','status']))\n",
    "campaigns_unique.show()\n",
    "print(f\"Total number of ACTUAL rows in Search Terms DF: {campaigns_df1.count()}\")\n",
    "print(f\"Total number of DUPLICATE rows in Search Terms DF: {campaigns_duplicate.count()}\")\n",
    "print(f\"Total number of UNIQUE rows in Search Terms DF: {campaigns_unique.count()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###############################################################\n",
    "### \"DISTINCT data of Adgroups\"\n",
    "###############################################################"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 150,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The DUPLICATE and DISTINCT data of Adgroups\n",
      "+------------+-----------+--------------------+-------+--------------------+----------+--------------------+\n",
      "| ad_group_id|campaign_id|               alias| status|   created_timestamp|created_by|  modified_timestamp|\n",
      "+------------+-----------+--------------------+-------+--------------------+----------+--------------------+\n",
      "| 61108919258| 1578412214|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|124352384626| 1578630361|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|103621538761| 1578411800|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 64246021402| 1578630391|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 60221413330| 1578630223|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 79415425560| 1578630238|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|100197881658| 1578630361|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|107353692955| 1578629887|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 67714968891| 1578412016|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 59895846912| 1578451581|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 58479980566| 1578630127|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 68075790428| 1578412214|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 56397124221| 1578412211|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|116947807623|12428869601|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 61671879343| 1578629881|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 72747813330| 1578411800|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 58315963623| 1578630361|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 59446353294| 1578411800|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "|136146566628| 1578629887|Shift - Shopping ...|ENABLED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "| 59617338135| 1578451386|Shift - Shopping ...|REMOVED|2022-02-03 00:48:...|   abspark|2022-02-03 00:48:...|\n",
      "+------------+-----------+--------------------+-------+--------------------+----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n",
      "Total number of ACTUAL rows in Search Terms DF: 221855\n",
      "Total number of DUPLICATE rows in Search Terms DF: 202584\n",
      "Total number of UNIQUE rows in Search Terms DF: 19271\n"
     ]
    }
   ],
   "source": [
    "print('The DUPLICATE and DISTINCT data of Adgroups')\n",
    "adgroups_unique = adgroups_df1.dropDuplicates()\n",
    "adgroups_duplicate = adgroups_df1.exceptAll(adgroups_df1.dropDuplicates())\n",
    "adgroups_unique.show()\n",
    "\n",
    "print(f\"Total number of ACTUAL rows in Search Terms DF: {adgroups_df1.count()}\")\n",
    "print(f\"Total number of DUPLICATE rows in Search Terms DF: {adgroups_duplicate.count()}\")\n",
    "print(f\"Total number of UNIQUE rows in Search Terms DF: {adgroups_unique.count()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "###############################################################\n",
    "### \"DISTINCT data of Search Terms\"\n",
    "###############################################################"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The duplicate and distinct data of Search Terms\n",
      "+----------+------------+-----------+------+----+----------------+-----------+--------------------+--------------------+---------------+--------------------+----------+\n",
      "|      date| ad_group_id|campaign_id|clicks|cost|conversion_value|conversions|         search_term|   created_timestamp|search_terms_id|  modified_timestamp|created_by|\n",
      "+----------+------------+-----------+------+----+----------------+-----------+--------------------+--------------------+---------------+--------------------+----------+\n",
      "|2019-01-31| 65082925368| 1578630361|     1|0.01|             0.0|          0|puma archive t7 t...|2022-02-03 00:48:...|            129|2022-02-03 00:48:...|   abspark|\n",
      "|2018-10-25| 59410043037| 1578411797|     1|0.06|             0.0|          0|nike sunray prote...|2022-02-03 00:48:...|            797|2022-02-03 00:48:...|   abspark|\n",
      "|2021-10-28|112825649726| 1645615867|     1|0.25|             0.0|          0|     berghaus jacket|2022-02-03 00:48:...|            924|2022-02-03 00:48:...|   abspark|\n",
      "|2019-09-16| 80593763674| 1578630361|     1|0.01|             0.0|          0|man city 125 shir...|2022-02-03 00:48:...|           1491|2022-02-03 00:48:...|   abspark|\n",
      "|2020-08-19|107057616860| 1578629887|     1|0.19|             0.0|          0| adidas cricket bats|2022-02-03 00:48:...|           1500|2022-02-03 00:48:...|   abspark|\n",
      "|2019-05-25| 63529724844| 1578451623|     1| 0.4|             0.0|          0|    amzon basketball|2022-02-03 00:48:...|           1700|2022-02-03 00:48:...|   abspark|\n",
      "|2019-01-03| 68301659491| 1578411800|     1|0.01|             0.0|          0|24 36 months nike...|2022-02-03 00:48:...|           1757|2022-02-03 00:48:...|   abspark|\n",
      "|2020-04-17| 58080929005| 1578411800|     1|0.08|             0.0|          0|barcelona 2017 ma...|2022-02-03 00:48:...|           2246|2022-02-03 00:48:...|   abspark|\n",
      "|2020-01-21| 84080465211| 1578411800|     2|0.36|             0.0|          0|nike childrens st...|2022-02-03 00:48:...|           2270|2022-02-03 00:48:...|   abspark|\n",
      "|2021-08-22| 59933784179| 1578451623|     1| 0.1|             0.0|          0|      nba baskitball|2022-02-03 00:48:...|           2284|2022-02-03 00:48:...|   abspark|\n",
      "|2019-02-19| 65122873145| 1578411995|     1|0.38|             0.0|          0|cricket shoes sal...|2022-02-03 00:48:...|           2409|2022-02-03 00:48:...|   abspark|\n",
      "|2018-11-18| 59897200536| 1578411797|     1|0.17|             0.0|          0|sports direct nik...|2022-02-03 00:48:...|           2564|2022-02-03 00:48:...|   abspark|\n",
      "|2019-02-09| 58760111239| 1578629887|     1|0.36|             0.0|          0|manchester united...|2022-02-03 00:48:...|           2634|2022-02-03 00:48:...|   abspark|\n",
      "|2018-11-20| 59334967853| 1578629887|     1|0.03|             0.0|          0| sliders kids adidas|2022-02-03 00:48:...|           2814|2022-02-03 00:48:...|   abspark|\n",
      "|2019-10-29| 80841726476| 1578411800|     1|0.08|             0.0|          0|    nike baby onesie|2022-02-03 00:48:...|           2854|2022-02-03 00:48:...|   abspark|\n",
      "|2019-11-13| 76826903415| 1578411797|     1|0.09|             0.0|          0|          nike snood|2022-02-03 00:48:...|           2868|2022-02-03 00:48:...|   abspark|\n",
      "|2020-05-12| 80841726476| 1578411800|     1| 0.1|             0.0|          0|      nike baby grow|2022-02-03 00:48:...|           2999|2022-02-03 00:48:...|   abspark|\n",
      "|2021-01-09|100333423909| 1578411794|     1|0.56|             0.0|          0|infant nike kawa ...|2022-02-03 00:48:...|           3130|2022-02-03 00:48:...|   abspark|\n",
      "|2021-09-01|120402204229| 1578412283|     1|0.71|             0.0|          0|   long jane wetsuit|2022-02-03 00:48:...|           3174|2022-02-03 00:48:...|   abspark|\n",
      "|2018-12-01| 58479769406| 1578411755|     1|0.09|             0.0|          0|running spikes re...|2022-02-03 00:48:...|           3552|2022-02-03 00:48:...|   abspark|\n",
      "+----------+------------+-----------+------+----+----------------+-----------+--------------------+--------------------+---------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n",
      "Total number of ACTUAL row's in Search Terms DF: 221855\n",
      "Total number of DUPLICATE row's in Search Terms DF: 0\n",
      "Total number of UNIQUE row's in Search Terms DF: 221855\n"
     ]
    }
   ],
   "source": [
    "print('The duplicate and distinct data of Search Terms')\n",
    "search_terms_unique = search_terms_df1.dropDuplicates()\n",
    "search_terms_duplicate = search_terms_df1.exceptAll(search_terms_df1.dropDuplicates())\n",
    "search_terms_unique.show()\n",
    "print(f\"Total number of ACTUAL row's in Search Terms DF: {search_terms_df1.count()}\")\n",
    "print(f\"Total number of DUPLICATE row's in Search Terms DF: {search_terms_duplicate.count()}\")\n",
    "print(f\"Total number of UNIQUE row's in Search Terms DF: {search_terms_unique.count()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 152,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- campaign_id: long (nullable = true)\n",
      " |-- structure_value: string (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- created_timestamp: timestamp (nullable = false)\n",
      " |-- created_by: string (nullable = false)\n",
      " |-- modified_timestamp: timestamp (nullable = false)\n",
      "\n",
      "root\n",
      " |-- ad_group_id: long (nullable = true)\n",
      " |-- campaign_id: long (nullable = true)\n",
      " |-- alias: string (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- created_timestamp: timestamp (nullable = false)\n",
      " |-- created_by: string (nullable = false)\n",
      " |-- modified_timestamp: timestamp (nullable = false)\n",
      "\n",
      "root\n",
      " |-- date: date (nullable = true)\n",
      " |-- ad_group_id: long (nullable = true)\n",
      " |-- campaign_id: long (nullable = true)\n",
      " |-- clicks: integer (nullable = true)\n",
      " |-- cost: double (nullable = true)\n",
      " |-- conversion_value: double (nullable = true)\n",
      " |-- conversions: integer (nullable = true)\n",
      " |-- search_term: string (nullable = true)\n",
      " |-- created_timestamp: timestamp (nullable = false)\n",
      " |-- search_terms_id: long (nullable = false)\n",
      " |-- modified_timestamp: timestamp (nullable = false)\n",
      " |-- created_by: string (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "campaigns_unique.printSchema()\n",
    "adgroups_unique.printSchema()\n",
    "search_terms_unique.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Transformations for REPORT schema data\n",
    "\n",
    "<h4>These trasnformed dataframes will be loaded into report data schema for business requirement</h4>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ad_group_id: long (nullable = true)\n",
      " |-- campaign_id: long (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- created_timestamp: timestamp (nullable = false)\n",
      " |-- created_by: string (nullable = false)\n",
      " |-- modified_timestamp: timestamp (nullable = false)\n",
      " |-- country: string (nullable = true)\n",
      " |-- priority: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "adgroups_split = adgroups_unique.withColumn(\"created_timestamp\", current_timestamp())\\\n",
    "    .withColumn(\"created_by\",lit('abspark'))\\\n",
    "    .withColumn(\"ad_group_id\",col('ad_group_id').cast('bigint'))\\\n",
    "    .withColumn(\"country\", split(col(\"alias\"), \" - \").getItem(2))\\\n",
    "    .withColumn(\"priority\", split(col(\"alias\"), \" - \").getItem(4))\\\n",
    "    .withColumn(\"modified_timestamp\",current_timestamp())\\\n",
    "    .withColumn(\"created_by\",lit('abspark'))\\\n",
    "    .withColumn(\"modified_timestamp\",current_timestamp()).drop('alias')\n",
    "adgroups_split.count()\n",
    "adgroups_split.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### \"final_agg \"data frame is created according to the task brief .md file in git repo whre it is end requiremnt for this challange"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 182,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Note: 'roas = return on add spent'\n"
     ]
    }
   ],
   "source": [
    "fianl_agg_df = adgroups_split.join(search_terms_unique, adgroups_split.ad_group_id == search_terms_unique.ad_group_id, 'inner')\\\n",
    "            .select(adgroups_split.ad_group_id,\n",
    "                    adgroups_split.campaign_id,\n",
    "                    adgroups_split.country,\n",
    "                    adgroups_split.priority,\n",
    "                    ((search_terms_unique.conversion_value)/search_terms_unique.cost).alias('roas'),\n",
    "                    search_terms_unique.date,\n",
    "                    adgroups_split.status,\n",
    "                    search_terms_unique.search_term,).filter(col('conversion_value')>0)\\\n",
    "                    .orderBy(adgroups_split.ad_group_id)\n",
    "print(\"Note: 'roas = return on add spent'\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 226,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4918"
      ]
     },
     "execution_count": 226,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "fianl_agg_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 177,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----------+-------+--------+------------------+----------+-------+--------------------+\n",
      "|ad_group_id|campaign_id|country|priority|              roas|      date| status|         search_term|\n",
      "+-----------+-----------+-------+--------+------------------+----------+-------+--------------------+\n",
      "|55067386730| 1578630361|     GB|    HIGH|229.79999999999998|2019-01-27|REMOVED|best green footba...|\n",
      "|55067402050| 1578412016|     GB|    HIGH| 832.6666666666667|2019-04-16|ENABLED|rose gold kicker ...|\n",
      "|55067982370| 1578630361|     GB|    HIGH|474.74999999999994|2019-02-18|REMOVED|   dortmund home top|\n",
      "|55068003730| 1578411800|     GB|    HIGH|            1598.0|2020-05-21|REMOVED|kids nike sweat s...|\n",
      "|55068004690| 1578411800|     GB|    HIGH|            1298.0|2019-03-30|REMOVED|  nike jersey shorts|\n",
      "|55070131090| 1578629881|     GB|     LOW|266.33333333333337|2018-11-18|REMOVED|   adidas crib shoes|\n",
      "|55070262370| 1578451158|     GB|  MEDIUM|30.965517241379313|2019-04-25|REMOVED|barcelona water b...|\n",
      "|55070262370| 1578451158|     GB|  MEDIUM|149.66666666666669|2018-11-30|REMOVED|barcelona water b...|\n",
      "|55070262410| 1578451161|     GB|    HIGH| 81.63636363636364|2019-02-21|REMOVED|barcelona fc drin...|\n",
      "|55070263130| 1578629887|     GB|    HIGH|            4198.0|2019-04-28|REMOVED|adidas havoc boxi...|\n",
      "|55070272730| 1578629887|     GB|    HIGH|1648.9999999999998|2019-08-10|REMOVED|football kits rea...|\n",
      "|55070501490| 1578411113|     GB|  MEDIUM|            62.375|2019-03-06|REMOVED|   acme tornado 2000|\n",
      "|55070504850| 1578411800|     GB|    HIGH|233.11111111111111|2019-02-23|REMOVED|nike womens train...|\n",
      "|55070512930| 1578411800|     GB|    HIGH|             649.0|2019-07-16|REMOVED|nike ankle protec...|\n",
      "|55071219570| 1578451584|     GB|    HIGH| 432.6666666666667|2019-03-15|ENABLED|ellesse vest camo...|\n",
      "|55484458530| 1578411800|     GB|    HIGH|466.33333333333337|2019-05-09|REMOVED|barcelona kit 5 6...|\n",
      "|56395711141| 1578451359|     GB|  MEDIUM|            1249.0|2018-11-21|REMOVED|all black infant ...|\n",
      "|56395713021| 1578411800|     GB|    HIGH|173.57894736842104|2020-08-06|ENABLED|nike pro compression|\n",
      "|56395713341| 1578411797|     GB|  MEDIUM|            1337.0|2020-12-12|REMOVED|       nike pro warm|\n",
      "|56395714981| 1578411800|     GB|    HIGH|            2298.0|2019-01-18|REMOVED|    nike mens tights|\n",
      "+-----------+-----------+-------+--------+------------------+----------+-------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "fianl_agg_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 167,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "19271"
      ]
     },
     "execution_count": 167,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "adgroups_split.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Loading into PostgreSQL \n",
    "\n",
    "<h2> Create three levels of data loading schemas</h2>\n",
    "<ul>\n",
    "    <li>Stage Data schema</li>\n",
    "    <li>Store Data schema</li>\n",
    "    <li>Report Data schema</li>\n",
    "</ul>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 215,
   "metadata": {},
   "outputs": [],
   "source": [
    "aws_config = json.load(open('aws_cofig.json'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 162,
   "metadata": {},
   "outputs": [],
   "source": [
    "#configure postgreSQL to create by using psycopg2\n",
    "conn = pg2.connect(user=aws_config['connection']['user'],\n",
    "                   password=aws_config['connection']['password'],\n",
    "                   host=aws_config['connection']['host'],\n",
    "                   port=aws_config['connection']['port'],\n",
    "                   database = aws_config['connection']['db'] )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 163,
   "metadata": {},
   "outputs": [],
   "source": [
    "conn.autocommit = True\n",
    "cur = conn.cursor()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h3>Creating  stage data schema</h3>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 164,
   "metadata": {},
   "outputs": [],
   "source": [
    "cur.execute(\"\"\" CREATE SCHEMA landingdataS3\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 196,
   "metadata": {},
   "outputs": [],
   "source": [
    "#JDBC driver for loading spark dataframes in to postgreSQL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 216,
   "metadata": {},
   "outputs": [],
   "source": [
    "driver = r\"C:\\Users\\Abhishikth\\OneDrive\\Desktop\\Data Engineering\\ProjectBidnamic\\postgresql-42.3.2.jar\"\n",
    "url = aws_config['connection']['url']\n",
    "user = aws_config['connection']['user']\n",
    "pasword =aws_config['connection']['password']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 218,
   "metadata": {},
   "outputs": [],
   "source": [
    "table1 = \"landingdataS3.campaigns\"\n",
    "table2 = \"landingdataS3.adgroups\"\n",
    "table3 = \"landingdataS3.search_terms\"\n",
    "\n",
    "campaigns_df1.write.format(\"jdbc\").option(\"driver\",driver).option(\"url\",url).option(\"dbtable\",table1)\\\n",
    "            .option(\"mode\",\"append\").save()\n",
    "adgroups_df1.write.format(\"jdbc\").option(\"driver\",driver).option(\"url\",url).option(\"dbtable\",table2)\\\n",
    "            .option(\"mode\",\"append\").save()\n",
    "search_terms_df1.write.format(\"jdbc\").option(\"driver\",driver).option(\"url\",url).option(\"dbtable\",table3)\\\n",
    "            .option(\"mode\",\"append\").save()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating schema for store data\n",
    "<h4>In this schema level duplicates are removed and data types are assigned acordingly</h4>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 221,
   "metadata": {},
   "outputs": [],
   "source": [
    "cur.execute(\"\"\" CREATE SCHEMA storedata\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 220,
   "metadata": {},
   "outputs": [],
   "source": [
    "table4 = \"storedata.campaigns\"\n",
    "table5 = \"storedata.adgroups\"\n",
    "table6 = \"storedata.search_terms\"\n",
    "\n",
    "campaigns_unique.write.format(\"jdbc\").option(\"driver\",driver).option(\"url\",url).option(\"dbtable\",table4)\\\n",
    "            .option(\"mode\",\"append\").save()\n",
    "adgroups_unique.write.format(\"jdbc\").option(\"driver\",driver).option(\"url\",url).option(\"dbtable\",table5)\\\n",
    "            .option(\"mode\",\"append\").save()\n",
    "search_terms_unique.write.format(\"jdbc\").option(\"driver\",driver).option(\"url\",url).option(\"dbtable\",table6)\\\n",
    "            .option(\"mode\",\"append\").save()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Creating schema for Report Data\n",
    "<h4>This schema handles the business required and tranformed data table where these table is created to know ROAS(RETURN ON ADD SPENT) aggregated by country and/or by priority and also for enabling fexible querying on table accroding to the business requirement</h4>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 222,
   "metadata": {},
   "outputs": [],
   "source": [
    "cur.execute(\"\"\" CREATE SCHEMA reportdata\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 227,
   "metadata": {},
   "outputs": [],
   "source": [
    "table7 = \"reportdata.reporting\"\n",
    "fianl_agg_df.write.format(\"jdbc\").option(\"driver\",driver).option(\"url\",url).option(\"dbtable\",table7)\\\n",
    "            .option(\"mode\",\"overwrite\").save()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}