Segmentation.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6QCe3ROKfj5L",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "b0cca7d3-9a6f-4202-a792-60a10dbbaf4e"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting pyspark\n",
            "  Downloading pyspark-3.5.0.tar.gz (316.9 MB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m316.9/316.9 MB\u001b[0m \u001b[31m4.3 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25h  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Requirement already satisfied: py4j==0.10.9.7 in /usr/local/lib/python3.10/dist-packages (from pyspark) (0.10.9.7)\n",
            "Building wheels for collected packages: pyspark\n",
            "  Building wheel for pyspark (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for pyspark: filename=pyspark-3.5.0-py2.py3-none-any.whl size=317425344 sha256=a09d6dbc19e6ceb4718b9e42e825517ee3d33c6fbabadd998c83cc5aaa13e6fc\n",
            "  Stored in directory: /root/.cache/pip/wheels/41/4e/10/c2cf2467f71c678cfc8a6b9ac9241e5e44a01940da8fbb17fc\n",
            "Successfully built pyspark\n",
            "Installing collected packages: pyspark\n",
            "Successfully installed pyspark-3.5.0\n",
            "Requirement already satisfied: openpyxl in /usr/local/lib/python3.10/dist-packages (3.1.2)\n",
            "Requirement already satisfied: et-xmlfile in /usr/local/lib/python3.10/dist-packages (from openpyxl) (1.1.0)\n"
          ]
        }
      ],
      "source": [
        "\n",
        "!pip install -q findspark\n",
        "!pip install pyspark\n",
        "!pip install openpyxl\n"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import sum, col\n",
        "# Create a Spark session\n",
        "spark = SparkSession.builder.appName(\"DataProcessingExample\").getOrCreate()\n",
        "\n",
        "# Load the dataset (replace \"your_data.csv\" with your actual file path)\n",
        "df = spark.read.csv(\"cleaned_data.csv\", header=True, inferSchema=True)\n",
        "\n",
        "# Sample 10% of the data\n",
        "sampled_df = df.sample(fraction=0.1, seed=42)"
      ],
      "metadata": {
        "id": "uC_1pqxsgE63",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 390
        },
        "outputId": "13bd871c-e8f8-41d7-fc1d-e6bd89c499f8"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "error",
          "ename": "AnalysisException",
          "evalue": "ignored",
          "traceback": [
            "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
            "\u001b[0;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
            "\u001b[0;32m<ipython-input-3-7b2de2ea13c2>\u001b[0m in \u001b[0;36m<cell line: 7>\u001b[0;34m()\u001b[0m\n\u001b[1;32m      5\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      6\u001b[0m \u001b[0;31m# Load the dataset (replace \"your_data.csv\" with your actual file path)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 7\u001b[0;31m \u001b[0mdf\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mspark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mread\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcsv\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"cleaned_data.csv\"\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mheader\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0minferSchema\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\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      8\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      9\u001b[0m \u001b[0;31m# Sample 10% of the data\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/usr/local/lib/python3.10/dist-packages/pyspark/sql/readwriter.py\u001b[0m in \u001b[0;36mcsv\u001b[0;34m(self, path, schema, sep, encoding, quote, escape, comment, header, inferSchema, ignoreLeadingWhiteSpace, ignoreTrailingWhiteSpace, nullValue, nanValue, positiveInf, negativeInf, dateFormat, timestampFormat, maxColumns, maxCharsPerColumn, maxMalformedLogPerPartition, mode, columnNameOfCorruptRecord, multiLine, charToEscapeQuoteEscaping, samplingRatio, enforceSchema, emptyValue, locale, lineSep, pathGlobFilter, recursiveFileLookup, modifiedBefore, modifiedAfter, unescapedQuoteHandling)\u001b[0m\n\u001b[1;32m    738\u001b[0m         \u001b[0;32mif\u001b[0m \u001b[0mtype\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mpath\u001b[0m\u001b[0;34m)\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0mlist\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    739\u001b[0m             \u001b[0;32massert\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_spark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jvm\u001b[0m \u001b[0;32mis\u001b[0m \u001b[0;32mnot\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 740\u001b[0;31m             \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_df\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jreader\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcsv\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_spark\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_sc\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_jvm\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPythonUtils\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtoSeq\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mpath\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[0m\u001b[1;32m    741\u001b[0m         \u001b[0;32melif\u001b[0m \u001b[0misinstance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mpath\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[1;32m    742\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/usr/local/lib/python3.10/dist-packages/py4j/java_gateway.py\u001b[0m in \u001b[0;36m__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1320\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m   1321\u001b[0m         \u001b[0manswer\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mgateway_client\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msend_command\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mcommand\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1322\u001b[0;31m         return_value = get_return_value(\n\u001b[0m\u001b[1;32m   1323\u001b[0m             answer, self.gateway_client, self.target_id, self.name)\n\u001b[1;32m   1324\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;32m/usr/local/lib/python3.10/dist-packages/pyspark/errors/exceptions/captured.py\u001b[0m in \u001b[0;36mdeco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    183\u001b[0m                 \u001b[0;31m# Hide where the exception came from that shows a non-Pythonic\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    184\u001b[0m                 \u001b[0;31m# JVM exception message.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 185\u001b[0;31m                 \u001b[0;32mraise\u001b[0m \u001b[0mconverted\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0;32mNone\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    186\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    187\u001b[0m                 \u001b[0;32mraise\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
            "\u001b[0;31mAnalysisException\u001b[0m: [PATH_NOT_FOUND] Path does not exist: file:/content/cleaned_data.csv."
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "sampled_df.show()\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "k8mLom5hnYd8",
        "outputId": "9c685aeb-bf8e-4b5c-d75c-744efb06a198"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+---------+--------+-----------+-----------+---------+----------+--------------+\n",
            "|StockCode|Quantity|InvoiceDate|InvoiceTime|UnitPrice|CustomerID|       Country|\n",
            "+---------+--------+-----------+-----------+---------+----------+--------------+\n",
            "|    22633|       6| 2010/12/01|08:28:00 AM|     1.85|     17850|United Kingdom|\n",
            "|    22622|       2| 2010/12/01|08:34:00 AM|     9.95|     13047|United Kingdom|\n",
            "|    21755|       3| 2010/12/01|08:34:00 AM|     5.95|     13047|United Kingdom|\n",
            "|    71053|       6| 2010/12/01|09:02:00 AM|     3.39|     17850|United Kingdom|\n",
            "|    82486|       4| 2010/12/01|09:02:00 AM|     6.95|     17850|United Kingdom|\n",
            "|    22752|       2| 2010/12/01|09:02:00 AM|     7.65|     17850|United Kingdom|\n",
            "|   85123A|       6| 2010/12/01|09:32:00 AM|     2.55|     17850|United Kingdom|\n",
            "|    21071|       6| 2010/12/01|09:32:00 AM|     1.06|     17850|United Kingdom|\n",
            "|   84029E|       6| 2010/12/01|09:32:00 AM|     3.39|     17850|United Kingdom|\n",
            "|    21033|      10| 2010/12/01|09:37:00 AM|     2.95|     14688|United Kingdom|\n",
            "|   84997B|      12| 2010/12/01|09:37:00 AM|     3.75|     14688|United Kingdom|\n",
            "|    21931|      10| 2010/12/01|09:37:00 AM|     1.95|     14688|United Kingdom|\n",
            "|    22961|      24| 2010/12/01|09:41:00 AM|     1.45|     17809|United Kingdom|\n",
            "|    22139|      23| 2010/12/01|09:41:00 AM|     4.25|     15311|United Kingdom|\n",
            "|    82567|       2| 2010/12/01|09:41:00 AM|      2.1|     15311|United Kingdom|\n",
            "|    22086|       4| 2010/12/01|09:41:00 AM|     2.95|     15311|United Kingdom|\n",
            "|    84832|       1| 2010/12/01|09:41:00 AM|     0.85|     15311|United Kingdom|\n",
            "|    22778|       3| 2010/12/01|09:41:00 AM|     3.95|     15311|United Kingdom|\n",
            "|    10002|      12| 2010/12/01|09:45:00 AM|     0.85|     16098|United Kingdom|\n",
            "|   85099B|     100| 2010/12/01|09:57:00 AM|     1.65|     16029|United Kingdom|\n",
            "+---------+--------+-----------+-----------+---------+----------+--------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Select only the relevant columns\n",
        "relevant_columns = ['CustomerID', 'StockCode', 'Quantity', 'UnitPrice']\n",
        "df = df.select(relevant_columns)\n"
      ],
      "metadata": {
        "id": "Ag5JOrfDgrmY"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Print the sampled DataFrame\n",
        "df.show()\n"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "yKbag8J2jm3m",
        "outputId": "33dbb7aa-fb2d-4d3c-f355-070906758f6a"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+---------+--------+---------+\n",
            "|CustomerID|StockCode|Quantity|UnitPrice|\n",
            "+----------+---------+--------+---------+\n",
            "|     17850|   85123A|       6|     2.55|\n",
            "|     17850|    71053|       6|     3.39|\n",
            "|     17850|   84406B|       8|     2.75|\n",
            "|     17850|   84029G|       6|     3.39|\n",
            "|     17850|   84029E|       6|     3.39|\n",
            "|     17850|    22752|       2|     7.65|\n",
            "|     17850|    21730|       6|     4.25|\n",
            "|     17850|    22633|       6|     1.85|\n",
            "|     17850|    22632|       6|     1.85|\n",
            "|     13047|    84879|      32|     1.69|\n",
            "|     13047|    22745|       6|      2.1|\n",
            "|     13047|    22748|       6|      2.1|\n",
            "|     13047|    22749|       8|     3.75|\n",
            "|     13047|    22310|       6|     1.65|\n",
            "|     13047|    84969|       6|     4.25|\n",
            "|     13047|    22623|       3|     4.95|\n",
            "|     13047|    22622|       2|     9.95|\n",
            "|     13047|    21754|       3|     5.95|\n",
            "|     13047|    21755|       3|     5.95|\n",
            "|     13047|    21777|       4|     7.95|\n",
            "+----------+---------+--------+---------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "df = df.withColumn('Quantity', col('Quantity').cast('int'))"
      ],
      "metadata": {
        "id": "5x6XGv0uzYaj"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "# Group by 'CustomerID' and 'StockCode', then aggregate the sum of 'Quantity' and 'UnitPrice'\n",
        "aggregated_df = df.groupBy('CustomerID', 'StockCode').agg(\n",
        "    (sum('Quantity') * sum('UnitPrice')).alias('TotalQuantityTimesSales')\n",
        ")\n"
      ],
      "metadata": {
        "id": "TMS9aOSNgsbi"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "aggregated_df.show()"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "sPP0clZOz0me",
        "outputId": "1fd537f6-55b1-4dea-a0f5-1d2970b88825"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "+----------+---------+-----------------------+\n",
            "|CustomerID|StockCode|TotalQuantityTimesSales|\n",
            "+----------+---------+-----------------------+\n",
            "|     18074|    22464|     19.799999999999997|\n",
            "|     17908|    20966|                    2.5|\n",
            "|     17908|   85099F|                   1.95|\n",
            "|     17920|    22141|     25.200000000000003|\n",
            "|     17968|    22500|                    9.9|\n",
            "|     17897|    22909|                   0.85|\n",
            "|     14729|    22582|                    5.1|\n",
            "|     15012|    21041|                   2.95|\n",
            "|     12433|    22665|     405.59999999999997|\n",
            "|     15922|    22183|                   40.5|\n",
            "|     17346|    22895|                   11.8|\n",
            "|     17841|    22941|                  238.0|\n",
            "|     13093|    20685|                  405.0|\n",
            "|     14741|    22457|      619.5000000000001|\n",
            "|     15658|    21756|                  17.85|\n",
            "|     14060|    21014|                   51.0|\n",
            "|     15363|    22382|                   16.5|\n",
            "|     17964|    22745|                    2.1|\n",
            "|     17235|   85184C|     35.400000000000006|\n",
            "|     17976|    84347|                   2.55|\n",
            "+----------+---------+-----------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Save DataFrame to CSV format\n",
        "aggregated_df.write.csv(\"final1.csv\", header=True, mode=\"overwrite\")\n"
      ],
      "metadata": {
        "id": "RYnUsmF9gyho"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}