Silver_Layer.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "6c76a167-ff49-4b1d-a3f2-4cd5303dcd60",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.types import *\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4a800722-da5c-4691-8357-e9930808616c",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"ECommerceDataPipeline\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "2b66bfae-59b8-4253-8c52-4dc04cc639b4",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "#/mnt/delta/tables/bronze/users\n",
    "usersDF = spark.read.format(\"delta\").load(\"/mnt/delta/tables/bronze/users\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b57f477e-fc65-42c7-86c8-2ca8942c6e38",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Normalize country codes to uppercase\n",
    "usersDF = usersDF.withColumn(\"countrycode\", upper(col(\"countrycode\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e760ac10-49db-4f7d-9772-1ce987b918e7",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Handling multiple languages elegantly with `expr` and `case when`\n",
    "usersDF = usersDF.withColumn(\"language_full\", \n",
    "                             expr(\"CASE WHEN language = 'EN' THEN 'English' \" +\n",
    "                                  \"WHEN language = 'FR' THEN 'French' \" +\n",
    "                                  \"ELSE 'Other' END\"))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "a3555d3e-43a1-4db6-8a78-ac4594115415",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Correcting potential data entry errors in `gender` column\n",
    "usersDF = usersDF.withColumn(\"gender\", \n",
    "                             when(col(\"gender\").startswith(\"M\"), \"Male\")\n",
    "                             .when(col(\"gender\").startswith(\"F\"), \"Female\")\n",
    "                             .otherwise(\"Other\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "6e567609-c0a4-4338-af88-6cd9ea3bd217",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "# Using `regexp_replace` to clean `civilitytitle` values\n",
    "usersDF = usersDF.withColumn(\"civilitytitle_clean\", \n",
    "                             regexp_replace(\"civilitytitle\", \"(Mme|Ms|Mrs)\", \"Ms\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "49e8ee1c-bbe6-4dfe-83d1-698a874dc92d",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Derive new column `years_since_last_login` from `dayssincelastlogin`\n",
    "usersDF = usersDF.withColumn(\"years_since_last_login\", col(\"dayssincelastlogin\") / 365)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "852992f9-c823-4111-b3b1-a53aac27bffb",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "# Calculate age of account in years and categorize into `account_age_group`\n",
    "usersDF = usersDF.withColumn(\"account_age_years\", round(col(\"seniority\") / 365, 2))\n",
    "usersDF = usersDF.withColumn(\"account_age_group\",\n",
    "                             when(col(\"account_age_years\") < 1, \"New\")\n",
    "                             .when((col(\"account_age_years\") >= 1) & (col(\"account_age_years\") < 3), \"Intermediate\")\n",
    "                             .otherwise(\"Experienced\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "24b4f2f3-e293-47e7-9bd5-4947eaef7c7b",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "# Add a column with the current year for comparison\n",
    "usersDF = usersDF.withColumn(\"current_year\", year(current_date()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "27af8e97-0e06-4c67-a337-4c4271e7061f",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Creatively combining strings to form a unique user descriptor\n",
    "usersDF = usersDF.withColumn(\"user_descriptor\", \n",
    "                             concat(col(\"gender\"), lit(\"_\"), \n",
    "                                    col(\"countrycode\"), lit(\"_\"), \n",
    "                                    expr(\"substring(civilitytitle_clean, 1, 3)\"), lit(\"_\"), \n",
    "                                    col(\"language_full\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e0be4c49-ab2f-490d-b503-2054cd2ddf18",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "usersDF = usersDF.withColumn(\"flag_long_title\", length(col(\"civilitytitle\")) > 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5539c3f3-efdc-4770-9860-a4831b0b7fdc",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "\n",
    "usersDF = usersDF.withColumn(\"hasanyapp\", col(\"hasanyapp\").cast(\"boolean\"))\n",
    "usersDF = usersDF.withColumn(\"hasandroidapp\", col(\"hasandroidapp\").cast(\"boolean\"))\n",
    "usersDF = usersDF.withColumn(\"hasiosapp\", col(\"hasiosapp\").cast(\"boolean\"))\n",
    "usersDF = usersDF.withColumn(\"hasprofilepicture\", col(\"hasprofilepicture\").cast(\"boolean\"))\n",
    "\n",
    "\n",
    "usersDF = usersDF.withColumn(\"socialnbfollowers\", col(\"socialnbfollowers\").cast(IntegerType()))\n",
    "usersDF = usersDF.withColumn(\"socialnbfollows\", col(\"socialnbfollows\").cast(IntegerType()))\n",
    "\n",
    "usersDF = usersDF.withColumn(\"productspassrate\", col(\"productspassrate\").cast(DecimalType(10, 2)))\n",
    "usersDF = usersDF.withColumn(\"seniorityasmonths\", col(\"seniorityasmonths\").cast(DecimalType(10, 2)))\n",
    "usersDF = usersDF.withColumn(\"seniorityasyears\", col(\"seniorityasyears\").cast(DecimalType(10, 2)))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "5e89d048-878c-4df4-9dfa-2ab00d1ce6d5",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "usersDF = usersDF.withColumn(\"dayssincelastlogin\",\n",
    "                             when(col(\"dayssincelastlogin\").isNotNull(),\n",
    "                                  col(\"dayssincelastlogin\").cast(IntegerType()))\n",
    "                             .otherwise(0))  #"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "4fc54188-8cda-4293-aaf7-6ed8f36afed3",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "usersDF.write.format(\"delta\").mode(\"overwrite\").save(\"/mnt/delta/tables/silver/users\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "f6c6ecb8-8bec-4a20-aeb6-469d539cb444",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "#/mnt/delta/tables/bronze/users\n",
    "buyersDF = spark.read.format(\"delta\").load(\"/mnt/delta/tables/bronze/buyers\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "c4b95c1c-7521-40fe-bf13-8ce818591ccb",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Casting Integer columns\n",
    "integer_columns = [\n",
    "    'buyers', 'topbuyers', 'femalebuyers', 'malebuyers',\n",
    "    'topfemalebuyers', 'topmalebuyers', 'totalproductsbought',\n",
    "    'totalproductswished', 'totalproductsliked', 'toptotalproductsbought',\n",
    "    'toptotalproductswished', 'toptotalproductsliked'\n",
    "]\n",
    "\n",
    "for column_name in integer_columns:\n",
    "    buyersDF = buyersDF.withColumn(column_name, col(column_name).cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "3c825b01-cade-4fa7-945d-fe4f147d14dc",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "# Casting Decimal columns\n",
    "decimal_columns = [\n",
    "    'topbuyerratio', 'femalebuyersratio', 'topfemalebuyersratio',\n",
    "    'boughtperwishlistratio', 'boughtperlikeratio', 'topboughtperwishlistratio',\n",
    "    'topboughtperlikeratio', 'meanproductsbought', 'meanproductswished',\n",
    "    'meanproductsliked', 'topmeanproductsbought', 'topmeanproductswished',\n",
    "    'topmeanproductsliked', 'meanofflinedays', 'topmeanofflinedays',\n",
    "    'meanfollowers', 'meanfollowing', 'topmeanfollowers', 'topmeanfollowing'\n",
    "]\n",
    "\n",
    "for column_name in decimal_columns:\n",
    "    buyersDF = buyersDF.withColumn(column_name, col(column_name).cast(DecimalType(10, 2)))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9610202b-1675-4767-925d-02b3be625959",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "# Normalize country names\n",
    "buyersDF = buyersDF.withColumn(\"country\", initcap(col(\"country\")))\n",
    "\n",
    "for col_name in integer_columns:\n",
    "    buyersDF = buyersDF.fillna({col_name: 0})\n",
    "\n",
    "# Calculate the ratio of female to male buyers\n",
    "buyersDF = buyersDF.withColumn(\"female_to_male_ratio\", \n",
    "                               round(col(\"femalebuyers\") / (col(\"malebuyers\") + 1), 2))\n",
    "\n",
    "# Determine the market potential by comparing wishlist and purchases\n",
    "buyersDF = buyersDF.withColumn(\"wishlist_to_purchase_ratio\", \n",
    "                               round(col(\"totalproductswished\") / (col(\"totalproductsbought\") + 1), 2))\n",
    "\n",
    "# Tag countries with a high engagement ratio\n",
    "high_engagement_threshold = 0.5\n",
    "buyersDF = buyersDF.withColumn(\"high_engagement\",\n",
    "                               when(col(\"boughtperwishlistratio\") > high_engagement_threshold, True)\n",
    "                               .otherwise(False))\n",
    "                               \n",
    "    # Flag markets with increasing female buyer participation\n",
    "buyersDF = buyersDF.withColumn(\"growing_female_market\",\n",
    "                               when(col(\"femalebuyersratio\") > col(\"topfemalebuyersratio\"), True)\n",
    "                               .otherwise(False))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "befd510b-e190-4a4f-86bf-7cd613bc5de9",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "buyersDF.write.format(\"delta\").mode(\"overwrite\").save(\"/mnt/delta/tables/silver/buyers\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "bfc9950d-f2d2-47e2-854b-8bb1023899d7",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "sellersDF = spark.read.format(\"delta\").load(\"/mnt/delta/tables/bronze/sellers\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "9cb31aab-6157-492d-86d2-9d79404292f3",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "sellersDF = sellersDF \\\n",
    "    .withColumn(\"nbsellers\", col(\"nbsellers\").cast(IntegerType())) \\\n",
    "    .withColumn(\"meanproductssold\", col(\"meanproductssold\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanproductslisted\", col(\"meanproductslisted\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meansellerpassrate\", col(\"meansellerpassrate\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"totalproductssold\", col(\"totalproductssold\").cast(IntegerType())) \\\n",
    "    .withColumn(\"totalproductslisted\", col(\"totalproductslisted\").cast(IntegerType())) \\\n",
    "    .withColumn(\"meanproductsbought\", col(\"meanproductsbought\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanproductswished\", col(\"meanproductswished\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanproductsliked\", col(\"meanproductsliked\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"totalbought\", col(\"totalbought\").cast(IntegerType())) \\\n",
    "    .withColumn(\"totalwished\", col(\"totalwished\").cast(IntegerType())) \\\n",
    "    .withColumn(\"totalproductsliked\", col(\"totalproductsliked\").cast(IntegerType())) \\\n",
    "    .withColumn(\"meanfollowers\", col(\"meanfollowers\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanfollows\", col(\"meanfollows\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"percentofappusers\", col(\"percentofappusers\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"percentofiosusers\", col(\"percentofiosusers\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanseniority\", col(\"meanseniority\").cast(DecimalType(10, 2)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "1b7dae62-24cb-4921-b63f-4d2f621e0995",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "# Normalize country names and gender values\n",
    "sellersDF = sellersDF.withColumn(\"country\", initcap(col(\"country\"))) \\\n",
    "                                                .withColumn(\"sex\", upper(col(\"sex\")))\n",
    "\n",
    "\n",
    "#Add a column to categorize the number of sellers\n",
    "sellersDF = sellersDF.withColumn(\"seller_size_category\", \n",
    "                               when(col(\"nbsellers\") < 500, \"Small\") \\\n",
    "                               .when((col(\"nbsellers\") >= 500) & (col(\"nbsellers\") < 2000), \"Medium\") \\\n",
    "                               .otherwise(\"Large\"))\n",
    "\n",
    "# Calculate the mean products listed per seller as an indicator of seller activity\n",
    "sellersDF = sellersDF.withColumn(\"mean_products_listed_per_seller\", \n",
    "                               round(col(\"totalproductslisted\") / col(\"nbsellers\"), 2))\n",
    "\n",
    "# Identify markets with high seller pass rate\n",
    "sellersDF = sellersDF.withColumn(\"high_seller_pass_rate\", \n",
    "                               when(col(\"meansellerpassrate\") > 0.75, \"High\") \\\n",
    "                               .otherwise(\"Normal\"))\n",
    "\n",
    "mean_pass_rate = sellersDF.select(round(avg(\"meansellerpassrate\"), 2).alias(\"avg_pass_rate\")).collect()[0][\"avg_pass_rate\"]\n",
    "\n",
    "sellersDF = sellersDF.withColumn(\"meansellerpassrate\",\n",
    "                                 when(col(\"meansellerpassrate\").isNull(), mean_pass_rate)\n",
    "                                 .otherwise(col(\"meansellerpassrate\")))\n",
    "\n",
    "sellersDF.write.format(\"delta\").mode(\"overwrite\").save(\"/mnt/delta/tables/silver/sellers\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "b8e3de96-0e68-4d28-88fe-92a062313753",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "countriesDF = spark.read.format(\"delta\").load(\"/mnt/delta/tables/bronze/countries\")\n",
    "\n",
    "countriesDF = countriesDF \\\n",
    "    .withColumn(\"sellers\", col(\"sellers\").cast(IntegerType())) \\\n",
    "    .withColumn(\"topsellers\", col(\"topsellers\").cast(IntegerType())) \\\n",
    "    .withColumn(\"topsellerratio\", col(\"topsellerratio\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"femalesellersratio\", col(\"femalesellersratio\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"topfemalesellersratio\", col(\"topfemalesellersratio\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"femalesellers\", col(\"femalesellers\").cast(IntegerType())) \\\n",
    "    .withColumn(\"malesellers\", col(\"malesellers\").cast(IntegerType())) \\\n",
    "    .withColumn(\"topfemalesellers\", col(\"topfemalesellers\").cast(IntegerType())) \\\n",
    "    .withColumn(\"topmalesellers\", col(\"topmalesellers\").cast(IntegerType())) \\\n",
    "    .withColumn(\"countrysoldratio\", col(\"countrysoldratio\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"bestsoldratio\", col(\"bestsoldratio\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"toptotalproductssold\", col(\"toptotalproductssold\").cast(IntegerType())) \\\n",
    "    .withColumn(\"totalproductssold\", col(\"totalproductssold\").cast(IntegerType())) \\\n",
    "    .withColumn(\"toptotalproductslisted\", col(\"toptotalproductslisted\").cast(IntegerType())) \\\n",
    "    .withColumn(\"totalproductslisted\", col(\"totalproductslisted\").cast(IntegerType())) \\\n",
    "    .withColumn(\"topmeanproductssold\", col(\"topmeanproductssold\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"topmeanproductslisted\", col(\"topmeanproductslisted\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanproductssold\", col(\"meanproductssold\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanproductslisted\", col(\"meanproductslisted\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanofflinedays\", col(\"meanofflinedays\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"topmeanofflinedays\", col(\"topmeanofflinedays\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanfollowers\", col(\"meanfollowers\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"meanfollowing\", col(\"meanfollowing\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"topmeanfollowers\", col(\"topmeanfollowers\").cast(DecimalType(10, 2))) \\\n",
    "    .withColumn(\"topmeanfollowing\", col(\"topmeanfollowing\").cast(DecimalType(10, 2)))\n",
    "\n",
    "countriesDF = countriesDF.withColumn(\"country\", initcap(col(\"country\")))\n",
    "\n",
    "\n",
    "# Calculating the ratio of top sellers to total sellers\n",
    "countriesDF = countriesDF.withColumn(\"top_seller_ratio\", \n",
    "                                        round(col(\"topsellers\") / col(\"sellers\"), 2))\n",
    "\n",
    "# countriesDF countries with a high ratio of female sellers\n",
    "countriesDF = countriesDF.withColumn(\"high_female_seller_ratio\", \n",
    "                                        when(col(\"femalesellersratio\") > 0.5, True).otherwise(False))\n",
    "\n",
    "# Adding a performance indicator based on the sold/listed ratio\n",
    "countriesDF = countriesDF.withColumn(\"performance_indicator\", \n",
    "                                        round(col(\"toptotalproductssold\") / (col(\"toptotalproductslisted\") + 1), 2))\n",
    "\n",
    "# Flag countries with exceptionally high performance\n",
    "performance_threshold = 0.8\n",
    "countriesDF = countriesDF.withColumn(\"high_performance\", \n",
    "                                        when(col(\"performance_indicator\") > performance_threshold, True).otherwise(False))\n",
    "\n",
    "countriesDF = countriesDF.withColumn(\"activity_level\",\n",
    "                                       when(col(\"meanofflinedays\") < 30, \"Highly Active\")\n",
    "                                       .when((col(\"meanofflinedays\") >= 30) & (col(\"meanofflinedays\") < 60), \"Moderately Active\")\n",
    "                                       .otherwise(\"Low Activity\"))\n",
    "\n",
    "\n",
    "countriesDF.write.format(\"delta\").mode(\"overwrite\").save(\"/mnt/delta/tables/silver/countries\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "1d183322-9dae-4aff-9a64-1703ab241b38",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "Silver_Layer",
   "widgets": {}
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}