Edit_Distance_Spark.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "source": [
        "# Step1: Install Librarys"
      ],
      "metadata": {
        "id": "Q1Wl1RbS0cqf"
      }
    },
    {
      "cell_type": "code",
      "execution_count": 1,
      "metadata": {
        "id": "7bfdMBkMz-6j",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "ceb23ef4-af79-4a41-bbc2-fb1521efbd0a"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Requirement already satisfied: pyspark in /usr/local/lib/python3.10/dist-packages (3.5.1)\n",
            "Requirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from pyspark) (0.10.9.7)\n"
          ]
        }
      ],
      "source": [
        "!pip install pyspark"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install wikipedia"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "Jssq_bipeftx",
        "outputId": "9851b131-0312-4810-aa8f-4f158ecb69a0"
      },
      "execution_count": 2,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Requirement already satisfied: wikipedia in /usr/local/lib/python3.10/dist-packages (1.4.0)\n",
            "Requirement already satisfied: beautifulsoup4 in /usr/local/lib/python3.10/dist-packages (from wikipedia) (4.12.3)\n",
            "Requirement already satisfied: requests<3.0.0,>=2.0.0 in /usr/local/lib/python3.10/dist-packages (from wikipedia) (2.31.0)\n",
            "Requirement already satisfied: charset-normalizer<4,>=2 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.0.0->wikipedia) (3.3.2)\n",
            "Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.0.0->wikipedia) (3.7)\n",
            "Requirement already satisfied: urllib3<3,>=1.21.1 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.0.0->wikipedia) (2.0.7)\n",
            "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/dist-packages (from requests<3.0.0,>=2.0.0->wikipedia) (2024.2.2)\n",
            "Requirement already satisfied: soupsieve>1.2 in /usr/local/lib/python3.10/dist-packages (from beautifulsoup4->wikipedia) (2.5)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Step2: Import Librarys"
      ],
      "metadata": {
        "id": "XKpaOtw7INZt"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import pyspark\n",
        "import wikipedia\n",
        "import numpy as np\n",
        "import re\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.window import Window\n",
        "from pyspark.sql.functions import col, row_number\n",
        "# import logging\n",
        "# logging.basicConfig(level=logging.DEBUG)\n"
      ],
      "metadata": {
        "id": "xP75d6QJIQhW"
      },
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Step3: Download Dataset"
      ],
      "metadata": {
        "id": "xY_a66k1eS-w"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "dataset = wikipedia.page('Python (programming language)').content\n",
        "dataset2 = [\"to be or not to be this is the problem\"]"
      ],
      "metadata": {
        "id": "JMuZD8KteWum"
      },
      "execution_count": 11,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Step4: Preprocessing Dataset"
      ],
      "metadata": {
        "id": "MdOjKnXier6H"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "dataset = re.sub(r'[\\n\\t]', ' ', dataset)\n",
        "dataset = re.sub(r'\\s+', ' ', dataset.strip())\n",
        "\n",
        "dataset = dataset.split()\n",
        "dataset = list(set(dataset))"
      ],
      "metadata": {
        "id": "DbP3IhT4evc-"
      },
      "execution_count": 12,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Step5: Config Spark"
      ],
      "metadata": {
        "id": "7nBdHHqTbu1x"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# sc = pyspark.SparkContext('local[*]')\n",
        "spark = SparkSession.builder.appName(\"EditDistanceApp\").getOrCreate()"
      ],
      "metadata": {
        "id": "Q9b5_Qt3byig"
      },
      "execution_count": 6,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Step6: Edit Distance method"
      ],
      "metadata": {
        "id": "kLgX6VJQIQ6O"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def edit_distance(word1, word2):\n",
        "    num_rows = len(word1)\n",
        "    num_cols = len(word2)\n",
        "\n",
        "    matrix = np.empty((num_rows + 2, num_cols + 2), dtype=object)\n",
        "\n",
        "    for i in range(num_rows + 2):\n",
        "        for j in range(num_cols + 2):\n",
        "            if i == 0 and j == 0:\n",
        "                matrix[i, j] = '#'\n",
        "            elif i == 0 and j == 1:\n",
        "                matrix[i, j] = '_'\n",
        "            elif i == 1 and j == 0:\n",
        "                matrix[i, j] = '_'\n",
        "            elif i == 0:\n",
        "                matrix[i, j] = word2[j - 2] # row word\n",
        "            elif j == 0:\n",
        "                matrix[i, j] = word1[i - 2] # col word\n",
        "            elif i == 1:\n",
        "                matrix[i, j] = str(j - 1)\n",
        "            elif i != 1 and j == 1:\n",
        "                matrix[i, j] = str(i - 1)\n",
        "            else:\n",
        "                first = (int(matrix[i-1, j])+1)\n",
        "                second = (int(matrix[i, j-1])+1)\n",
        "                if matrix[i, 0] == matrix[0, j]:\n",
        "                    third = (int(matrix[i-1, j-1]))\n",
        "                else :\n",
        "                    third = (int(matrix[i-1, j-1])+2)\n",
        "                matrix[i, j] = str(min(first, second, third))\n",
        "    return matrix[num_rows + 1, num_cols + 1]\n",
        "\n",
        "print(edit_distance(\"INTENTION\", \"EXECUTION\"))\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "FYCs_NQEIYAH",
        "outputId": "0a966118-9909-4e70-a27c-4478e7a2644c"
      },
      "execution_count": 7,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "8\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Step7: Calculate Edit distance for all words"
      ],
      "metadata": {
        "id": "fmZYnay1d_bu"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "dataset2 = [\"Ahmad\", \"Asghar\", \"Akbar\", \"Ehsan\", \"Mohsen\"]\n",
        "\n",
        "rdd = spark.sparkContext.parallelize(dataset)\n",
        "\n",
        "word_pairs = rdd.cartesian(rdd).filter(lambda x: x[0]!= x[1])\\\n",
        "                               .map(lambda x: (x[0], x[1], edit_distance(x[0], x[1])))\n",
        "\n",
        "df = word_pairs.toDF([\"Word1\", \"Word2\", \"EditDistance\"])\n",
        "\n",
        "# df = df.toPandas()\n",
        "\n",
        "window_spec = Window.partitionBy(\"Word1\").orderBy(\"EditDistance\")\n",
        "\n",
        "df_with_row_num = df.withColumn(\"row_num\", row_number().over(window_spec))\n",
        "\n",
        "result_df = df_with_row_num.filter(col(\"row_num\") == 1).drop(\"row_num\")\n",
        "\n",
        "result_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "1fEdjLzIvpQR",
        "outputId": "1a6d33f2-8bfe-47a2-830f-6ad95bb7d202"
      },
      "execution_count": 14,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+-----------------+----------+------------+\n",
            "|            Word1|     Word2|EditDistance|\n",
            "+-----------------+----------+------------+\n",
            "|              \"2\"| \"spam={0}|          10|\n",
            "|        \"2.7.18+\"|         ?|          10|\n",
            "|            \"22\".|     state|          10|\n",
            "|      \"@-quoting\"|       ten|          10|\n",
            "|              \"AI|        AI|           1|\n",
            "|\"BDFL-emeritus\").|      emit|          13|\n",
            "|          \"Hello,|   largely|          10|\n",
            "|               \"I|         I|           1|\n",
            "|          \"PyAIML|       'u'|          10|\n",
            "|          \"Python|    Python|           1|\n",
            "|       \"Pythonic\"|        of|          10|\n",
            "|   \"Pythonistas\".|  CPython,|          10|\n",
            "|             \"The|       The|           1|\n",
            "|              \"To| \"spam={0}|          10|\n",
            "|         \"adding\"|      e.g.|          10|\n",
            "|              \"as|        as|           1|\n",
            "|      \"backported|backported|           1|\n",
            "|           \"bar\".|    blocks|          10|\n",
            "|       \"batteries|    Server|          10|\n",
            "|      \"benevolent|   violate|          10|\n",
            "+-----------------+----------+------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    }
  ]
}