sga-task.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# User routes on the site\n",
    "## Description\n",
    "**Clickstream** is a sequence of user actions on a website. It allows you to understand how users interact with the site. In this task, you need to find the most frequent custom routes.\n",
    "\n",
    "## Input data\n",
    "Input data is а table with clickstream data in file `hdfs:/data/clickstream.csv`.\n",
    "\n",
    "### Table structure\n",
    "* `user_id (int)` - Unique user identifier.\n",
    "* `session_id (int)` - Unique identifier for the user session. The user's session lasts until the identifier changes.\n",
    "* `event_type (string)` - Event type from the list:\n",
    "    * **page** - visit to the page\n",
    "    * **event** - any action on the page\n",
    "    * <b>&lt;custom&gt;</b> - string with any other type\n",
    "* `event_type (string)` - Page on the site.\n",
    "* `timestamp (int)` - Unix-timestamp of action.\n",
    "\n",
    "### Browser errors\n",
    "Errors can sometimes occur in the user's browser - after such an error appears, we can no longer trust the data of this session and all the following lines after the error or at the same time with it are considered corrupted and **should not be counted** in statistics.\n",
    "\n",
    "When an error occurs on the page, a random string containing the word **error** will be written to the `event_type` field.\n",
    "\n",
    "### Sample of user session\n",
    "<pre>\n",
    "+-------+----------+------------+----------+----------+\n",
    "|user_id|session_id|  event_type|event_page| timestamp|\n",
    "+-------+----------+------------+----------+----------+\n",
    "|    562|       507|        page|      main|1620494781|\n",
    "|    562|       507|       event|      main|1620494788|\n",
    "|    562|       507|       event|      main|1620494798|\n",
    "|    562|       507|        page|    family|1620494820|\n",
    "|    562|       507|       event|    family|1620494828|\n",
    "|    562|       507|        page|      main|1620494848|\n",
    "|    562|       507|wNaxLlerrorU|      main|1620494865|\n",
    "|    562|       507|       event|      main|1620494873|\n",
    "|    562|       507|        page|      news|1620494875|\n",
    "|    562|       507|        page|   tariffs|1620494876|\n",
    "|    562|       507|       event|   tariffs|1620494884|\n",
    "|    562|       514|        page|      main|1620728918|\n",
    "|    562|       514|       event|      main|1620729174|\n",
    "|    562|       514|        page|   archive|1620729674|\n",
    "|    562|       514|        page|     bonus|1620729797|\n",
    "|    562|       514|        page|   tariffs|1620731090|\n",
    "|    562|       514|       event|   tariffs|1620731187|\n",
    "+-------+----------+------------+----------+----------+\n",
    "</pre>\n",
    "\n",
    "#### Correct user routes for a given user:\n",
    "* **Session 507**: main-family-main\n",
    "* **Session 514**: main-archive-bonus-tariffs\n",
    "\n",
    "Route elements are ordered by the time they appear in the clickstream, from earliest to latest.\n",
    "\n",
    "The route must be accounted for completely before the end of the session or an error in the session.\n",
    "\n",
    "## Task\n",
    "You need to use the Spark SQL, Spark RDD and Spark DF interfaces to create a solution file, the lines of which contain **the 30 most frequent user routes** on the site.\n",
    "\n",
    "Each line of the file should contain the `route` and `count` values **separated by tabs**, where:\n",
    "* `route` - route on the site, consisting of pages separated by \"-\".\n",
    "* `count` - the number of user sessions in which this route was.\n",
    "\n",
    "The lines must be **ordered in descending order** of the `count` field.\n",
    "\n",
    "## Criteria\n",
    "You can get maximum of 3.5 points (final grade) for this assignment, depedning on the number of interface you manage to leverage. The criteria are as follows:\n",
    "\n",
    "* 0.5 points – Spark SQL solution with 1 query\n",
    "* 0.5 points – Spark SQL solution with <=2 queries\n",
    "* 0.5 points – Spark RDD solution\n",
    "* 0.5 points – Spark DF solution\n",
    "* 0.5 points – your solution algorithm is relatively optimized, i.e.: no O^2 or O^3 complexities; appropriate object usage; no data leaks etc. This is evaluated by staff.\n",
    "* 1 point – 1 on 1 screening session. During this session staff member can ask you questions regarding your solution logic, framework usage, questionable parts of your code etc. If your code is clean enough, the staff member can just ask you to solve a theoretical problem connected to Spark.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: pyspark==3.2.4 in /opt/conda/lib/python3.10/site-packages (3.2.4)\n",
      "Requirement already satisfied: py4j==0.10.9.5 in /opt/conda/lib/python3.10/site-packages (from pyspark==3.2.4) (0.10.9.5)\n"
     ]
    }
   ],
   "source": [
    "!pip install pyspark==3.2.4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import necessary libraries\n",
    "import findspark, pyspark\n",
    "findspark.init()\n",
    "\n",
    "from pyspark.sql import SparkSession, functions as F\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark import SparkContext\n",
    "from itertools import groupby\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "2024-10-20 20:43:43,275 WARN yarn.Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.\n"
     ]
    }
   ],
   "source": [
    "# Initialize SparkSession and SparkContext\n",
    "spark = SparkSession.builder.appName(\"UserRoutes\").getOrCreate()\n",
    "sc = SparkContext.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://3d36e97c8b6d:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.2.4</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>yarn</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>UserRoutes</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7fffe1e17be0>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://3d36e97c8b6d:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.2.4</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>yarn</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>UserRoutes</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=yarn appName=UserRoutes>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Found 2 items\n",
      "-rw-r--r--   1 root supergroup     30.7 M 2023-09-24 20:38 /data/clickstream.csv\n",
      "drwxr-xr-x   - root supergroup          0 2023-09-24 20:38 /data/transactions\n"
     ]
    }
   ],
   "source": [
    "! hdfs dfs -ls -h /data\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "user_id\tsession_id\tevent_type\tevent_page\ttimestamp\n",
      "562\t507\tpage\tmain\t1695584127\n",
      "562\t507\tevent\tmain\t1695584134\n",
      "562\t507\tevent\tmain\t1695584144\n",
      "562\t507\tevent\tmain\t1695584147\n",
      "562\t507\twNaxLlerrorU\tmain\t1695584154\n",
      "562\t507\tevent\tmain\t1695584154\n",
      "562\t507\tevent\tmain\t1695584154\n",
      "562\t507\tevent\tmain\t1695584160\n",
      "562\t507\tpage\trabota\t1695584166\n",
      "562\t507\tevent\trabota\t1695584174\n",
      "562\t507\tevent\trabota\t1695584181\n",
      "562\t507\tevent\trabota\t1695584189\n",
      "562\t507\tpage\tmain\t1695584194\n",
      "562\t507\tevent\tmain\t1695584204\n",
      "562\t507\tevent\tmain\t1695584211\n",
      "562\t507\tevent\tmain\t1695584211\n",
      "562\t507\tevent\tmain\t1695584219\n",
      "562\t507\tpage\tbonus\t1695584221\n",
      "562\t507\tpage\tonline\t1695584222\n",
      "562\t507\tevent\tonline\t1695584230\n",
      "3539\t849\tpage\tmain\t1695584238\n",
      "3539\t849\tevent\tmain\t1695584252\n",
      "3539\t849\tpage\tonline\t1695584261\n",
      "3539\t849\tpage\tbonus\t1695584269\n",
      "3539\t849\tevent\tbonus\t1695584278\n",
      "3539\t849\tpage\tnews\t1695584285\n",
      "3539\t849\tpage\tmain\t1695584291\n",
      "3539\t849\tevent\tmain\t1695584301\n",
      "3539\t849\tpage\tnews\t1695584306\n",
      "3539\t849\tevent\tnews\t1695584307\n",
      "3539\t849\tpage\tvklad\t1695584317\n",
      "3539\t849\tpage\trabot"
     ]
    }
   ],
   "source": [
    "! hdfs dfs -head /data/clickstream.csv\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Step 1: Load Data from HDFS\n",
    "data_path = \"hdfs:///data/clickstream.csv\"\n",
    "clickstream_df = spark.read.option(\"header\", \"true\").option(\"delimiter\", \"\\t\").csv(data_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Register DataFrame as a temporary SQL table\n",
    "clickstream_df.createOrReplaceTempView(\"clickstream\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " ## Spark SQL solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Step 2: SQL-Based Approach for Cleaning and Route Extraction\n",
    "\n",
    "# SQL Query to flag errors, remove them, and extract the route per session\n",
    "sql_query = \"\"\"\n",
    "WITH ErrorFlagged AS (\n",
    "    SELECT\n",
    "        user_id,\n",
    "        session_id,\n",
    "        event_page,\n",
    "        timestamp,\n",
    "        MAX(CASE WHEN event_type LIKE '%error%' THEN 1 ELSE 0 END) \n",
    "            OVER (PARTITION BY user_id, session_id ORDER BY timestamp) AS error_flag,\n",
    "        LAG(event_page) OVER (PARTITION BY user_id, session_id ORDER BY timestamp) AS previous_page\n",
    "    FROM clickstream\n",
    "),\n",
    "FilteredRoutes AS (\n",
    "    SELECT\n",
    "        user_id,\n",
    "        session_id,\n",
    "        event_page,\n",
    "        timestamp\n",
    "    FROM ErrorFlagged\n",
    "    WHERE error_flag = 0 AND (previous_page IS NULL OR previous_page != event_page)\n",
    "),\n",
    "AggregatedRoutes AS (\n",
    "    SELECT\n",
    "        user_id,\n",
    "        session_id,\n",
    "        COLLECT_LIST(event_page) AS route\n",
    "    FROM FilteredRoutes\n",
    "    GROUP BY user_id, session_id\n",
    ")\n",
    "SELECT\n",
    "    CONCAT_WS('-', route) AS route_string,\n",
    "    COUNT(*) AS count\n",
    "FROM AggregatedRoutes\n",
    "GROUP BY route_string\n",
    "ORDER BY count DESC\n",
    "LIMIT 30\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 3:================================================>        (12 + 2) / 14]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|        route_string|count|\n",
      "+--------------------+-----+\n",
      "|                main| 8184|\n",
      "|        main-archive| 1112|\n",
      "|         main-rabota| 1047|\n",
      "|       main-internet|  897|\n",
      "|          main-bonus|  870|\n",
      "|           main-news|  769|\n",
      "|        main-tariffs|  677|\n",
      "|         main-online|  587|\n",
      "|          main-vklad|  517|\n",
      "| main-rabota-archive|  170|\n",
      "| main-archive-rabota|  167|\n",
      "|  main-bonus-archive|  143|\n",
      "|   main-rabota-bonus|  139|\n",
      "|   main-bonus-rabota|  135|\n",
      "|    main-news-rabota|  135|\n",
      "|main-archive-inte...|  131|\n",
      "|main-internet-rabota|  129|\n",
      "|    main-rabota-news|  129|\n",
      "|   main-archive-news|  126|\n",
      "|main-rabota-internet|  124|\n",
      "|main-internet-arc...|  123|\n",
      "|  main-archive-bonus|  117|\n",
      "| main-internet-bonus|  115|\n",
      "|main-tariffs-inte...|  114|\n",
      "|   main-news-archive|  113|\n",
      "|  main-news-internet|  109|\n",
      "|main-archive-tariffs|  104|\n",
      "|  main-internet-news|  103|\n",
      "|main-tariffs-archive|  103|\n",
      "|    main-rabota-main|   94|\n",
      "+--------------------+-----+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Execute the SQL query\n",
    "sql_result = spark.sql(sql_query)\n",
    "sql_result.show(30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark DF solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Step 3: DataFrame Approach for Cleaning and Route Extraction\n",
    "\n",
    "# 3.1 - Identify errors using a window function\n",
    "event_window = Window.partitionBy(\"user_id\", \"session_id\").orderBy(\"timestamp\")\n",
    "cleaned_df = (\n",
    "    clickstream_df.withColumn(\"error_flag\", F.max(F.when(F.col(\"event_type\").rlike(\".*[Ee]rror.*\"), 1).otherwise(0)).over(event_window))\n",
    "                  .filter(F.col(\"error_flag\") == 0)\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 3.2 - Filter out consecutive duplicate pages within a session\n",
    "cleaned_df = (\n",
    "    cleaned_df.withColumn(\"duplicate_flag\", \n",
    "                          F.when(F.lag(\"event_page\").over(event_window) == F.col(\"event_page\"), 1).otherwise(0))\n",
    "              .filter(F.col(\"duplicate_flag\") == 0)\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 4: Create and Aggregate Routes in DataFrame Approach\n",
    "routes_window = Window.partitionBy(\"user_id\", \"session_id\").orderBy(\"timestamp\")\n",
    "final_routes = (\n",
    "    cleaned_df.filter(F.col(\"event_type\") == \"page\")\n",
    "              .withColumn(\"route\", F.collect_list(\"event_page\").over(routes_window))\n",
    "              .groupBy(\"user_id\", \"session_id\")\n",
    "              .agg(F.max(\"route\").alias(\"route\"))\n",
    "              .withColumn(\"route_string\", F.concat_ws(\"-\", \"route\"))\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Calculate and Display the 30 Most Frequent Routes\n",
    "result_df = (\n",
    "    final_routes.groupBy(\"route_string\")\n",
    "                .count()\n",
    "                .orderBy(F.desc(\"count\"))\n",
    "                .limit(30)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 9:===================================================>     (17 + 2) / 19]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|        route_string|count|\n",
      "+--------------------+-----+\n",
      "|                main| 8184|\n",
      "|        main-archive| 1113|\n",
      "|         main-rabota| 1047|\n",
      "|       main-internet|  897|\n",
      "|          main-bonus|  870|\n",
      "|           main-news|  769|\n",
      "|        main-tariffs|  677|\n",
      "|         main-online|  587|\n",
      "|          main-vklad|  518|\n",
      "| main-rabota-archive|  170|\n",
      "| main-archive-rabota|  167|\n",
      "|  main-bonus-archive|  143|\n",
      "|   main-rabota-bonus|  139|\n",
      "|   main-bonus-rabota|  135|\n",
      "|    main-news-rabota|  135|\n",
      "|main-archive-inte...|  132|\n",
      "|    main-rabota-news|  130|\n",
      "|main-internet-rabota|  129|\n",
      "|   main-archive-news|  126|\n",
      "|main-rabota-internet|  124|\n",
      "|main-internet-arc...|  123|\n",
      "|  main-archive-bonus|  117|\n",
      "| main-internet-bonus|  115|\n",
      "|main-tariffs-inte...|  114|\n",
      "|   main-news-archive|  113|\n",
      "|  main-news-internet|  109|\n",
      "|main-archive-tariffs|  104|\n",
      "|  main-internet-news|  103|\n",
      "|main-tariffs-archive|  103|\n",
      "|    main-rabota-main|   94|\n",
      "+--------------------+-----+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "result_df.show(30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Spark RDD solution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 5: Alternative RDD Approach for Understanding the Logic\n",
    "\n",
    "# Create RDD from DataFrame\n",
    "rdd = clickstream_df.rdd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Identify errors in RDD\n",
    "error_events = (\n",
    "    rdd.filter(lambda row: 'error' in row['event_type'])\n",
    "       .map(lambda row: (f\"{row['user_id']}_{row['session_id']}\", int(row['timestamp'])))\n",
    "       .groupByKey()\n",
    "       .map(lambda entry: (entry[0], min(entry[1])))\n",
    "       .collectAsMap()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Broadcast error events to all nodes\n",
    "error_events_broadcast = sc.broadcast(error_events)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to remove consecutive duplicates from a list of pages\n",
    "def remove_consecutive_duplicates(entry):\n",
    "    sorted_pages = [page for page, _ in sorted(entry[1], key=lambda x: x[1])]\n",
    "    pages = [k for k, _ in groupby(sorted_pages)]\n",
    "    return entry[0], pages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Process RDD to extract and count routes\n",
    "routes_rdd = (\n",
    "    rdd.map(lambda row: (f\"{row['user_id']}_{row['session_id']}\", (row['event_type'], row['event_page'], int(row['timestamp']))))\n",
    "       .filter(lambda entry: entry[0] not in error_events_broadcast.value or entry[1][2] < error_events_broadcast.value[entry[0]])\n",
    "       .filter(lambda entry: entry[1][0] == \"page\")\n",
    "       .map(lambda entry: (entry[0], (entry[1][1], entry[1][2])))\n",
    "       .groupByKey()\n",
    "       .map(remove_consecutive_duplicates)\n",
    "       .map(lambda entry: (entry[0], \"-\".join(entry[1])))\n",
    "       .map(lambda entry: (entry[1], 1))\n",
    "       .reduceByKey(lambda a, b: a + b)\n",
    "       .sortBy(lambda x: x[1], ascending=False)\n",
    "       .take(30)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "main\t8184\n",
      "main-archive\t1113\n",
      "main-rabota\t1047\n",
      "main-internet\t897\n",
      "main-bonus\t870\n",
      "main-news\t769\n",
      "main-tariffs\t677\n",
      "main-online\t587\n",
      "main-vklad\t518\n",
      "main-rabota-archive\t170\n",
      "main-archive-rabota\t167\n",
      "main-bonus-archive\t143\n",
      "main-rabota-bonus\t139\n",
      "main-news-rabota\t135\n",
      "main-bonus-rabota\t135\n",
      "main-archive-internet\t132\n",
      "main-rabota-news\t130\n",
      "main-internet-rabota\t129\n",
      "main-archive-news\t126\n",
      "main-rabota-internet\t124\n",
      "main-internet-archive\t123\n",
      "main-archive-bonus\t117\n",
      "main-internet-bonus\t115\n",
      "main-tariffs-internet\t114\n",
      "main-news-archive\t113\n",
      "main-news-internet\t109\n",
      "main-archive-tariffs\t104\n",
      "main-tariffs-archive\t103\n",
      "main-internet-news\t103\n",
      "main-rabota-main\t94\n"
     ]
    }
   ],
   "source": [
    "# Display RDD results\n",
    "for route, count in routes_rdd:\n",
    "    print(f\"{route}\\t{count}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|        route_string|count|\n",
      "+--------------------+-----+\n",
      "|                main| 8184|\n",
      "|        main-archive| 1113|\n",
      "|         main-rabota| 1047|\n",
      "|       main-internet|  897|\n",
      "|          main-bonus|  870|\n",
      "|           main-news|  769|\n",
      "|        main-tariffs|  677|\n",
      "|         main-online|  587|\n",
      "|          main-vklad|  518|\n",
      "| main-rabota-archive|  170|\n",
      "| main-archive-rabota|  167|\n",
      "|  main-bonus-archive|  143|\n",
      "|   main-rabota-bonus|  139|\n",
      "|    main-news-rabota|  135|\n",
      "|   main-bonus-rabota|  135|\n",
      "|main-archive-inte...|  132|\n",
      "|    main-rabota-news|  130|\n",
      "|main-internet-rabota|  129|\n",
      "|   main-archive-news|  126|\n",
      "|main-rabota-internet|  124|\n",
      "|main-internet-arc...|  123|\n",
      "|  main-archive-bonus|  117|\n",
      "| main-internet-bonus|  115|\n",
      "|main-tariffs-inte...|  114|\n",
      "|   main-news-archive|  113|\n",
      "|  main-news-internet|  109|\n",
      "|main-archive-tariffs|  104|\n",
      "|main-tariffs-archive|  103|\n",
      "|  main-internet-news|  103|\n",
      "|    main-rabota-main|   94|\n",
      "+--------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Displaying RDD results in tabular format\n",
    "rdd_result_df = spark.createDataFrame(routes_rdd, [\"route_string\", \"count\"])\n",
    "rdd_result_df.show(30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Checking on the grader"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+---------+\n",
      "|        route_string|count|   method|\n",
      "+--------------------+-----+---------+\n",
      "|                main| 8184|      SQL|\n",
      "|        main-archive| 1112|      SQL|\n",
      "|         main-rabota| 1047|      SQL|\n",
      "|       main-internet|  897|      SQL|\n",
      "|          main-bonus|  870|      SQL|\n",
      "|           main-news|  769|      SQL|\n",
      "|        main-tariffs|  677|      SQL|\n",
      "|         main-online|  587|      SQL|\n",
      "|          main-vklad|  517|      SQL|\n",
      "| main-rabota-archive|  170|      SQL|\n",
      "| main-archive-rabota|  167|      SQL|\n",
      "|  main-bonus-archive|  143|      SQL|\n",
      "|   main-rabota-bonus|  139|      SQL|\n",
      "|   main-bonus-rabota|  135|      SQL|\n",
      "|    main-news-rabota|  135|      SQL|\n",
      "|main-archive-inte...|  131|      SQL|\n",
      "|main-internet-rabota|  129|      SQL|\n",
      "|    main-rabota-news|  129|      SQL|\n",
      "|   main-archive-news|  126|      SQL|\n",
      "|main-rabota-internet|  124|      SQL|\n",
      "|main-internet-arc...|  123|      SQL|\n",
      "|  main-archive-bonus|  117|      SQL|\n",
      "| main-internet-bonus|  115|      SQL|\n",
      "|main-tariffs-inte...|  114|      SQL|\n",
      "|   main-news-archive|  113|      SQL|\n",
      "|  main-news-internet|  109|      SQL|\n",
      "|main-archive-tariffs|  104|      SQL|\n",
      "|  main-internet-news|  103|      SQL|\n",
      "|main-tariffs-archive|  103|      SQL|\n",
      "|    main-rabota-main|   94|      SQL|\n",
      "|                main| 8184|DataFrame|\n",
      "|        main-archive| 1113|DataFrame|\n",
      "|         main-rabota| 1047|DataFrame|\n",
      "|       main-internet|  897|DataFrame|\n",
      "|          main-bonus|  870|DataFrame|\n",
      "|           main-news|  769|DataFrame|\n",
      "|        main-tariffs|  677|DataFrame|\n",
      "|         main-online|  587|DataFrame|\n",
      "|          main-vklad|  518|DataFrame|\n",
      "| main-rabota-archive|  170|DataFrame|\n",
      "| main-archive-rabota|  167|DataFrame|\n",
      "|  main-bonus-archive|  143|DataFrame|\n",
      "|   main-rabota-bonus|  139|DataFrame|\n",
      "|   main-bonus-rabota|  135|DataFrame|\n",
      "|    main-news-rabota|  135|DataFrame|\n",
      "|main-archive-inte...|  132|DataFrame|\n",
      "|    main-rabota-news|  130|DataFrame|\n",
      "|main-internet-rabota|  129|DataFrame|\n",
      "|   main-archive-news|  126|DataFrame|\n",
      "|main-rabota-internet|  124|DataFrame|\n",
      "|main-internet-arc...|  123|DataFrame|\n",
      "|  main-archive-bonus|  117|DataFrame|\n",
      "| main-internet-bonus|  115|DataFrame|\n",
      "|main-tariffs-inte...|  114|DataFrame|\n",
      "|   main-news-archive|  113|DataFrame|\n",
      "|  main-news-internet|  109|DataFrame|\n",
      "|main-archive-tariffs|  104|DataFrame|\n",
      "|  main-internet-news|  103|DataFrame|\n",
      "|main-tariffs-archive|  103|DataFrame|\n",
      "|    main-rabota-main|   94|DataFrame|\n",
      "|                main| 8184|      RDD|\n",
      "|        main-archive| 1113|      RDD|\n",
      "|         main-rabota| 1047|      RDD|\n",
      "|       main-internet|  897|      RDD|\n",
      "|          main-bonus|  870|      RDD|\n",
      "|           main-news|  769|      RDD|\n",
      "|        main-tariffs|  677|      RDD|\n",
      "|         main-online|  587|      RDD|\n",
      "|          main-vklad|  518|      RDD|\n",
      "| main-rabota-archive|  170|      RDD|\n",
      "| main-archive-rabota|  167|      RDD|\n",
      "|  main-bonus-archive|  143|      RDD|\n",
      "|   main-rabota-bonus|  139|      RDD|\n",
      "|    main-news-rabota|  135|      RDD|\n",
      "|   main-bonus-rabota|  135|      RDD|\n",
      "|main-archive-inte...|  132|      RDD|\n",
      "|    main-rabota-news|  130|      RDD|\n",
      "|main-internet-rabota|  129|      RDD|\n",
      "|   main-archive-news|  126|      RDD|\n",
      "|main-rabota-internet|  124|      RDD|\n",
      "|main-internet-arc...|  123|      RDD|\n",
      "|  main-archive-bonus|  117|      RDD|\n",
      "| main-internet-bonus|  115|      RDD|\n",
      "|main-tariffs-inte...|  114|      RDD|\n",
      "|   main-news-archive|  113|      RDD|\n",
      "|  main-news-internet|  109|      RDD|\n",
      "|main-archive-tariffs|  104|      RDD|\n",
      "|main-tariffs-archive|  103|      RDD|\n",
      "|  main-internet-news|  103|      RDD|\n",
      "|    main-rabota-main|   94|      RDD|\n",
      "+--------------------+-----+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import functions as F\n",
    "\n",
    "# Add an identifier to each result\n",
    "sql_result_with_id = sql_result.withColumn(\"method\", F.lit(\"SQL\"))\n",
    "result_df_with_id = result_df.withColumn(\"method\", F.lit(\"DataFrame\"))\n",
    "rdd_result_df_with_id = rdd_result_df.withColumn(\"method\", F.lit(\"RDD\"))\n",
    "\n",
    "# Union all three DataFrames together\n",
    "combined_result = sql_result_with_id.union(result_df_with_id).union(rdd_result_df_with_id)\n",
    "\n",
    "# Display the combined result table\n",
    "combined_result.show(90)  # Displaying 30 results for each method (SQL, DataFrame, and RDD)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'result.json'"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Assuming result_df is already defined from the DataFrame-based solution\n",
    "\n",
    "import json\n",
    "\n",
    "# Extracting the top 10 frequent routes from the result DataFrame\n",
    "top_10_routes_df = result_df.limit(10).collect()\n",
    "\n",
    "# Preparing the dictionary to save as JSON\n",
    "top_10_routes_dict = {f\"{row['route_string']}\": row['count'] for index, row in enumerate(top_10_routes_df)}\n",
    "\n",
    "# Defining the output file path\n",
    "output_file_path = \"result.json\"\n",
    "\n",
    "# Saving the dictionary to a JSON file\n",
    "with open(output_file_path, 'w') as json_file:\n",
    "    json.dump(top_10_routes_dict, json_file, indent=4)\n",
    "\n",
    "# Returning the file path for reference\n",
    "output_file_path\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.9999999999999999\n",
      "Correct main answer!\n",
      "Correct main-archive answer!\n",
      "Correct main-rabota answer!\n",
      "Correct main-internet answer!\n",
      "Correct main-bonus answer!\n",
      "Correct main-news answer!\n",
      "Correct main-tariffs answer!\n",
      "Correct main-online answer!\n",
      "Correct main-vklad answer!\n",
      "Correct main-rabota-archive answer!\n"
     ]
    }
   ],
   "source": [
    "!curl -F file=@result.json 51.250.123.136:80/MDS-LSML1/Andreevromano/w6/2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Stop Spark session\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "coursera": {
   "schema_names": [
    "week-4-spark-homework"
   ]
  },
  "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.10.9"
  },
  "vscode": {
   "interpreter": {
    "hash": "31f2aee4e71d21fbe5cf8b01ff0e069b9275f58929596ceb00d14d90e3e16cd6"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}