spark/5_1_dynamic_partition_pruning.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dynamic Partitioning Pruning\n",
    "- Pruning partitions at runtime\n",
    "- Problem Statement: Analyse the listening activity of users on the release date of a song on/after `2020-01-01`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import warnings\n",
    "warnings.filterwarnings(\"ignore\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.interactiveshell import InteractiveShell\n",
    "InteractiveShell.ast_node_interactivity = 'all'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {
      "byteLimit": 2048000,
      "rowLimit": 10000
     },
     "inputWidgets": {},
     "nuid": "e02171b5-8b8d-4b81-8826-2fb3ee1a8a24",
     "showTitle": false,
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.storagelevel import StorageLevel\n",
    "from pyspark.sql.types import *\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "23/11/13 18:06:48 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    }
   ],
   "source": [
    "spark = (\n",
    "    SparkSession\n",
    "    .builder\n",
    "    .config(\"spark.driver.memory\", \"10g\")\n",
    "    .master(\"local[*]\")\n",
    "    .appName(\"6_1_dynamic_partition_pruning\")\n",
    "    .getOrCreate()\n",
    ")\n",
    "sc = spark.sparkContext\n",
    "sc.setLogLevel(\"ERROR\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_listening_actv = spark.read.csv(\"../data/partitioning/raw/Spotify_Listening_Activity.csv\", header=True, inferSchema=True)\n",
    "df_listening_actv = (\n",
    "    df_listening_actv\n",
    "    .withColumnRenamed(\"listen_date\", \"listen_time\")\n",
    "    .withColumn(\"listen_date\", F.to_date(\"listen_time\", \"yyyy-MM-dd HH:mm:ss.SSSSSS\"))\n",
    ")\n",
    "\n",
    "# Partitioning listening activity by the listen date\n",
    "(\n",
    "    df_listening_actv\n",
    "    .write\n",
    "    .partitionBy(\"listen_date\")\n",
    "    .mode(\"overwrite\")\n",
    "    .parquet(\"../data/partitioning/partitioned/listening_activity_pt\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------+--------------------------+---------------+-----------+\n",
      "|activity_id|song_id|listen_time               |listen_duration|listen_date|\n",
      "+-----------+-------+--------------------------+---------------+-----------+\n",
      "|4456       |16     |2023-07-18 10:15:47.023264|151            |2023-07-18 |\n",
      "|4457       |65     |2023-07-18 10:15:47.023264|181            |2023-07-18 |\n",
      "|4458       |60     |2023-07-18 10:15:47.023264|280            |2023-07-18 |\n",
      "|4459       |3      |2023-07-18 10:15:47.023264|249            |2023-07-18 |\n",
      "|4460       |45     |2023-07-18 10:15:47.023264|130            |2023-07-18 |\n",
      "+-----------+-------+--------------------------+---------------+-----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_listening_actv_pt = spark.read.parquet(\"../data/partitioning/partitioned/listening_activity_pt\")\n",
    "df_listening_actv_pt.show(5, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- song_id: integer (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- artist_id: integer (nullable = true)\n",
      " |-- release_date: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_songs = spark.read.csv(\"../data/partitioning/raw/Spotify_Songs.csv\", header=True, inferSchema=True)\n",
    "df_songs.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+---------+--------------------------+------------+\n",
      "|song_id|title |artist_id|release_datetime          |release_date|\n",
      "+-------+------+---------+--------------------------+------------+\n",
      "|1      |Song_1|2        |2021-10-15 10:15:47.006571|2021-10-15  |\n",
      "|2      |Song_2|45       |2020-12-07 10:15:47.006588|2020-12-07  |\n",
      "|3      |Song_3|25       |2022-07-11 10:15:47.006591|2022-07-11  |\n",
      "|4      |Song_4|25       |2019-03-09 10:15:47.006593|2019-03-09  |\n",
      "|5      |Song_5|26       |2019-09-07 10:15:47.006596|2019-09-07  |\n",
      "+-------+------+---------+--------------------------+------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "root\n",
      " |-- song_id: integer (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- artist_id: integer (nullable = true)\n",
      " |-- release_datetime: string (nullable = true)\n",
      " |-- release_date: date (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_songs = (\n",
    "    df_songs\n",
    "    .withColumnRenamed(\"release_date\", \"release_datetime\")\n",
    "    .withColumn(\"release_date\", F.to_date(\"release_datetime\", \"yyyy-MM-dd HH:mm:ss.SSSSSS\"))\n",
    ")\n",
    "df_songs.show(5, False)\n",
    "df_songs.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pick songs released in 2020\n",
    "df_selected_songs = df_songs.filter(F.col(\"release_date\") > F.lit(\"2019-12-31\"))\n",
    "\n",
    "\n",
    "df_listening_actv_of_selected_songs = df_listening_actv_pt.join(\n",
    "    df_selected_songs, \n",
    "    on=(df_songs.release_date == df_listening_actv_pt.listen_date) & (df_songs.song_id == df_listening_actv_pt.song_id), \n",
    "    how=\"inner\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Parsed Logical Plan ==\n",
      "Join Inner, ((release_date#104 = listen_date#48) AND (song_id#91 = song_id#45))\n",
      ":- Relation [activity_id#44,song_id#45,listen_time#46,listen_duration#47,listen_date#48] parquet\n",
      "+- Filter (release_date#104 > cast(2019-12-31 as date))\n",
      "   +- Project [song_id#91, title#92, artist_id#93, release_datetime#99, to_date('release_datetime, Some(yyyy-MM-dd HH:mm:ss.SSSSSS)) AS release_date#104]\n",
      "      +- Project [song_id#91, title#92, artist_id#93, release_date#94 AS release_datetime#99]\n",
      "         +- Relation [song_id#91,title#92,artist_id#93,release_date#94] csv\n",
      "\n",
      "== Analyzed Logical Plan ==\n",
      "activity_id: int, song_id: int, listen_time: string, listen_duration: int, listen_date: date, song_id: int, title: string, artist_id: int, release_datetime: string, release_date: date\n",
      "Join Inner, ((release_date#104 = listen_date#48) AND (song_id#91 = song_id#45))\n",
      ":- Relation [activity_id#44,song_id#45,listen_time#46,listen_duration#47,listen_date#48] parquet\n",
      "+- Filter (release_date#104 > cast(2019-12-31 as date))\n",
      "   +- Project [song_id#91, title#92, artist_id#93, release_datetime#99, to_date('release_datetime, Some(yyyy-MM-dd HH:mm:ss.SSSSSS)) AS release_date#104]\n",
      "      +- Project [song_id#91, title#92, artist_id#93, release_date#94 AS release_datetime#99]\n",
      "         +- Relation [song_id#91,title#92,artist_id#93,release_date#94] csv\n",
      "\n",
      "== Optimized Logical Plan ==\n",
      "Join Inner, ((release_date#104 = listen_date#48) AND (song_id#91 = song_id#45))\n",
      ":- Filter ((((listen_date#48 > 2019-12-31) AND isnotnull(listen_date#48)) AND isnotnull(song_id#45)) AND dynamicpruning#155 [listen_date#48])\n",
      ":  :  +- Project [song_id#91, title#92, artist_id#93, release_date#94 AS release_datetime#99, cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) AS release_date#104]\n",
      ":  :     +- Filter ((cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) > 2019-12-31) AND isnotnull(song_id#91))\n",
      ":  :        +- Relation [song_id#91,title#92,artist_id#93,release_date#94] csv\n",
      ":  +- Relation [activity_id#44,song_id#45,listen_time#46,listen_duration#47,listen_date#48] parquet\n",
      "+- Project [song_id#91, title#92, artist_id#93, release_date#94 AS release_datetime#99, cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) AS release_date#104]\n",
      "   +- Filter ((cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) > 2019-12-31) AND isnotnull(song_id#91))\n",
      "      +- Relation [song_id#91,title#92,artist_id#93,release_date#94] csv\n",
      "\n",
      "== Physical Plan ==\n",
      "AdaptiveSparkPlan isFinalPlan=false\n",
      "+- BroadcastHashJoin [listen_date#48, song_id#45], [release_date#104, song_id#91], Inner, BuildRight, false\n",
      "   :- Filter isnotnull(song_id#45)\n",
      "   :  +- FileScan parquet [activity_id#44,song_id#45,listen_time#46,listen_duration#47,listen_date#48] Batched: true, DataFilters: [isnotnull(song_id#45)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/Users/afaqueahmad/Documents/YouTube/spark-experiments/data/parti..., PartitionFilters: [(listen_date#48 > 2019-12-31), isnotnull(listen_date#48), dynamicpruningexpression(listen_date#4..., PushedFilters: [IsNotNull(song_id)], ReadSchema: struct<activity_id:int,song_id:int,listen_time:string,listen_duration:int>\n",
      "   :        +- SubqueryAdaptiveBroadcast dynamicpruning#155, 0, true, Project [song_id#91, title#92, artist_id#93, release_date#94 AS release_datetime#99, cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) AS release_date#104], [release_date#104, song_id#91]\n",
      "   :           +- AdaptiveSparkPlan isFinalPlan=false\n",
      "   :              +- Project [song_id#91, title#92, artist_id#93, release_date#94 AS release_datetime#99, cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) AS release_date#104]\n",
      "   :                 +- Filter ((cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) > 2019-12-31) AND isnotnull(song_id#91))\n",
      "   :                    +- FileScan csv [song_id#91,title#92,artist_id#93,release_date#94] Batched: false, DataFilters: [(cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata..., Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/Users/afaqueahmad/Documents/YouTube/spark-experiments/data/parti..., PartitionFilters: [], PushedFilters: [IsNotNull(song_id)], ReadSchema: struct<song_id:int,title:string,artist_id:int,release_date:string>\n",
      "   +- BroadcastExchange HashedRelationBroadcastMode(List(input[4, date, true], input[0, int, true]),false), [id=#112]\n",
      "      +- Project [song_id#91, title#92, artist_id#93, release_date#94 AS release_datetime#99, cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) AS release_date#104]\n",
      "         +- Filter ((cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata), false) as date) > 2019-12-31) AND isnotnull(song_id#91))\n",
      "            +- FileScan csv [song_id#91,title#92,artist_id#93,release_date#94] Batched: false, DataFilters: [(cast(gettimestamp(release_date#94, yyyy-MM-dd HH:mm:ss.SSSSSS, TimestampType, Some(Asia/Kolkata..., Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/Users/afaqueahmad/Documents/YouTube/spark-experiments/data/parti..., PartitionFilters: [], PushedFilters: [IsNotNull(song_id)], ReadSchema: struct<song_id:int,title:string,artist_id:int,release_date:string>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_listening_actv_of_selected_songs.explain(True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------+--------------------+---------------+-----------+-------+-------+---------+--------------------+------------+\n",
      "|activity_id|song_id|         listen_time|listen_duration|listen_date|song_id|  title|artist_id|    release_datetime|release_date|\n",
      "+-----------+-------+--------------------+---------------+-----------+-------+-------+---------+--------------------+------------+\n",
      "|       9760|     89|2023-07-24 10:15:...|             81| 2023-07-24|     89|Song_89|       33|2023-07-24 10:15:...|  2023-07-24|\n",
      "|       9768|     89|2023-07-24 10:15:...|            295| 2023-07-24|     89|Song_89|       33|2023-07-24 10:15:...|  2023-07-24|\n",
      "|       9799|     89|2023-07-24 10:15:...|            272| 2023-07-24|     89|Song_89|       33|2023-07-24 10:15:...|  2023-07-24|\n",
      "|       7322|     64|2023-10-25 10:15:...|             95| 2023-10-25|     64|Song_64|       32|2023-10-25 10:15:...|  2023-10-25|\n",
      "+-----------+-------+--------------------+---------------+-----------+-------+-------+---------+--------------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_listening_actv_of_selected_songs.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "dashboards": [],
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "data-skew-explanation",
   "notebookOrigID": 4018749166498458,
   "widgets": {}
  },
  "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.8.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}