LSH4.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "authorship_tag": "ABX9TyMrpSAn3P433b+9ziNHYLLG",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/isabellacadisco/Algorithms-for-Massive-Data-course/blob/main/LSH4.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "id": "EIRt-eoQIRBv"
      },
      "outputs": [],
      "source": [
        "import os"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# download dataset from kaggle and save it in drive\n",
        "\n",
        "\n",
        "#import os\n",
        "os.environ['KAGGLE_USERNAME'] = \"isabellacadisco\"\n",
        "os.environ['KAGGLE_KEY'] = \"847325691c2399e41dc9c0b78fbe499e\"\n",
        "!kaggle datasets download -d xhlulu/medal-emnlp"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Q448BThyIi1s",
        "outputId": "9e1b08f9-ddfc-46ce-ecb6-5839f3effcc2"
      },
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Downloading medal-emnlp.zip to /content\n",
            "100% 6.81G/6.82G [01:30<00:00, 91.8MB/s]\n",
            "100% 6.82G/6.82G [01:30<00:00, 81.1MB/s]\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!unzip medal-emnlp.zip #unzip dataset"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "tjTyHo68Ijqs",
        "outputId": "b26ff036-12f7-4d9b-c47d-51327c218116"
      },
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Archive:  medal-emnlp.zip\n",
            "  inflating: full_data.csv           \n",
            "  inflating: pretrain_subset/test.csv  \n",
            "  inflating: pretrain_subset/train.csv  \n",
            "  inflating: pretrain_subset/valid.csv  \n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# set up spark\n",
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null #install java as VM\n",
        "!wget -q https://dlcdn.apache.org/spark/spark-3.4.0/spark-3.4.0-bin-hadoop3.tgz #download Spark\n",
        "!tar xf spark-3.4.0-bin-hadoop3.tgz #unzip the file\n",
        "!pip install -q findspark"
      ],
      "metadata": {
        "id": "hqarKL1nIkqo"
      },
      "execution_count": 4,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark-3.4.0-bin-hadoop3\""
      ],
      "metadata": {
        "id": "qQhx_0peImBA"
      },
      "execution_count": 5,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import findspark\n",
        "findspark.init(\"spark-3.4.0-bin-hadoop3\")# SPARK_HOME\n",
        "from pyspark.sql import SparkSession\n",
        "spark = SparkSession.builder.master(\"local[*]\").getOrCreate() #SPARK SESSION"
      ],
      "metadata": {
        "id": "Es9Ts25YInJw"
      },
      "execution_count": 6,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import pyspark\n",
        "type(spark)"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "b7_U_8Y1IoXs",
        "outputId": "75b5c412-508d-4d86-89cb-a4d7b4f8c06c"
      },
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "pyspark.sql.session.SparkSession"
            ]
          },
          "metadata": {},
          "execution_count": 7
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "file_path = './full_data.csv'"
      ],
      "metadata": {
        "id": "oKrIbpJKIp4c"
      },
      "execution_count": 8,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "data_df = spark.read.csv(file_path, header=True)"
      ],
      "metadata": {
        "id": "Pyw-Z0l_IsBm"
      },
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql.functions import monotonically_increasing_id"
      ],
      "metadata": {
        "id": "zgarTiNyI4ry"
      },
      "execution_count": 10,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "dataset = data_df.select('TEXT').withColumn('doc_id', monotonically_increasing_id())"
      ],
      "metadata": {
        "id": "BqzPAv0TI5EM"
      },
      "execution_count": 11,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "dataset.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Zgz13JqeI6HF",
        "outputId": "24908b30-91be-440f-84c5-2fd7b8d820e8"
      },
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------------+------+\n",
            "|                TEXT|doc_id|\n",
            "+--------------------+------+\n",
            "|alphabisabolol ha...|     0|\n",
            "|a report is given...|     1|\n",
            "|the virostatic co...|     2|\n",
            "|rmi rmi and rmi a...|     3|\n",
            "|a doubleblind stu...|     4|\n",
            "|stroma from eithe...|     5|\n",
            "|the effect of the...|     6|\n",
            "|in one experiment...|     7|\n",
            "|the presence of a...|     8|\n",
            "|the reaction of g...|     9|\n",
            "|choline acetyltra...|    10|\n",
            "|increasing concen...|    11|\n",
            "|the properties of...|    12|\n",
            "|primary amines re...|    13|\n",
            "|a purification pr...|    14|\n",
            "|dihydrofolate red...|    15|\n",
            "|ionization effect...|    16|\n",
            "|kinetic analyses ...|    17|\n",
            "|the nearultraviol...|    18|\n",
            "|the circular pola...|    19|\n",
            "+--------------------+------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "small_dataset = dataset.sample(withReplacement=False, fraction=0.000007, seed=42)"
      ],
      "metadata": {
        "id": "x9GK0hrjI9bw"
      },
      "execution_count": 13,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "small_dataset.count()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "S-m_LHLoJf4I",
        "outputId": "fcce59d6-d027-440f-afc8-327d35c22ef5"
      },
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "94"
            ]
          },
          "metadata": {},
          "execution_count": 14
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "small_dataset.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "6htErXW7Jism",
        "outputId": "711f42c8-4435-4d7a-9cd0-accbe980d62b"
      },
      "execution_count": 15,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------------+------------+\n",
            "|                TEXT|      doc_id|\n",
            "+--------------------+------------+\n",
            "|a recent outbreak...|      106539|\n",
            "|wefast gradientec...| 25769866227|\n",
            "|this paper deals ...| 42949764250|\n",
            "|there is a close ...| 51539638497|\n",
            "|occasional and un...| 51539642085|\n",
            "|the relative perc...| 77309421895|\n",
            "|mixed alloys for ...|103079248410|\n",
            "|a high incidence ...|103079293994|\n",
            "|a housetohouse su...|103079331152|\n",
            "|to characterize t...|120259124985|\n",
            "|we studied the ef...|120259129263|\n",
            "|the effects of cl...|137439020479|\n",
            "|lipopolysaccharid...|154618911990|\n",
            "|this case report ...|154618921437|\n",
            "|a study of one of...|163208873442|\n",
            "|rat fetuses at da...|188978671675|\n",
            "|the catenins are ...|206158468968|\n",
            "|when tissues are ...|214748392950|\n",
            "|SS concentrations...|214748459668|\n",
            "|the sensitivity o...|223338408901|\n",
            "+--------------------+------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# TEXT PROCESSOR\n",
        "!pip install spacy\n",
        "!python -m spacy download en_core_web_sm\n",
        "\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "S2WpIvjSJk_l",
        "outputId": "11028b82-6a2b-4f80-e90d-b4e5f0a60cfb"
      },
      "execution_count": 16,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Requirement already satisfied: spacy in /usr/local/lib/python3.10/dist-packages (3.6.1)\n",
            "Requirement already satisfied: spacy-legacy<3.1.0,>=3.0.11 in /usr/local/lib/python3.10/dist-packages (from spacy) (3.0.12)\n",
            "Requirement already satisfied: spacy-loggers<2.0.0,>=1.0.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (1.0.4)\n",
            "Requirement already satisfied: murmurhash<1.1.0,>=0.28.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (1.0.9)\n",
            "Requirement already satisfied: cymem<2.1.0,>=2.0.2 in /usr/local/lib/python3.10/dist-packages (from spacy) (2.0.7)\n",
            "Requirement already satisfied: preshed<3.1.0,>=3.0.2 in /usr/local/lib/python3.10/dist-packages (from spacy) (3.0.8)\n",
            "Requirement already satisfied: thinc<8.2.0,>=8.1.8 in /usr/local/lib/python3.10/dist-packages (from spacy) (8.1.12)\n",
            "Requirement already satisfied: wasabi<1.2.0,>=0.9.1 in /usr/local/lib/python3.10/dist-packages (from spacy) (1.1.2)\n",
            "Requirement already satisfied: srsly<3.0.0,>=2.4.3 in /usr/local/lib/python3.10/dist-packages (from spacy) (2.4.7)\n",
            "Requirement already satisfied: catalogue<2.1.0,>=2.0.6 in /usr/local/lib/python3.10/dist-packages (from spacy) (2.0.9)\n",
            "Requirement already satisfied: typer<0.10.0,>=0.3.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (0.9.0)\n",
            "Requirement already satisfied: pathy>=0.10.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (0.10.2)\n",
            "Requirement already satisfied: smart-open<7.0.0,>=5.2.1 in /usr/local/lib/python3.10/dist-packages (from spacy) (6.4.0)\n",
            "Requirement already satisfied: tqdm<5.0.0,>=4.38.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (4.66.1)\n",
            "Requirement already satisfied: numpy>=1.15.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (1.23.5)\n",
            "Requirement already satisfied: requests<3.0.0,>=2.13.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (2.31.0)\n",
            "Requirement already satisfied: pydantic!=1.8,!=1.8.1,<3.0.0,>=1.7.4 in /usr/local/lib/python3.10/dist-packages (from spacy) (1.10.12)\n",
            "Requirement already satisfied: jinja2 in /usr/local/lib/python3.10/dist-packages (from spacy) (3.1.2)\n",
            "Requirement already satisfied: setuptools in /usr/local/lib/python3.10/dist-packages (from spacy) (67.7.2)\n",
            "Requirement already satisfied: packaging>=20.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (23.1)\n",
            "Requirement already satisfied: langcodes<4.0.0,>=3.2.0 in /usr/local/lib/python3.10/dist-packages (from spacy) (3.3.0)\n",
            "Requirement already satisfied: typing-extensions>=4.2.0 in /usr/local/lib/python3.10/dist-packages (from pydantic!=1.8,!=1.8.1,<3.0.0,>=1.7.4->spacy) (4.5.0)\n",
            "Requirement already satisfied: charset-normalizer<4,>=2 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.13.0->spacy) (3.2.0)\n",
            "Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.13.0->spacy) (3.4)\n",
            "Requirement already satisfied: urllib3<3,>=1.21.1 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.13.0->spacy) (2.0.4)\n",
            "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.13.0->spacy) (2023.7.22)\n",
            "Requirement already satisfied: blis<0.8.0,>=0.7.8 in /usr/local/lib/python3.10/dist-packages (from thinc<8.2.0,>=8.1.8->spacy) (0.7.10)\n",
            "Requirement already satisfied: confection<1.0.0,>=0.0.1 in /usr/local/lib/python3.10/dist-packages (from thinc<8.2.0,>=8.1.8->spacy) (0.1.2)\n",
            "Requirement already satisfied: click<9.0.0,>=7.1.1 in /usr/local/lib/python3.10/dist-packages (from typer<0.10.0,>=0.3.0->spacy) (8.1.7)\n",
            "Requirement already satisfied: MarkupSafe>=2.0 in /usr/local/lib/python3.10/dist-packages (from jinja2->spacy) (2.1.3)\n",
            "2023-09-11 21:37:29.960582: I tensorflow/core/platform/cpu_feature_guard.cc:182] This TensorFlow binary is optimized to use available CPU instructions in performance-critical operations.\n",
            "To enable the following instructions: AVX2 FMA, in other operations, rebuild TensorFlow with the appropriate compiler flags.\n",
            "2023-09-11 21:37:32.039187: W tensorflow/compiler/tf2tensorrt/utils/py_utils.cc:38] TF-TRT Warning: Could not find TensorRT\n",
            "Collecting en-core-web-sm==3.6.0\n",
            "  Downloading https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.6.0/en_core_web_sm-3.6.0-py3-none-any.whl (12.8 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m12.8/12.8 MB\u001b[0m \u001b[31m41.0 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hRequirement already satisfied: spacy<3.7.0,>=3.6.0 in /usr/local/lib/python3.10/dist-packages (from en-core-web-sm==3.6.0) (3.6.1)\n",
            "Requirement already satisfied: spacy-legacy<3.1.0,>=3.0.11 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (3.0.12)\n",
            "Requirement already satisfied: spacy-loggers<2.0.0,>=1.0.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (1.0.4)\n",
            "Requirement already satisfied: murmurhash<1.1.0,>=0.28.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (1.0.9)\n",
            "Requirement already satisfied: cymem<2.1.0,>=2.0.2 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (2.0.7)\n",
            "Requirement already satisfied: preshed<3.1.0,>=3.0.2 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (3.0.8)\n",
            "Requirement already satisfied: thinc<8.2.0,>=8.1.8 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (8.1.12)\n",
            "Requirement already satisfied: wasabi<1.2.0,>=0.9.1 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (1.1.2)\n",
            "Requirement already satisfied: srsly<3.0.0,>=2.4.3 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (2.4.7)\n",
            "Requirement already satisfied: catalogue<2.1.0,>=2.0.6 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (2.0.9)\n",
            "Requirement already satisfied: typer<0.10.0,>=0.3.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (0.9.0)\n",
            "Requirement already satisfied: pathy>=0.10.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (0.10.2)\n",
            "Requirement already satisfied: smart-open<7.0.0,>=5.2.1 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (6.4.0)\n",
            "Requirement already satisfied: tqdm<5.0.0,>=4.38.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (4.66.1)\n",
            "Requirement already satisfied: numpy>=1.15.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (1.23.5)\n",
            "Requirement already satisfied: requests<3.0.0,>=2.13.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (2.31.0)\n",
            "Requirement already satisfied: pydantic!=1.8,!=1.8.1,<3.0.0,>=1.7.4 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (1.10.12)\n",
            "Requirement already satisfied: jinja2 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (3.1.2)\n",
            "Requirement already satisfied: setuptools in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (67.7.2)\n",
            "Requirement already satisfied: packaging>=20.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (23.1)\n",
            "Requirement already satisfied: langcodes<4.0.0,>=3.2.0 in /usr/local/lib/python3.10/dist-packages (from spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (3.3.0)\n",
            "Requirement already satisfied: typing-extensions>=4.2.0 in /usr/local/lib/python3.10/dist-packages (from pydantic!=1.8,!=1.8.1,<3.0.0,>=1.7.4->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (4.5.0)\n",
            "Requirement already satisfied: charset-normalizer<4,>=2 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.13.0->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (3.2.0)\n",
            "Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.13.0->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (3.4)\n",
            "Requirement already satisfied: urllib3<3,>=1.21.1 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.13.0->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (2.0.4)\n",
            "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.13.0->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (2023.7.22)\n",
            "Requirement already satisfied: blis<0.8.0,>=0.7.8 in /usr/local/lib/python3.10/dist-packages (from thinc<8.2.0,>=8.1.8->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (0.7.10)\n",
            "Requirement already satisfied: confection<1.0.0,>=0.0.1 in /usr/local/lib/python3.10/dist-packages (from thinc<8.2.0,>=8.1.8->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (0.1.2)\n",
            "Requirement already satisfied: click<9.0.0,>=7.1.1 in /usr/local/lib/python3.10/dist-packages (from typer<0.10.0,>=0.3.0->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (8.1.7)\n",
            "Requirement already satisfied: MarkupSafe>=2.0 in /usr/local/lib/python3.10/dist-packages (from jinja2->spacy<3.7.0,>=3.6.0->en-core-web-sm==3.6.0) (2.1.3)\n",
            "\u001b[38;5;2m✔ Download and installation successful\u001b[0m\n",
            "You can now load the package via spacy.load('en_core_web_sm')\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Import spaCy and download the English model\n",
        "import spacy\n",
        "import en_core_web_sm\n"
      ],
      "metadata": {
        "id": "VKfDdmHLJxTD"
      },
      "execution_count": 17,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Load the spaCy English model FORSE NON SERVE BOH\n",
        "nlp = en_core_web_sm.load()"
      ],
      "metadata": {
        "id": "HQvgsPw7Jx3L"
      },
      "execution_count": 18,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "class TextPreprocessor:\n",
        "    def __init__(self):\n",
        "        # Initialize spaCy resources\n",
        "        self.nlp = spacy.load(\"en_core_web_sm\")\n",
        "\n",
        "    def preprocess(self, document):\n",
        "        # Apply spaCy NLP pipeline\n",
        "        doc = self.nlp(document)\n",
        "\n",
        "        # Lemmatize and remove stopwords\n",
        "        tokens = [token.lemma_ for token in doc if not token.is_stop]\n",
        "\n",
        "        # Join tokens back into a text\n",
        "        cleaned_text = \" \".join(tokens)\n",
        "\n",
        "        return cleaned_text"
      ],
      "metadata": {
        "id": "BWVu602xJ0fi"
      },
      "execution_count": 19,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "class TextPreprocessor_map:\n",
        "    def __init__(self, dataframe):\n",
        "        # Initialize spaCy resources\n",
        "        self.nlp = spacy.load(\"en_core_web_sm\")\n",
        "\n",
        "        doc_id = dataframe[1]\n",
        "        text = dataframe[0]\n",
        "\n",
        "    def preprocess(self, dataframe):\n",
        "        # Apply spaCy NLP pipeline\n",
        "        doc = self.nlp(text)\n",
        "\n",
        "        # Lemmatize and remove stopwords\n",
        "        tokens = [token.lemma_ for token in doc if not token.is_stop]\n",
        "\n",
        "        # Join tokens back into a text\n",
        "        cleaned_text = \" \".join(tokens)\n",
        "\n",
        "        return (text, doc)"
      ],
      "metadata": {
        "id": "RgOa1_xrQpJG"
      },
      "execution_count": 43,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "data_rdd = small_dataset.rdd"
      ],
      "metadata": {
        "id": "5hRe7vBkSmhy"
      },
      "execution_count": 47,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "map_prep = TextPreprocessor_map"
      ],
      "metadata": {
        "id": "1-g8P6duRnwn"
      },
      "execution_count": 44,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "cleaned_data = data_rdd.map(map_prep)"
      ],
      "metadata": {
        "id": "H1ghGMfWRQi9"
      },
      "execution_count": 48,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "cleaned_data.toDF().show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 989
        },
        "id": "Dd8CqMR8SN2C",
        "outputId": "c940f1a3-4a19-4169-c15a-132c1cf6bd4d"
      },
      "execution_count": 49,
      "outputs": [
        {
          "output_type": "error",
          "ename": "TypeError",
          "evalue": "ignored",
          "traceback": [
            "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
            "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_type\u001b[0;34m(obj, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1527\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1528\u001b[0;31m             return _infer_schema(\n\u001b[0m\u001b[1;32m   1529\u001b[0m                 \u001b[0mobj\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_schema\u001b[0;34m(row, names, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1564\u001b[0m     \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1565\u001b[0;31m         \u001b[0;32mraise\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Can not infer schema for type: %s\"\u001b[0m \u001b[0;34m%\u001b[0m \u001b[0mtype\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mrow\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1566\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mTypeError\u001b[0m: Can not infer schema for type: <class 'spacy.pipeline.dep_parser.DependencyParser'>",
            "\nDuring handling of the above exception, another exception occurred:\n",
            "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_schema\u001b[0;34m(row, names, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1572\u001b[0m                     \u001b[0mk\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1573\u001b[0;31m                     _infer_type(\n\u001b[0m\u001b[1;32m   1574\u001b[0m                         \u001b[0mv\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_type\u001b[0;34m(obj, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1533\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1534\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"not supported type: %s\"\u001b[0m \u001b[0;34m%\u001b[0m \u001b[0mtype\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mobj\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1535\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mTypeError\u001b[0m: not supported type: <class 'spacy.pipeline.dep_parser.DependencyParser'>",
            "\nThe above exception was the direct cause of the following exception:\n",
            "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_type\u001b[0;34m(obj, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1527\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1528\u001b[0;31m             return _infer_schema(\n\u001b[0m\u001b[1;32m   1529\u001b[0m                 \u001b[0mobj\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_schema\u001b[0;34m(row, names, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1582\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mTypeError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1583\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Unable to infer the type of the field {}.\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mk\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1584\u001b[0m     \u001b[0;32mreturn\u001b[0m \u001b[0mStructType\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfields\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mTypeError\u001b[0m: Unable to infer the type of the field _2.",
            "\nDuring handling of the above exception, another exception occurred:\n",
            "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_schema\u001b[0;34m(row, names, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1572\u001b[0m                     \u001b[0mk\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1573\u001b[0;31m                     _infer_type(\n\u001b[0m\u001b[1;32m   1574\u001b[0m                         \u001b[0mv\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_type\u001b[0;34m(obj, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1513\u001b[0m                 return ArrayType(\n\u001b[0;32m-> 1514\u001b[0;31m                     reduce(\n\u001b[0m\u001b[1;32m   1515\u001b[0m                         \u001b[0m_merge_type\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m<genexpr>\u001b[0;34m(.0)\u001b[0m\n\u001b[1;32m   1515\u001b[0m                         \u001b[0m_merge_type\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1516\u001b[0;31m                         \u001b[0;34m(\u001b[0m\u001b[0m_infer_type\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mv\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0minfer_dict_as_struct\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mprefer_timestamp_ntz\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mfor\u001b[0m \u001b[0mv\u001b[0m \u001b[0;32min\u001b[0m \u001b[0mobj\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1517\u001b[0m                     ),\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_type\u001b[0;34m(obj, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1533\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1534\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"not supported type: %s\"\u001b[0m \u001b[0;34m%\u001b[0m \u001b[0mtype\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mobj\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1535\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mTypeError\u001b[0m: not supported type: <class 'tuple'>",
            "\nThe above exception was the direct cause of the following exception:\n",
            "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_type\u001b[0;34m(obj, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1527\u001b[0m         \u001b[0;32mtry\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1528\u001b[0;31m             return _infer_schema(\n\u001b[0m\u001b[1;32m   1529\u001b[0m                 \u001b[0mobj\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_schema\u001b[0;34m(row, names, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1582\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mTypeError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1583\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Unable to infer the type of the field {}.\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mk\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1584\u001b[0m     \u001b[0;32mreturn\u001b[0m \u001b[0mStructType\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfields\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mTypeError\u001b[0m: Unable to infer the type of the field _components.",
            "\nDuring handling of the above exception, another exception occurred:\n",
            "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_schema\u001b[0;34m(row, names, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1572\u001b[0m                     \u001b[0mk\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1573\u001b[0;31m                     _infer_type(\n\u001b[0m\u001b[1;32m   1574\u001b[0m                         \u001b[0mv\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_type\u001b[0;34m(obj, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1533\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1534\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"not supported type: %s\"\u001b[0m \u001b[0;34m%\u001b[0m \u001b[0mtype\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mobj\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1535\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mTypeError\u001b[0m: not supported type: <class 'spacy.lang.en.English'>",
            "\nThe above exception was the direct cause of the following exception:\n",
            "\u001b[0;31mTypeError\u001b[0m                                 Traceback (most recent call last)",
            "\u001b[0;32m<ipython-input-49-818494101d63>\u001b[0m in \u001b[0;36m<cell line: 1>\u001b[0;34m()\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mcleaned_data\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtoDF\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mshow\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/session.py\u001b[0m in \u001b[0;36mtoDF\u001b[0;34m(self, schema, sampleRatio)\u001b[0m\n\u001b[1;32m    113\u001b[0m         \u001b[0;34m+\u001b[0m\u001b[0;34m-\u001b[0m\u001b[0;34m-\u001b[0m\u001b[0;34m-\u001b[0m\u001b[0;34m+\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    114\u001b[0m         \"\"\"\n\u001b[0;32m--> 115\u001b[0;31m         \u001b[0;32mreturn\u001b[0m \u001b[0msparkSession\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcreateDataFrame\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mschema\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msampleRatio\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    116\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    117\u001b[0m     \u001b[0mRDD\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtoDF\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mtoDF\u001b[0m  \u001b[0;31m# type: ignore[assignment]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/session.py\u001b[0m in \u001b[0;36mcreateDataFrame\u001b[0;34m(self, data, schema, samplingRatio, verifySchema)\u001b[0m\n\u001b[1;32m   1274\u001b[0m                 \u001b[0mdata\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mschema\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msamplingRatio\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mverifySchema\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1275\u001b[0m             )\n\u001b[0;32m-> 1276\u001b[0;31m         return self._create_dataframe(\n\u001b[0m\u001b[1;32m   1277\u001b[0m             \u001b[0mdata\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mschema\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msamplingRatio\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mverifySchema\u001b[0m  \u001b[0;31m# type: ignore[arg-type]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1278\u001b[0m         )\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/session.py\u001b[0m in \u001b[0;36m_create_dataframe\u001b[0;34m(self, data, schema, samplingRatio, verifySchema)\u001b[0m\n\u001b[1;32m   1314\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1315\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0misinstance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdata\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mRDD\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1316\u001b[0;31m             \u001b[0mrdd\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstruct\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_createFromRDD\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mdata\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mmap\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mprepare\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mschema\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msamplingRatio\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1317\u001b[0m         \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1318\u001b[0m             \u001b[0mrdd\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mstruct\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_createFromLocal\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mmap\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mprepare\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mdata\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mschema\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/session.py\u001b[0m in \u001b[0;36m_createFromRDD\u001b[0;34m(self, rdd, schema, samplingRatio)\u001b[0m\n\u001b[1;32m    929\u001b[0m         \"\"\"\n\u001b[1;32m    930\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mschema\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mNone\u001b[0m \u001b[0;32mor\u001b[0m \u001b[0misinstance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mschema\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;34m(\u001b[0m\u001b[0mlist\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mtuple\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 931\u001b[0;31m             \u001b[0mstruct\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_inferSchema\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mrdd\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0msamplingRatio\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mnames\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mschema\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    932\u001b[0m             \u001b[0mconverter\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0m_create_converter\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mstruct\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    933\u001b[0m             \u001b[0mtupled_rdd\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mrdd\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mmap\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mconverter\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/session.py\u001b[0m in \u001b[0;36m_inferSchema\u001b[0;34m(self, rdd, samplingRatio, names)\u001b[0m\n\u001b[1;32m    880\u001b[0m         \u001b[0mprefer_timestamp_ntz\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mis_timestamp_ntz_preferred\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    881\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0msamplingRatio\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 882\u001b[0;31m             schema = _infer_schema(\n\u001b[0m\u001b[1;32m    883\u001b[0m                 \u001b[0mfirst\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    884\u001b[0m                 \u001b[0mnames\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mnames\u001b[0m\u001b[0;34m,\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/content/spark-3.4.0-bin-hadoop3/python/pyspark/sql/types.py\u001b[0m in \u001b[0;36m_infer_schema\u001b[0;34m(row, names, infer_dict_as_struct, infer_array_from_first_element, prefer_timestamp_ntz)\u001b[0m\n\u001b[1;32m   1581\u001b[0m             )\n\u001b[1;32m   1582\u001b[0m         \u001b[0;32mexcept\u001b[0m \u001b[0mTypeError\u001b[0m \u001b[0;32mas\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1583\u001b[0;31m             \u001b[0;32mraise\u001b[0m \u001b[0mTypeError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Unable to infer the type of the field {}.\"\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mformat\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mk\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0me\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m   1584\u001b[0m     \u001b[0;32mreturn\u001b[0m \u001b[0mStructType\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mfields\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1585\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mTypeError\u001b[0m: Unable to infer the type of the field nlp."
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "text_preprocessor_spacy = TextPreprocessor()"
      ],
      "metadata": {
        "id": "ayVfcvXSJ6W8"
      },
      "execution_count": 20,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import string"
      ],
      "metadata": {
        "id": "X6ipXOdTUwj5"
      },
      "execution_count": 63,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "def preprocess_text(document):\n",
        "    # Initialize spaCy resources\n",
        "    nlp = spacy.load(\"en_core_web_sm\")\n",
        "\n",
        "    # Apply spaCy NLP pipeline\n",
        "    doc = nlp(document)\n",
        "\n",
        "    # Lemmatize and remove stopwords\n",
        "    tokens = [token.lemma_.lower() for token in doc if not token.is_stop and token.text not in string.punctuation]\n",
        "\n",
        "    # Join tokens back into a text\n",
        "    cleaned_text = \" \".join(tokens)\n",
        "\n",
        "    return cleaned_text"
      ],
      "metadata": {
        "id": "kiZ49sS5TG8E"
      },
      "execution_count": 64,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# UDF --\n",
        "\n",
        "from pyspark.sql.functions import udf\n",
        "from pyspark.sql.types import StringType"
      ],
      "metadata": {
        "id": "ari7OeNsO_XE"
      },
      "execution_count": 65,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Register the UDF with Spark\n",
        "preprocess_udf_spark = udf(preprocess_text, StringType())"
      ],
      "metadata": {
        "id": "YpRUZEaLPMMF"
      },
      "execution_count": 66,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "df_prep = small_dataset.withColumn(\"preprocessed_text\", preprocess_udf_spark(small_dataset['TEXT']))"
      ],
      "metadata": {
        "id": "znS4y9QVPPT6"
      },
      "execution_count": 67,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "df_prep.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "RHB0Mz9pPdYh",
        "outputId": "f5c2d1cd-97c9-4068-fa36-374b719921c8"
      },
      "execution_count": 68,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+--------------------+------------+--------------------+\n",
            "|                TEXT|      doc_id|   preprocessed_text|\n",
            "+--------------------+------------+--------------------+\n",
            "|a recent outbreak...|      106539|recent outbreak s...|\n",
            "|wefast gradientec...| 25769866227|wefast gradientec...|\n",
            "|this paper deals ...| 42949764250|paper deal patien...|\n",
            "|there is a close ...| 51539638497|close association...|\n",
            "|occasional and un...| 51539642085|occasional unpred...|\n",
            "|the relative perc...| 77309421895|relative percent ...|\n",
            "|mixed alloys for ...|103079248410|mix alloy dental ...|\n",
            "|a high incidence ...|103079293994|high incidence go...|\n",
            "|a housetohouse su...|103079331152|housetohouse surv...|\n",
            "|to characterize t...|120259124985|characterize cont...|\n",
            "|we studied the ef...|120259129263|study efficacy ca...|\n",
            "|the effects of cl...|137439020479|effect clonidine ...|\n",
            "|lipopolysaccharid...|154618911990|lipopolysaccharid...|\n",
            "|this case report ...|154618921437|case report yearo...|\n",
            "|a study of one of...|163208873442|study af influenc...|\n",
            "|rat fetuses at da...|188978671675|rat fetuse day ge...|\n",
            "|the catenins are ...|206158468968|catenin polypepti...|\n",
            "|when tissues are ...|214748392950|tissue process el...|\n",
            "|SS concentrations...|214748459668|ss concentration ...|\n",
            "|the sensitivity o...|223338408901|sensitivity pcd w...|\n",
            "+--------------------+------------+--------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "xuwFLQoOPfTi"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# HASH FAMILY\n",
        "\n",
        "\n",
        "# ci sono degli import che non ho eseguito ma ho potuto comunque creare classe\n",
        "# forse sono già in colab boh\n",
        "\n",
        "import re, hashlib, math, time\n",
        "from random import randint, seed\n",
        "\n",
        "# però sono abbastanza certa che il seed mi serva\n",
        "\n",
        "seed(16)\n",
        "\n",
        "# HO MESSO BYTE RITORNO A 4, PENSO 2^32 BUCKET THEN\n",
        "# BUT WHO KNOWS\n",
        "\n",
        "class hashFamily:\n",
        "    def __init__(self, i):\n",
        "        self.resultSize = 4 # how many bytes we want back\n",
        "        self.maxLen = 20 # how long can our i be (in decimal)\n",
        "        self.salt = str(i).zfill(self.maxLen)[-self.maxLen:]\n",
        "        self.id = i\n",
        "\n",
        "    def get_hash_value(self, el_to_hash):\n",
        "        return int(hashlib.sha1(str(el_to_hash).encode('utf-8') + self.salt.encode('utf-8')).hexdigest()[-self.resultSize:], 16)\n"
      ],
      "metadata": {
        "id": "EkJFwwfdJ8eH"
      },
      "execution_count": 21,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# SHINGLER\n",
        "\n",
        "\n",
        "class shingler:\n",
        "    def __init__(self, k):\n",
        "\n",
        "        if k > 0:\n",
        "            self.k = int(k)\n",
        "        else:\n",
        "            self.k = 10\n",
        "\n",
        "    #inner class utility\n",
        "    #def process_doc(self, document):\n",
        "     #   return re.sub(\"( )+|(\\n)+\",\" \",document).lower()\n",
        "\n",
        "    def get_shingles(self, document):\n",
        "        shingles = set()\n",
        "\n",
        "        #spacy_proc = TextPreprocessor()\n",
        "\n",
        "        document= text_preprocessor_spacy.preprocess(document)\n",
        "\n",
        "        for i in range(0, len(document)-self.k+1 ):\n",
        "            shingles.add(document[i:i+self.k])\n",
        "        return shingles\n",
        "\n",
        "    def get_k(self):\n",
        "        return self.k\n",
        "\n",
        "    #return sorted hash\n",
        "    def get_hashed_shingles(self, shingles_set):\n",
        "        hash_function = hashFamily(0)\n",
        "        return sorted( {hash_function.get_hash_value(s) for s in shingles_set} )"
      ],
      "metadata": {
        "id": "UTK5paiJJ9XW"
      },
      "execution_count": 22,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "def shingling_map(row):\n",
        "    out = []\n",
        "    sh_instance = shingler(10)\n",
        "    hashed_shingles = sh_instance.get_hashed_shingles( sh_instance.get_shingles( row['TEXT'] ) )\n",
        "    signature_size = 50\n",
        "    for i in range(0,signature_size): #signature size\n",
        "        out.append( (row['doc_id'], hashed_shingles, i ) )\n",
        "    #return an iterator to use flatMap => produce more than one key-value pair as output (namely one per hash function)\n",
        "    return iter(out)\n",
        "\n",
        "\n"
      ],
      "metadata": {
        "id": "hkaJ63p3KEe8"
      },
      "execution_count": 23,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_0 = small_dataset.rdd.flatMap(shingling_map)"
      ],
      "metadata": {
        "id": "2wYEefrkKHgV"
      },
      "execution_count": 24,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_0_df = rdd_0.toDF()"
      ],
      "metadata": {
        "id": "pqYYZmQAKLuE"
      },
      "execution_count": 25,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_0_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Ly1UMb4eKMUx",
        "outputId": "347f54d6-39a0-41c7-ca71-a8f08f761fa9"
      },
      "execution_count": 26,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+------+--------------------+---+\n",
            "|    _1|                  _2| _3|\n",
            "+------+--------------------+---+\n",
            "|106539|[76, 297, 337, 51...|  0|\n",
            "|106539|[76, 297, 337, 51...|  1|\n",
            "|106539|[76, 297, 337, 51...|  2|\n",
            "|106539|[76, 297, 337, 51...|  3|\n",
            "|106539|[76, 297, 337, 51...|  4|\n",
            "|106539|[76, 297, 337, 51...|  5|\n",
            "|106539|[76, 297, 337, 51...|  6|\n",
            "|106539|[76, 297, 337, 51...|  7|\n",
            "|106539|[76, 297, 337, 51...|  8|\n",
            "|106539|[76, 297, 337, 51...|  9|\n",
            "|106539|[76, 297, 337, 51...| 10|\n",
            "|106539|[76, 297, 337, 51...| 11|\n",
            "|106539|[76, 297, 337, 51...| 12|\n",
            "|106539|[76, 297, 337, 51...| 13|\n",
            "|106539|[76, 297, 337, 51...| 14|\n",
            "|106539|[76, 297, 337, 51...| 15|\n",
            "|106539|[76, 297, 337, 51...| 16|\n",
            "|106539|[76, 297, 337, 51...| 17|\n",
            "|106539|[76, 297, 337, 51...| 18|\n",
            "|106539|[76, 297, 337, 51...| 19|\n",
            "+------+--------------------+---+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "rdd_0_df.count()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Kyr3VarRKPsf",
        "outputId": "edf80efd-4170-4563-9a43-e5f58d910fae"
      },
      "execution_count": 27,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "4700"
            ]
          },
          "metadata": {},
          "execution_count": 27
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "che è esattamente 94 (sample size) * 50 (sig matrix size) BELLA"
      ],
      "metadata": {
        "id": "1jDn3w9VM98E"
      }
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "kyq-9s2mLzFD"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "## TO DO:\n",
        "- CAPIRE COME E DOVE E QUANDO AGGIUNGERE CALCOLO LUNGHEZZA MEDIA IN CARATTERI DI TUTTI I DOC PER SCELTA OTTIMALE DI K\n",
        "- SCELTA SIZE SIGNATURE MATRIX\n",
        "\n",
        "- SIMILAR PAIRS DETECTION IMPLEMENTAZIONE CORRETTA\n",
        "- SCELTA NUMERO BAND AND ROWS FOR LSH\n",
        "\n",
        "- AMPLIFICATION AND OR???"
      ],
      "metadata": {
        "id": "5yzEwdYXNbsc"
      }
    }
  ]
}