pyspark-intro.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "# from pyspark.sql.function import col, sum, when, count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 1: Initialize Spark session\n",
    "spark = SparkSession.builder \\\n",
    "    .appName(\"Sales ETL Pipeline\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "<>:3: SyntaxWarning: invalid escape sequence '\\d'\n",
      "<>:3: SyntaxWarning: invalid escape sequence '\\d'\n",
      "C:\\Users\\User\\AppData\\Local\\Temp\\ipykernel_876\\1932672822.py:3: SyntaxWarning: invalid escape sequence '\\d'\n",
      "  df = pd.read_excel(\"F:\\data-science\\datasets\\online_retail_II.xlsx\")\n",
      "C:\\Users\\User\\AppData\\Local\\Temp\\ipykernel_876\\1932672822.py:3: SyntaxWarning: invalid escape sequence '\\d'\n",
      "  df = pd.read_excel(\"F:\\data-science\\datasets\\online_retail_II.xlsx\")\n"
     ]
    },
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'distutils'",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mModuleNotFoundError\u001b[0m                       Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[11], line 4\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[38;5;28;01mimport\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;21;01mpandas\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mas\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;21;01mpd\u001b[39;00m\n\u001b[0;32m      3\u001b[0m df \u001b[38;5;241m=\u001b[39m pd\u001b[38;5;241m.\u001b[39mread_excel(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mF:\u001b[39m\u001b[38;5;124m\\\u001b[39m\u001b[38;5;124mdata-science\u001b[39m\u001b[38;5;124m\\\u001b[39m\u001b[38;5;124mdatasets\u001b[39m\u001b[38;5;124m\\\u001b[39m\u001b[38;5;124monline_retail_II.xlsx\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m----> 4\u001b[0m sales_df \u001b[38;5;241m=\u001b[39m \u001b[43mspark\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcreateDataFrame\u001b[49m\u001b[43m(\u001b[49m\u001b[43mdf\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m      5\u001b[0m sales_df\u001b[38;5;241m.\u001b[39mshow(\u001b[38;5;241m5\u001b[39m)\n",
      "File \u001b[1;32mf:\\data-science\\.venv\\Lib\\site-packages\\pyspark\\sql\\session.py:1440\u001b[0m, in \u001b[0;36mSparkSession.createDataFrame\u001b[1;34m(self, data, schema, samplingRatio, verifySchema)\u001b[0m\n\u001b[0;32m   1436\u001b[0m     data \u001b[38;5;241m=\u001b[39m pd\u001b[38;5;241m.\u001b[39mDataFrame(data, columns\u001b[38;5;241m=\u001b[39mcolumn_names)\n\u001b[0;32m   1438\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m has_pandas \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(data, pd\u001b[38;5;241m.\u001b[39mDataFrame):\n\u001b[0;32m   1439\u001b[0m     \u001b[38;5;66;03m# Create a DataFrame from pandas DataFrame.\u001b[39;00m\n\u001b[1;32m-> 1440\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43msuper\u001b[39;49m\u001b[43m(\u001b[49m\u001b[43mSparkSession\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[43m)\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcreateDataFrame\u001b[49m\u001b[43m(\u001b[49m\u001b[43m  \u001b[49m\u001b[38;5;66;43;03m# type: ignore[call-overload]\u001b[39;49;00m\n\u001b[0;32m   1441\u001b[0m \u001b[43m        \u001b[49m\u001b[43mdata\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mschema\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43msamplingRatio\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[43mverifySchema\u001b[49m\n\u001b[0;32m   1442\u001b[0m \u001b[43m    \u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m   1443\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_create_dataframe(\n\u001b[0;32m   1444\u001b[0m     data, schema, samplingRatio, verifySchema  \u001b[38;5;66;03m# type: ignore[arg-type]\u001b[39;00m\n\u001b[0;32m   1445\u001b[0m )\n",
      "File \u001b[1;32mf:\\data-science\\.venv\\Lib\\site-packages\\pyspark\\sql\\pandas\\conversion.py:330\u001b[0m, in \u001b[0;36mSparkConversionMixin.createDataFrame\u001b[1;34m(self, data, schema, samplingRatio, verifySchema)\u001b[0m\n\u001b[0;32m    326\u001b[0m \u001b[38;5;28;01massert\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(\u001b[38;5;28mself\u001b[39m, SparkSession)\n\u001b[0;32m    328\u001b[0m \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;21;01mpyspark\u001b[39;00m\u001b[38;5;21;01m.\u001b[39;00m\u001b[38;5;21;01msql\u001b[39;00m\u001b[38;5;21;01m.\u001b[39;00m\u001b[38;5;21;01mpandas\u001b[39;00m\u001b[38;5;21;01m.\u001b[39;00m\u001b[38;5;21;01mutils\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mimport\u001b[39;00m require_minimum_pandas_version\n\u001b[1;32m--> 330\u001b[0m \u001b[43mrequire_minimum_pandas_version\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m    332\u001b[0m timezone \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_jconf\u001b[38;5;241m.\u001b[39msessionLocalTimeZone()\n\u001b[0;32m    334\u001b[0m \u001b[38;5;66;03m# If no schema supplied by user then get the names of columns only\u001b[39;00m\n",
      "File \u001b[1;32mf:\\data-science\\.venv\\Lib\\site-packages\\pyspark\\sql\\pandas\\utils.py:24\u001b[0m, in \u001b[0;36mrequire_minimum_pandas_version\u001b[1;34m()\u001b[0m\n\u001b[0;32m     21\u001b[0m \u001b[38;5;66;03m# TODO(HyukjinKwon): Relocate and deduplicate the version specification.\u001b[39;00m\n\u001b[0;32m     22\u001b[0m minimum_pandas_version \u001b[38;5;241m=\u001b[39m \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124m1.0.5\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[1;32m---> 24\u001b[0m \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;21;01mdistutils\u001b[39;00m\u001b[38;5;21;01m.\u001b[39;00m\u001b[38;5;21;01mversion\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mimport\u001b[39;00m LooseVersion\n\u001b[0;32m     26\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m     27\u001b[0m     \u001b[38;5;28;01mimport\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;21;01mpandas\u001b[39;00m\n",
      "\u001b[1;31mModuleNotFoundError\u001b[0m: No module named 'distutils'"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "\n",
    "df = pd.read_excel(\"F:\\data-science\\datasets\\online_retail_II.xlsx\")\n",
    "sales_df = spark.createDataFrame(df)\n",
    "sales_df.show(5)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "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.12.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}