Part 2 - Student Success and Engagement Metrics Calculation.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "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>10</td><td>application_1716494797375_0012</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http://ip-172-31-43-251.ec2.internal:20888/proxy/application_1716494797375_0012/\">Link</a></td><td><a target=\"_blank\" href=\"http://ip-172-31-38-199.ec2.internal:8042/node/containerlogs/container_1716494797375_0012_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, countDistinct, avg, max"
   ]
  },
  {
   "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": 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 calculate_completion_rate(df):\n",
    "    \"\"\" Calculate the course completion rate for each student \"\"\"\n",
    "    return df.groupBy('uuid', 'ucid').agg(\n",
    "        (countDistinct(col('is_correct')) / countDistinct(col('upid'))).alias('completion_rate')\n",
    "    )"
   ]
  },
  {
   "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"
    }
   ],
   "source": [
    "def add_score_column(df):\n",
    "    \"\"\" Add a 'score' column to DataFrame, where True is 1, False is 0 \"\"\"\n",
    "    return df.withColumn('score', col('is_correct').cast('int'))"
   ]
  },
  {
   "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": [
    "def calculate_average_scores(df):\n",
    "    \"\"\" Calculate the average scores for each student in each course \"\"\"\n",
    "    return df.groupBy('uuid', 'ucid').agg(\n",
    "        avg('score').alias('average_score')\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "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 analyze_first_login(df):\n",
    "    \"\"\" Analyze the first login date for each student \"\"\"\n",
    "    return df.withColumn('first_login_date_TW', to_timestamp(col('first_login_date_TW'), 'yyyy-MM-dd')) \\\n",
    "             .groupBy('user_city').agg(\n",
    "                 max('first_login_date_TW').alias('latest_first_login')\n",
    "             )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "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_interactions(df):\n",
    "    \"\"\" Count distinct hints used by each student \"\"\"\n",
    "    return df.groupBy('uuid').agg(\n",
    "        countDistinct('used_hint_cnt').alias('hints_used')\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "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|                ucid|    completion_rate|\n",
      "+--------------------+--------------------+-------------------+\n",
      "|vWgIeDzvzqwgaktmo...|ad/P0gKMpXTOaZ/Sh...| 0.3333333333333333|\n",
      "|8pdEjpATA3PsMhjIW...|Fj4qR65Oj+FrvKsCq...|0.09090909090909091|\n",
      "|BNTYlBA3WKuWYunH8...|0jLx3N/h3zT921WmS...|0.09090909090909091|\n",
      "|l+ygbw9Dm9Ytfp61B...|Iq4bn747ApUwO7SvD...| 0.1111111111111111|\n",
      "|ZOx27o/qLMk2l4H+0...|cvnGl+iUiKphfjbPc...|0.14285714285714285|\n",
      "|5exF/tOQXBjuOKvzh...|KzqwnxUK+iRBndR+l...|               0.25|\n",
      "|gb9z96Tip7MciEXAw...|w74XdHKzME2SVRy/4...|               0.25|\n",
      "|REoRF8OmojzA+MwEZ...|BMWFDTPQo2WDehX7l...|0.14285714285714285|\n",
      "|qE8Hag7q5i9BYE7n8...|hVo+IVNVaOniJLhdg...|                0.2|\n",
      "|ZosxtIsY4WlWBjUeU...|Iq4bn747ApUwO7SvD...|                0.2|\n",
      "+--------------------+--------------------+-------------------+\n",
      "only showing top 10 rows\n",
      "\n",
      "+--------------------+--------------------+-------------+\n",
      "|                uuid|                ucid|average_score|\n",
      "+--------------------+--------------------+-------------+\n",
      "|sAqTmfm48t9B+X1be...|2dZAIlKn+MzyIa5L8...|          1.0|\n",
      "|7zAjIWMYdQ2kzbkgs...|hv7kHCAIdj7thZUml...|          1.0|\n",
      "|K6LAx91Pm2+FUHUOb...|tOR47i7wGbuHkGxkh...|          1.0|\n",
      "|XvRkqD49KL2JCt/o8...|jImuGvSym6odgRoYw...|          1.0|\n",
      "|hiWs697xdf5b+GVm2...|SoyS9lUIh/q7oCg6X...|          1.0|\n",
      "|+WnIGdKGsa4pYa4Ah...|fuM14IacAZTrOR0wL...|          1.0|\n",
      "|Dt5BIIhgB0ucuGIrD...|Wk1PobqnhB1FsjjiU...|          1.0|\n",
      "|8yrpWerP5PwUTRUjZ...|ZOO50WfENqtWGe4uB...|          1.0|\n",
      "|gLii7LzaFI+uKd95u...|aaqqTl3AwAAh2Zu0f...|          1.0|\n",
      "|scgtuHl5stOlsU30c...|GogTiwoGv+VkAzLYe...|          1.0|\n",
      "+--------------------+--------------------+-------------+\n",
      "only showing top 10 rows\n",
      "\n",
      "+---------+-------------------+\n",
      "|user_city| latest_first_login|\n",
      "+---------+-------------------+\n",
      "|       ty|2019-07-31 00:00:00|\n",
      "|      hlc|2019-07-28 00:00:00|\n",
      "|       kh|2019-07-31 00:00:00|\n",
      "|       km|2019-07-08 00:00:00|\n",
      "|       hc|2019-07-29 00:00:00|\n",
      "|       ml|2019-07-29 00:00:00|\n",
      "|       cy|2019-07-30 00:00:00|\n",
      "|      ylc|2019-07-31 00:00:00|\n",
      "|      chc|2019-07-30 00:00:00|\n",
      "|      ilc|2019-07-31 00:00:00|\n",
      "+---------+-------------------+\n",
      "only showing top 10 rows\n",
      "\n",
      "+--------------------+----------+\n",
      "|                uuid|hints_used|\n",
      "+--------------------+----------+\n",
      "|jSAalZ78mzgLy/5j/...|         1|\n",
      "|DKn2f611ZxvLpmJ1D...|         1|\n",
      "|nvHP5Ukz4LZnbsefM...|         1|\n",
      "|XzjV7d232DDbc3fwZ...|         1|\n",
      "|YBNvIhO4GkHGpkqgq...|         1|\n",
      "|1FmQp/R/gJIyHOozM...|         1|\n",
      "|ITPYHxkECD2MhcBwG...|         1|\n",
      "|1faxUZS2oLddJoAdg...|         1|\n",
      "|FUONqME8YydtnaA/S...|         1|\n",
      "|zy3wxEvezQRoVzHYV...|         1|\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",
    "# Load processed data from S3\n",
    "processed_df = spark.read.parquet(\"s3://kz-final-project/processed_data/\")\n",
    "processed_df = add_score_column(processed_df)  # Prepare the data frame with scores\n",
    "\n",
    "# Calculate metrics\n",
    "course_completion = calculate_completion_rate(processed_df)\n",
    "average_scores = calculate_average_scores(processed_df)\n",
    "first_login_analysis = analyze_first_login(info_user_data_df)\n",
    "interactions = count_interactions(processed_df)\n",
    "\n",
    "# Display the results\n",
    "course_completion.show(10)\n",
    "average_scores.show(10)\n",
    "first_login_analysis.show(10)\n",
    "interactions.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "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.mode('overwrite').parquet(\"s3://kz-final-project/processed_data2/\")"
   ]
  }
 ],
 "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
}