Part 1 - Data Exploration and Preprocessing.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Starting Spark application\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>9</td><td>application_1716494797375_0011</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-43-251.ec2.internal:20888/proxy/application_1716494797375_0011/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-46-189.ec2.internal:8042/node/containerlogs/container_1716494797375_0011_01_000001/livy\">Link</a></td><td>✔</td></tr></table>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SparkSession available as 'spark'.\n"
     ]
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, to_timestamp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Create or retrieve a Spark session\n",
    "spark = SparkSession.builder \\\n",
    "    .appName(\"Online Education Analysis\") \\\n",
    "    .config(\"spark.executor.memory\", \"4g\") \\\n",
    "    .config(\"spark.driver.memory\", \"4g\") \\\n",
    "    .config(\"spark.pyspark.python\", \"python3\") \\\n",
    "    .config(\"spark.pyspark.virtualenv.enabled\", \"true\") \\\n",
    "    .config(\"spark.pyspark.virtualenv.type\", \"native\") \\\n",
    "    .config(\"spark.pyspark.virtualenv.bin.path\", \"/usr/bin/virtualenv\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def load_data(file_name):\n",
    "    \"\"\" Load data from CSV file on S3 using Spark \"\"\"\n",
    "    s3_path = f\"s3a://kz-final-project/{file_name}\"\n",
    "    return spark.read.csv(s3_path, header=True, inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def preprocess_data(df):\n",
    "    \"\"\" Preprocess data by converting timestamps and filtering records \"\"\"\n",
    "    df = df.withColumn('timestamp_TW', to_timestamp(col('timestamp_TW')))\n",
    "    return df.filter(col('is_correct') == True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "def count_user_attempts(df):\n",
    "    \"\"\" Count the number of attempts per user \"\"\"\n",
    "    return df.groupBy('uuid').count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|                uuid|count|\n",
      "+--------------------+-----+\n",
      "|XzjV7d232DDbc3fwZ...| 1381|\n",
      "|mSxpLrWH1dwKumKuP...|   82|\n",
      "|e+ritEYXMn9OE97kj...|  610|\n",
      "|nVR2dN2v+4pNErS9k...|  144|\n",
      "|zr+bafKBCxuS4QqLC...|  415|\n",
      "|0r0JQhzLmXSnXQy/O...|  397|\n",
      "|i+5HEweslnkJx/DFW...|  962|\n",
      "|cXpVxuFVbcITI4mcw...|  751|\n",
      "|b8z8nmT8NPGkv27oP...|  136|\n",
      "|SVurddKEOBdHB2R6W...|   29|\n",
      "+--------------------+-----+\n",
      "only showing top 10 rows"
     ]
    }
   ],
   "source": [
    "# Load data\n",
    "info_content_df = load_data('Info_Content.csv')\n",
    "info_user_data_df = load_data('Info_UserData.csv')\n",
    "log_problem_df = load_data('Log_Problem.csv')\n",
    "\n",
    "# Process data\n",
    "processed_df = preprocess_data(log_problem_df)\n",
    "\n",
    "# Analyze data\n",
    "user_attempts = count_user_attempts(processed_df)\n",
    "user_attempts.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Save processed data to S3\n",
    "processed_df.write.parquet(\"s3://kz-final-project/processed_data/\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 3
   },
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}