Ilin_6131_LR2_BigData.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "68877c33-085a-4f6b-a4a7-feede9d0c8e3",
   "metadata": {},
   "source": [
    "# Импортирование необходимых модулей"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "5c54a224-5cab-48a9-81d0-d89e060a69e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "from typing import List\n",
    "\n",
    "import pyspark.sql as sql\n",
    "\n",
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.types import DoubleType, IntegerType, ArrayType, StringType\n",
    "from pyspark.sql.functions import udf, explode, rank\n",
    "from pyspark.sql.functions import col, max, sum, desc, countDistinct"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1f1590a-13bf-4bea-9de1-9372533eb9f1",
   "metadata": {},
   "source": [
    "# Инициализация сессии"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "038bad39-8181-4f65-a2c7-555f9e639336",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"L2_reports_with_apache_spark\") \\\n",
    "    .config(\"spark.jars.packages\", \"com.databricks:spark-xml_2.12:0.13.0\")\\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "e404cd45-a855-48de-93cf-8b8b5765786a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'3.1.2.0-eep-800'"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "410d4fb9-a0ac-47bb-98ca-ec881b8ba780",
   "metadata": {},
   "source": [
    "Перенос локальных файлов в распределенную файловую систему(HDFS)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "93bb4aad-6808-4fb3-9fa5-3a906f533bbc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "put: `/user/glebilin6/programming-languages.csv': File exists\n"
     ]
    }
   ],
   "source": [
    "# !hadoop fs -put posts__sample.xml /user/glebilin6\n",
    "# !hadoop fs -put programming-languages.csv /user/glebilin6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "8081ccff-8d50-4f22-921f-54e9c11f131e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-rwxr-xr-x   3 glebilin6 glebilin6      40269 2023-11-25 20:16 /user/glebilin6/programming-languages.csv\n"
     ]
    }
   ],
   "source": [
    "# Проверка наличия файлов в HDFS\n",
    "# !hadoop fs -ls /user/glebilin6/posts__sample.xml\n",
    "# !hadoop fs -ls /user/glebilin6/programming-languages.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6983f59d-d684-4296-81c0-b6a3dfb98904",
   "metadata": {},
   "source": [
    "# Загрузка данных"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "73492328-ef2c-4a9b-a73e-6171fd3f6405",
   "metadata": {},
   "source": [
    "## Указание путей с файлами датасетов"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "d3b9e550-2243-485d-99d5-cf78baa708a5",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "# data_path = os.path.join(os.curdir, \"data\")\n",
    "posts_path = os.path.join(\"posts__sample.xml\")\n",
    "prog_lang_path = os.path.join(\"programming-languages.csv\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7af5f647-7891-44a9-84ea-a05996539b7d",
   "metadata": {},
   "source": [
    "## Чтение данных о упоминаемости языков программирования в постах на Stack Overflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "3944ab72-d0fd-4d6a-9807-d3bf962f7041",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Posts\n",
      "root\n",
      " |-- _AcceptedAnswerId: long (nullable = true)\n",
      " |-- _AnswerCount: long (nullable = true)\n",
      " |-- _Body: string (nullable = true)\n",
      " |-- _ClosedDate: timestamp (nullable = true)\n",
      " |-- _CommentCount: long (nullable = true)\n",
      " |-- _CommunityOwnedDate: timestamp (nullable = true)\n",
      " |-- _CreationDate: timestamp (nullable = true)\n",
      " |-- _FavoriteCount: long (nullable = true)\n",
      " |-- _Id: long (nullable = true)\n",
      " |-- _LastActivityDate: timestamp (nullable = true)\n",
      " |-- _LastEditDate: timestamp (nullable = true)\n",
      " |-- _LastEditorDisplayName: string (nullable = true)\n",
      " |-- _LastEditorUserId: long (nullable = true)\n",
      " |-- _OwnerDisplayName: string (nullable = true)\n",
      " |-- _OwnerUserId: long (nullable = true)\n",
      " |-- _ParentId: long (nullable = true)\n",
      " |-- _PostTypeId: long (nullable = true)\n",
      " |-- _Score: long (nullable = true)\n",
      " |-- _Tags: string (nullable = true)\n",
      " |-- _Title: string (nullable = true)\n",
      " |-- _ViewCount: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "posts_data = spark.read.format('xml').options(rowTag='row').load(posts_path)\n",
    "\n",
    "print(\"Posts\")\n",
    "posts_data.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf522166-ee62-4184-9d47-a9567e4427d0",
   "metadata": {},
   "source": [
    "## Чтение данных о языках программирования"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "1d858945-de74-4666-8e7c-8a557f1a8007",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Programming languages\n",
      "root\n",
      " |-- name: string (nullable = true)\n",
      " |-- wikipedia_url: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "prog_lang_data = spark.read\\\n",
    ".option(\"header\", True)\\\n",
    ".option(\"inferSchema\", True)\\\n",
    ".option(\"timestampFormat\", 'M/d/y H:m')\\\n",
    ".csv(prog_lang_path)\n",
    "\n",
    "print(\"Programming languages\")\n",
    "prog_lang_data.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4837ddd7-be02-44aa-b76f-87c7ea8e4859",
   "metadata": {},
   "source": [
    "# Работа с данными"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e50d7a3a-8882-4775-a8ab-29725a1fe161",
   "metadata": {},
   "source": [
    "## Функции для обработки содержимого столбцов с тегами и датой последней активности поста"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "b7a0e348-ece0-4e49-8eb3-59d055b3fc72",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_tags(tags_string):\n",
    "    if tags_string is None:\n",
    "        return []\n",
    "    \n",
    "    pattern = r'<(.+?)>'\n",
    "    tags = re.findall(pattern, tags_string)\n",
    "    \n",
    "    return tags\n",
    "\n",
    "def get_year(date_and_time):\n",
    "    return date_and_time.year\n",
    "\n",
    "get_tags_udf = udf(get_tags, ArrayType(StringType()))\n",
    "get_year_udf = udf(get_year, IntegerType())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d6d5036c-1f0d-4379-a70f-5fc8674fce3c",
   "metadata": {},
   "source": [
    "## Выбор необходимых столбцов"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "5a0a1004-e951-46fd-9bf2-161ae38114ef",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1 Row(tags=['c#', 'floating-point', 'type-conversion', 'double', 'decimal'], year=2019, views=42817)\n",
      "2 Row(tags=['html', 'css', 'internet-explorer-7'], year=2019, views=18214)\n",
      "3 Row(tags=[], year=2017, views=None)\n",
      "4 Row(tags=['c#', '.net', 'datetime'], year=2019, views=555183)\n",
      "5 Row(tags=['c#', 'datetime', 'time', 'datediff', 'relative-time-span'], year=2019, views=149445)\n",
      "6 Row(tags=[], year=2018, views=None)\n",
      "7 Row(tags=['html', 'browser', 'timezone', 'user-agent', 'timezone-offset'], year=2019, views=176405)\n",
      "8 Row(tags=['.net', 'math'], year=2018, views=123231)\n",
      "9 Row(tags=[], year=2010, views=None)\n",
      "10 Row(tags=[], year=2010, views=None)\n"
     ]
    }
   ],
   "source": [
    "posts_data_simplified = posts_data \\\n",
    "                        .withColumn(\"tags\", get_tags_udf(posts_data[\"_Tags\"])) \\\n",
    "                        .withColumn(\"year\", get_year_udf(posts_data[\"_LastActivityDate\"]))\n",
    "posts_data_simplified = posts_data_simplified.select(\n",
    "    col(\"tags\"),\n",
    "    col(\"year\"),\n",
    "    col(\"_ViewCount\").alias(\"views\")\n",
    ")\n",
    "\n",
    "first_rows = posts_data_simplified.head(10)\n",
    "for i, row in enumerate(first_rows):\n",
    "    print(i+1, row)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "274922b5-3583-4f7b-bdd1-0bd1079e485c",
   "metadata": {},
   "source": [
    "## Составление полного отчета"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "0967b9ff-cf1c-4ded-83d4-5601827ce30b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------------+-----------+\n",
      "|year|                 tag|total_views|\n",
      "+----+--------------------+-----------+\n",
      "|2008|                  c#|      25401|\n",
      "|2008|                .net|      24321|\n",
      "|2008|            database|      19682|\n",
      "|2008|               local|      19682|\n",
      "|2008|                java|      11532|\n",
      "|2008|         inheritance|      10971|\n",
      "|2008|       accessibility|       7700|\n",
      "|2008|           variables|       7700|\n",
      "|2008|               excel|       6540|\n",
      "|2008|          automation|       6540|\n",
      "|2008|           interface|       3271|\n",
      "|2008|      castle-windsor|       2927|\n",
      "|2008|dependency-injection|       2927|\n",
      "|2008|       configuration|       2927|\n",
      "|2008|               linux|       2393|\n",
      "|2008|       ruby-on-rails|       1843|\n",
      "|2008|                ruby|       1843|\n",
      "|2008|            .net-3.0|       1432|\n",
      "|2008|  visual-studio-2008|       1432|\n",
      "|2008|  visual-studio-2005|       1432|\n",
      "+----+--------------------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Разбиение массива тегов на отдельные столбцы\n",
    "posts_data_sorted = posts_data_simplified.select(\"year\", explode(\"tags\").alias(\"tag\"), \"views\")\n",
    "\n",
    "# Группировка по году последней активности и тегам, суммирование всех просмотров для каждого языка программирования в пределах одного года\n",
    "posts_data_sorted = posts_data_sorted.groupBy(\"year\", \"tag\").agg(sum(\"views\").alias(\"total_views\"))\n",
    "\n",
    "# Сортировка по году и количеству просмотров\n",
    "posts_data_sorted = posts_data_sorted.orderBy(\"year\", desc(\"total_views\"))\n",
    "\n",
    "# Отображение отчета\n",
    "posts_data_sorted.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e9473f4d-91d8-4e68-b3f0-9d1cae92912e",
   "metadata": {},
   "source": [
    "## Составление итогового отчета"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "089ad1aa-8f74-4c13-bb17-6483c174be9b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------------------+-----------+\n",
      "|year|                 tag|total_views|\n",
      "+----+--------------------+-----------+\n",
      "|2008|                  c#|      25401|\n",
      "|2008|                .net|      24321|\n",
      "|2008|            database|      19682|\n",
      "|2008|               local|      19682|\n",
      "|2008|                java|      11532|\n",
      "|2009|                  c#|      73661|\n",
      "|2009|                .net|      39167|\n",
      "|2009|              python|      32219|\n",
      "|2009|                 c++|      29381|\n",
      "|2009|            winforms|      25670|\n",
      "|2010|                  c#|     128597|\n",
      "|2010|              arrays|      80868|\n",
      "|2010|                java|      53333|\n",
      "|2010|              matlab|      51865|\n",
      "|2010|multidimensional-...|      51865|\n",
      "|2011|                  c#|     238076|\n",
      "|2011|                java|     121315|\n",
      "|2011|                .net|     120734|\n",
      "|2011|                 css|     119302|\n",
      "|2011|             android|     107283|\n",
      "+----+--------------------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#  Определяем спецификацию Window, разбитую по годам и упорядоченную в порядке убывания по показателю total_views\n",
    "window_spec = Window.partitionBy(\"year\").orderBy(posts_data_sorted[\"total_views\"].desc())\n",
    "\n",
    "# Добавляем колонку rank в DataFrame\n",
    "ranked_df = posts_data_sorted.withColumn(\"rank\", rank().over(window_spec))\n",
    "\n",
    "# Отбираем первые 5 строк (5 наиболее популярных языков программирования в определенный год)\n",
    "result_df = ranked_df.filter(ranked_df[\"rank\"] <= 5)\n",
    "\n",
    "# Выбирам колоки год, язык программирования и общее число просмотров\n",
    "result_df = result_df.select(\"year\", \"tag\", \"total_views\")\n",
    "\n",
    "# Представляем результат по топ 5 самым популярным языкам программирования на StackOverflow с 2008 года\n",
    "posts_data_sorted_result = result_df.orderBy(\"year\", desc(\"total_views\"))\n",
    "\n",
    "# Show the final DataFrame\n",
    "posts_data_sorted_result.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "0b52f90f-d88e-403e-b6c6-20e73b1bf2da",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Сохраняем данные в формат Parquet\n",
    "import tempfile\n",
    "with tempfile.TemporaryDirectory() as d:\n",
    "    # Write a DataFrame into a Parquet file\n",
    "    posts_data_sorted_result.write.parquet(d)\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "facf15ea-3524-48bc-b581-f5a5efb4937b",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}