examples/07_filters.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8233fbaa",
   "metadata": {},
   "source": [
    "# Filtering log data with RePlay\n",
    "\n",
    "This notebook presents the RePlay functionality for interactions data filtering.<br>\n",
    "You can conveniently filter your datasets, for example, leave only positive interactions, select the desired period of time or number of interactions, filter out users and items with low number of interactions.\n",
    "<br>RePlay offers filtering functions:\n",
    "\n",
    "* MinCountFilter - filter out users or items with a few interactions\n",
    "* LowRatingFilter - filter out low ratings/relevance values\n",
    "* NumInteractionsFilter - select required number of the first/last user's interactions\n",
    "* GlobalDaysFilter - select required number of days from interactions\n",
    "* EntityDaysFilter - select required number of days from each entity interactions\n",
    "* TimePeriodFilter - select interactions from required time period\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b5d09b9d",
   "metadata": {},
   "outputs": [],
   "source": [
    "! pip install rs-datasets -q"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "d6a098cd",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style type='text/css'>\n",
       ".datatable table.frame { margin-bottom: 0; }\n",
       ".datatable table.frame thead { border-bottom: none; }\n",
       ".datatable table.frame tr.coltypes td {  color: #FFFFFF;  line-height: 6px;  padding: 0 0.5em;}\n",
       ".datatable .bool    { background: #DDDD99; }\n",
       ".datatable .object  { background: #565656; }\n",
       ".datatable .int     { background: #5D9E5D; }\n",
       ".datatable .float   { background: #4040CC; }\n",
       ".datatable .str     { background: #CC4040; }\n",
       ".datatable .time    { background: #40CC40; }\n",
       ".datatable .row_index {  background: var(--jp-border-color3);  border-right: 1px solid var(--jp-border-color0);  color: var(--jp-ui-font-color3);  font-size: 9px;}\n",
       ".datatable .frame tbody td { text-align: left; }\n",
       ".datatable .frame tr.coltypes .row_index {  background: var(--jp-border-color0);}\n",
       ".datatable th:nth-child(2) { padding-left: 12px; }\n",
       ".datatable .hellipsis {  color: var(--jp-cell-editor-border-color);}\n",
       ".datatable .vellipsis {  background: var(--jp-layout-color0);  color: var(--jp-cell-editor-border-color);}\n",
       ".datatable .na {  color: var(--jp-cell-editor-border-color);  font-size: 80%;}\n",
       ".datatable .sp {  opacity: 0.25;}\n",
       ".datatable .footer { font-size: 9px; }\n",
       ".datatable .frame_dimensions {  background: var(--jp-border-color3);  border-top: 1px solid var(--jp-border-color0);  color: var(--jp-ui-font-color3);  display: inline-block;  opacity: 0.6;  padding: 1px 10px 1px 5px;}\n",
       "</style>\n"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import pandas as pd\n",
    "from rs_datasets import MovieLens\n",
    "\n",
    "from pyspark.sql import functions as sf\n",
    "from pyspark.sql.types import TimestampType\n",
    "\n",
    "from replay.utils.session_handler import State\n",
    "from replay.utils.spark_utils import get_log_info, convert2spark, process_timestamp_column\n",
    "from replay.preprocessing.filters import (\n",
    "    MinCountFilter,\n",
    "    LowRatingFilter,\n",
    "    NumInteractionsFilter,\n",
    "    EntityDaysFilter,\n",
    "    GlobalDaysFilter,\n",
    "    TimePeriodFilter,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d6b251d3",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = State().session\n",
    "spark.sparkContext.setLogLevel('ERROR')\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e82a2b1e",
   "metadata": {},
   "source": [
    "## Get started\n",
    "\n",
    "Download the dataset **MovieLens**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "e327edc5",
   "metadata": {},
   "outputs": [],
   "source": [
    "data = MovieLens(\"1m\")\n",
    "log = data.ratings"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "12f451b8",
   "metadata": {},
   "source": [
    "Use the `convert2spark` function to convert pandas.DataFrame to spark.DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "b88a3190",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_spark = convert2spark(log)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dcd4fd5b",
   "metadata": {},
   "source": [
    "Convert timestamp to appropriate format"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "a1cb14a3",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_spark = process_timestamp_column(log_spark, \"timestamp\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "09d267a8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+------+-------------------+\n",
      "|user_id|item_id|rating|          timestamp|\n",
      "+-------+-------+------+-------------------+\n",
      "|      1|   1193|     5|2001-01-01 01:12:40|\n",
      "|      1|    661|     3|2001-01-01 01:35:09|\n",
      "|      1|    914|     3|2001-01-01 01:32:48|\n",
      "|      1|   3408|     4|2001-01-01 01:04:35|\n",
      "+-------+-------+------+-------------------+\n",
      "only showing top 4 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "log_spark.show(4)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c679d9b2",
   "metadata": {},
   "source": [
    "## MinCountFilter\n",
    "\n",
    "Filter out users or items with number of interactions less than given threshold.\n",
    "\n",
    "Parameters:\n",
    "\n",
    "* `num_entries` - threshold for number of interactions\n",
    "    \n",
    "* `groupby_column` - column relative to which filtering occurs\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ee48c81a",
   "metadata": {},
   "source": [
    "The picture shows how filtering works using the example of a simple interaction log.<br>\n",
    "Colored cells represent rating for given user and item. The color depends on the rating."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1254e3b6",
   "metadata": {},
   "source": [
    "![title](img/filter_by_min_count.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "56a3488f",
   "metadata": {},
   "source": [
    "Let's leave users who have `>= 50` interactions with items"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "04c5b172",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_filter = MinCountFilter(num_entries=50, groupby_column=\"user_id\").transform(log_spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "482888fe",
   "metadata": {},
   "source": [
    "Check minimum number of interactions per user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "32a4a8ef",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------+\n",
      "|Minimum number of interactions per user's|\n",
      "+-----------------------------------------+\n",
      "|                                       50|\n",
      "+-----------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "log_filter.groupBy(\"user_id\").count().select(sf.min(\"count\").alias(\"Minimum number of interactions per user's\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "82fbc1e6",
   "metadata": {},
   "source": [
    "Number of users has decreased"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "f8f06e12",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Before filtering\n",
      " total lines: 1000209, total users: 6040, total items: 3706\n",
      "After filtering\n",
      " total lines: 943471, total users: 4297, total items: 3689\n"
     ]
    }
   ],
   "source": [
    "print(\"Before filtering\\n\", get_log_info(log_spark, user_col=\"user_id\", item_col=\"item_id\"))\n",
    "print(\"After filtering\\n\", get_log_info(log_filter, user_col=\"user_id\", item_col=\"item_id\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "034b2b23",
   "metadata": {},
   "source": [
    "Let's leave items that `>= 100` users interacted with"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "62f47cb3",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_filter = MinCountFilter(num_entries=100, groupby_column=\"item_id\").transform(log_spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d634e27e",
   "metadata": {},
   "source": [
    "Check minimum number of interactions per item"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "e61f9eb6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------+\n",
      "|Minimum number of interactions per item's|\n",
      "+-----------------------------------------+\n",
      "|                                      100|\n",
      "+-----------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "log_filter.groupBy(\"item_id\").count().select(sf.min(\"count\").alias(\"Minimum number of interactions per item's\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4c2eed97",
   "metadata": {},
   "source": [
    "Number of items has decreased"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "b5ce4e2c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Before filtering\n",
      " total lines: 1000209, total users: 6040, total items: 3706\n",
      "After filtering\n",
      " total lines: 942225, total users: 6040, total items: 2019\n"
     ]
    }
   ],
   "source": [
    "print(\"Before filtering\\n\", get_log_info(log_spark, user_col=\"user_id\", item_col=\"item_id\"))\n",
    "print(\"After filtering\\n\", get_log_info(log_filter, user_col=\"user_id\", item_col=\"item_id\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "20e93391",
   "metadata": {},
   "source": [
    "## LowRatingFilter\n",
    "\n",
    "Filter out interactions with ratings/relevance less than given threshold.\n",
    "\n",
    "Parameters:\n",
    "\n",
    "* `value` - rating/relevance threshold\n",
    "\n",
    "* `rating_column` - rating/relevance column\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "91a8a676",
   "metadata": {},
   "source": [
    "![title](img/filter_out_low_ratings.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1d0782f0",
   "metadata": {},
   "source": [
    "Let's leave interactions with rating `>= 4`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "3f0e03a5",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_filter = LowRatingFilter(value=4, rating_column=\"rating\").transform(log_spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1cac27ef",
   "metadata": {},
   "source": [
    "Check minimum and maximum rating"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "edb2d13f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|Minimum rating|\n",
      "+--------------+\n",
      "|             4|\n",
      "+--------------+\n",
      "\n",
      "+--------------+\n",
      "|Maximum rating|\n",
      "+--------------+\n",
      "|             5|\n",
      "+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "log_filter.select(sf.min(\"rating\").alias(\"Minimum rating\")).show()\n",
    "log_filter.select(sf.max(\"rating\").alias(\"Maximum rating\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5da6c527",
   "metadata": {},
   "source": [
    "The number of users and interactions has decreased"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "26e52544",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Before filtering\n",
      " total lines: 1000209, total users: 6040, total items: 3706\n",
      "After filtering\n",
      " total lines: 575281, total users: 6038, total items: 3533\n"
     ]
    }
   ],
   "source": [
    "print(\"Before filtering\\n\", get_log_info(log_spark, user_col=\"user_id\", item_col=\"item_id\"))\n",
    "print(\"After filtering\\n\", get_log_info(log_filter, user_col=\"user_id\", item_col=\"item_id\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "180c13fd",
   "metadata": {},
   "source": [
    "## NumInteractionsFilter\n",
    "\n",
    "Select required number of the first/last user's interactions.\n",
    "\n",
    "Parameters:\n",
    "\n",
    "    \n",
    "* `num_interactions` - threshold value for the number of interactions for each user\n",
    "    \n",
    "* `first` - indicator, if the value is `True`, take first `num_interactions` for each user, if the value is `False`, take last `num_interactions` for each user\n",
    "    \n",
    "* `timestamp_column` - date column name\n",
    "    \n",
    "* `query_column` - user id column name\n",
    "    \n",
    "* `item_column` - item id column name\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9cd4eec3",
   "metadata": {},
   "source": [
    "![title](img/take_num_user_interactions.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1b98f154",
   "metadata": {},
   "source": [
    "Let's leave the last `2` interactions for each user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "5ba9ee15",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_filter = NumInteractionsFilter(\n",
    "    num_interactions=2,\n",
    "    first=False,\n",
    "    query_column=\"user_id\",\n",
    "    item_column=\"item_id\",\n",
    ").transform(log_spark)\n",
    "                                        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "616588ed",
   "metadata": {},
   "source": [
    "Check maximum number of interactions per user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "9e25783d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------+\n",
      "|Maximum number of interactions per user's|\n",
      "+-----------------------------------------+\n",
      "|                                        2|\n",
      "+-----------------------------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "log_filter.groupBy(\"user_id\").count().select(sf.max(\"count\").alias(\"Maximum number of interactions per user's\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e913abd8",
   "metadata": {},
   "source": [
    "Note that the number of users has not changed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "187732f0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Before filtering\n",
      " total lines: 1000209, total users: 6040, total items: 3706\n",
      "After filtering\n",
      " total lines: 12080, total users: 6040, total items: 2430\n"
     ]
    }
   ],
   "source": [
    "print(\"Before filtering\\n\", get_log_info(log_spark, user_col=\"user_id\", item_col=\"item_id\"))\n",
    "print(\"After filtering\\n\", get_log_info(log_filter, user_col=\"user_id\", item_col=\"item_id\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e5b9ca40",
   "metadata": {},
   "source": [
    "## EntityDaysFilter\n",
    "\n",
    "Select required number of days from each user's or item's interactions.\n",
    "\n",
    "Parameters:\n",
    "   \n",
    "* `days` - time interval in days\n",
    "    \n",
    "* `first` - indicator, if the value is `True`, take first days of each user's interactions, if the value is `False`, take last days of each user's interactions\n",
    "    \n",
    "* `timestamp_column` - date column name\n",
    "    \n",
    "* `entity_column` - user/item id column name"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ce4d1a12",
   "metadata": {},
   "source": [
    "![title](img/take_num_days_of_user_hist.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d26bab04",
   "metadata": {},
   "source": [
    "Let's leave the last `week` of each user's interactions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "f01e2571",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_filter = EntityDaysFilter(\n",
    "    days=7,\n",
    "    first=False,\n",
    "    entity_column=\"user_id\",\n",
    "    timestamp_column=\"timestamp\",\n",
    ").transform(log_spark)\n",
    "                                        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "059fd0e8",
   "metadata": {},
   "source": [
    "Check maximum length of user history"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "7b06d4af",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------------+\n",
      "|Maximum difference in days per user's|\n",
      "+-------------------------------------+\n",
      "|                                    7|\n",
      "+-------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    log_filter\n",
    "    .groupBy(\"user_id\")\n",
    "    .agg(sf.datediff(sf.max(\"timestamp\"), sf.min(\"timestamp\")).alias(\"diff\"))\n",
    "    .select(sf.max(\"diff\").alias(\"Maximum difference in days per user's\"))\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a6a70e07",
   "metadata": {},
   "source": [
    "The number of users has not changed, but the number of interactions has decreased"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "4619db6f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Before filtering\n",
      " total lines: 1000209, total users: 6040, total items: 3706\n",
      "After filtering\n",
      " total lines: 551897, total users: 6040, total items: 3599\n"
     ]
    }
   ],
   "source": [
    "print(\"Before filtering\\n\", get_log_info(log_spark, user_col=\"user_id\", item_col=\"item_id\"))\n",
    "print(\"After filtering\\n\", get_log_info(log_filter, user_col=\"user_id\", item_col=\"item_id\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b0115419",
   "metadata": {},
   "source": [
    "## GlobalDaysFilter\n",
    "\n",
    "Select required number of days from interaction log.\n",
    "\n",
    "Parameters:\n",
    "\n",
    "    \n",
    "* `days` - time interval in days\n",
    "    \n",
    "* `first` - indicator, if the value is `True`, take first days from log, if the value is `False`, take last days from log\n",
    "    \n",
    "* `timestamp_column` - date column name"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "09c9caf8",
   "metadata": {},
   "source": [
    "![title](img/take_num_days_of_global_hist.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a50387dc",
   "metadata": {},
   "source": [
    "Let's leave the last `3 weeks` of interactions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "64021d7d",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_filter = GlobalDaysFilter(days=21, first=False).transform(log_spark)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "5352aac1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------------------------+\n",
      "|Maximum difference in days among all interactions|\n",
      "+-------------------------------------------------+\n",
      "|                                               21|\n",
      "+-------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    log_filter\n",
    "    .agg(sf.datediff(sf.max(\"timestamp\"), sf.min(\"timestamp\")).alias(\"diff\"))\n",
    "    .select(sf.max(\"diff\").alias(\"Maximum difference in days among all interactions\"))\n",
    ").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "0596608d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|Maximum date in log|\n",
      "+-------------------+\n",
      "|2003-02-28 20:49:50|\n",
      "+-------------------+\n",
      "\n",
      "+----------------------------+\n",
      "|Maximum date in filtered log|\n",
      "+----------------------------+\n",
      "|         2003-02-28 20:49:50|\n",
      "+----------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    log_spark\n",
    "    .select(\n",
    "        sf.max(\"timestamp\")\n",
    "        .alias(\"Maximum date in log\")\n",
    "    )\n",
    ").show()\n",
    "\n",
    "(\n",
    "    log_filter\n",
    "    .select(\n",
    "        sf.max(\"timestamp\")\n",
    "        .alias(\"Maximum date in filtered log\")\n",
    "    )\n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "117154d6",
   "metadata": {},
   "source": [
    "After such filtering, the number of interactions and users significantly decreased"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "e74ada59",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Before filtering\n",
      " total lines: 1000209, total users: 6040, total items: 3706\n",
      "After filtering\n",
      " total lines: 920, total users: 82, total items: 704\n"
     ]
    }
   ],
   "source": [
    "print(\"Before filtering\\n\", get_log_info(log_spark, user_col=\"user_id\", item_col=\"item_id\"))\n",
    "print(\"After filtering\\n\", get_log_info(log_filter, user_col=\"user_id\", item_col=\"item_id\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "691fd243",
   "metadata": {},
   "source": [
    "## TimePeriodFilter\n",
    "\n",
    "Select interactions from required time period.\n",
    "\n",
    "Parameters:\n",
    "\n",
    "    \n",
    "* `start_date` - beginning of required period\n",
    "    \n",
    "* `end_date` - end of required period\n",
    "    \n",
    "* `timestamp_column` - date column name"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "611d0fe8",
   "metadata": {},
   "source": [
    "![title](img/take_time_period.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9ec1af1f",
   "metadata": {},
   "source": [
    "Let's leave the data for `March` only"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "54a9fb0b",
   "metadata": {},
   "outputs": [],
   "source": [
    "log_filter = TimePeriodFilter(start_date=\"2001-03-01\", end_date=\"2001-04-01\").transform(log_spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d0af5a43",
   "metadata": {},
   "source": [
    "Check maximum and minimum date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "3be0c2e0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------------------+\n",
      "|       Minimum date|       Maximum date|\n",
      "+-------------------+-------------------+\n",
      "|2001-03-01 03:08:52|2001-03-31 23:12:22|\n",
      "+-------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(\n",
    "    log_filter\n",
    "    .select(\n",
    "        sf.min(\"timestamp\").alias(\"Minimum date\"),\n",
    "        sf.max(\"timestamp\").alias(\"Maximum date\")\n",
    "    )                            \n",
    ").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "11c55dbe",
   "metadata": {},
   "source": [
    "The number of records has noticeably decreased"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "f5824b04",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Before filtering\n",
      " total lines: 1000209, total users: 6040, total items: 3706\n",
      "After filtering\n",
      " total lines: 5986, total users: 321, total items: 2010\n"
     ]
    }
   ],
   "source": [
    "print(\"Before filtering\\n\", get_log_info(log_spark, user_col=\"user_id\", item_col=\"item_id\"))\n",
    "print(\"After filtering\\n\", get_log_info(log_filter, user_col=\"user_id\", item_col=\"item_id\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "162ca29c",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.9.12 ('replay-rec-00fCXNSq-py3.9')",
   "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.12"
  },
  "vscode": {
   "interpreter": {
    "hash": "40679255dc34d238a22556a84da5e66fac803417c68574ec8b323d2329ab9a09"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}