13 - Big Data Fundamentals with PySpark.ipynb

{
 "cells": [
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "1 - Introduction to Big Data analysis with Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Understanding SparkContext\n",
    "\n",
    "\n",
    "# Print the version of SparkContext\n",
    "print(\"The version of Spark Context in the PySpark shell is\", sc.version)\n",
    "\n",
    "# Print the Python version of SparkContext\n",
    "print(\"The Python version of Spark Context in the PySpark shell is\", sc.pythonVer)\n",
    "\n",
    "# Print the master of SparkContext\n",
    "print(\"The master of Spark Context in the PySpark shell is\", sc.master)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Interactive Use of PySpark\n",
    "\n",
    "# Create a python list of numbers from 1 to 100 \n",
    "numb = range(1, 100)\n",
    "\n",
    "# Load the list into PySpark  \n",
    "spark_data = sc.parallelize(numb)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Loading data in PySpark shell\n",
    "\n",
    "# Load a local file into PySpark shell\n",
    "lines = sc.textFile(file_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Use of lambda() with map()\n",
    "\n",
    "# Print my_list in the console\n",
    "print(\"Input list is\", my_list)\n",
    "\n",
    "# Square all numbers in my_list\n",
    "squared_list_lambda = list(map(lambda x: x**2, my_list))\n",
    "\n",
    "# Print the result of the map function\n",
    "print(\"The squared numbers are\", squared_list_lambda)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Use of lambda() with filter()\n",
    "\n",
    "# Print my_list2 in the console\n",
    "print(\"Input list is:\", my_list2)\n",
    "\n",
    "# Filter numbers divisible by 10\n",
    "filtered_list = list(filter(lambda x: (x%10 == 0), my_list2))\n",
    "\n",
    "# Print the numbers divisible by 10\n",
    "print(\"Numbers divisible by 10 are:\", filtered_list)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "2 - Programming in PySpark RDD’s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## RDDs from Parallelized collections\n",
    "\n",
    "# Create an RDD from a list of words\n",
    "RDD = sc.parallelize([\"Spark\", \"is\", \"a\", \"framework\", \"for\", \"Big Data processing\"])\n",
    "\n",
    "# Print out the type of the created object\n",
    "print(\"The type of RDD is\", type(RDD))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## RDDs from External Datasets\n",
    "\n",
    "# Print the file_path\n",
    "print(\"The file_path is\", sc.textFile(file_path))\n",
    "\n",
    "# Create a fileRDD from file_path\n",
    "fileRDD = sc.parallelize(file_path)\n",
    "\n",
    "# Check the type of fileRDD\n",
    "print(\"The file type of fileRDD is\", type(fileRDD))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Partitions in your data\n",
    "\n",
    "# Check the number of partitions in fileRDD\n",
    "print(\"Number of partitions in fileRDD is\", fileRDD.getNumPartitions())\n",
    "\n",
    "# Create a fileRDD_part from file_path with 5 partitions\n",
    "fileRDD_part = sc.textFile(file_path, minPartitions = 5)\n",
    "\n",
    "# Check the number of partitions in fileRDD_part\n",
    "print(\"Number of partitions in fileRDD_part is\", fileRDD_part.getNumPartitions())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Map and Collect\n",
    "\n",
    "# Create map() transformation to cube numbers\n",
    "cubedRDD = numbRDD.map(lambda x: x **3)\n",
    "\n",
    "# Collect the results\n",
    "numbers_all = cubedRDD.collect()\n",
    "\n",
    "# Print the numbers from numbers_all\n",
    "for numb in numbers_all:\n",
    "\tprint(numb)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Filter and Count\n",
    "\n",
    "# Filter the fileRDD to select lines with Spark keyword\n",
    "fileRDD_filter = fileRDD.filter(lambda line: 'Spark' in line)\n",
    "\n",
    "# How many lines are there in fileRDD?\n",
    "print(\"The total number of lines with the keyword Spark is\", fileRDD_filter.count())\n",
    "\n",
    "# Print the first four lines of fileRDD\n",
    "for line in fileRDD_filter.take(4): \n",
    "    print(line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Filter and Count\n",
    "\n",
    "# Filter the fileRDD to select lines with Spark keyword\n",
    "fileRDD_filter = fileRDD.filter(lambda line: 'Spark' in line)\n",
    "\n",
    "# How many lines are there in fileRDD?\n",
    "print(\"The total number of lines with the keyword Spark is\", fileRDD_filter.count())\n",
    "\n",
    "# Print the first four lines of fileRDD\n",
    "for line in fileRDD_filter.take(4): \n",
    "    print(line)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## ReduceBykey and Collect\n",
    "\n",
    "# Sort the reduced RDD with the key by descending order\n",
    "Rdd_Reduced_Sort = Rdd_Reduced.sortByKey(ascending=False)\n",
    "\n",
    "# Iterate over the result and print the output\n",
    "for num in Rdd_Reduced_Sort.collect():\n",
    "    print(\"Key {} has {} Counts\".format(num[0], num[1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## CountingBykeys\n",
    "\n",
    "# Transform the rdd with countByKey()\n",
    "total = Rdd.countByKey()\n",
    "\n",
    "# What is the type of total?\n",
    "print(\"The type of total is\", type(total))\n",
    "\n",
    "# Iterate over the total and print the output\n",
    "for k, v in total.items(): \n",
    "    print(\"key\", k, \"has\", v, \"counts\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Create a base RDD and transform it\n",
    "\n",
    "# Create a baseRDD from the file path\n",
    "baseRDD = sc.textFile(file_path)\n",
    "\n",
    "# Split the lines of baseRDD into words\n",
    "splitRDD = baseRDD.flatMap(lambda x: x.split())\n",
    "\n",
    "# Count the total number of words\n",
    "print(\"Total number of words in splitRDD:\", splitRDD.count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Remove stop words and reduce the dataset\n",
    "\n",
    "# Convert the words in lower case and remove stop words from stop_words\n",
    "splitRDD_no_stop = splitRDD.filter(lambda x: x.lower() not in stop_words)\n",
    "\n",
    "# Create a tuple of the word and 1 \n",
    "splitRDD_no_stop_words = splitRDD_no_stop.map(lambda w: (w, 1))\n",
    "\n",
    "# Count of the number of occurences of each word\n",
    "resultRDD = splitRDD_no_stop_words.reduceByKey(lambda x, y: x + y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Print word frequencies\n",
    "\n",
    "# Display the first 10 words and their frequencies\n",
    "for word in resultRDD.take(10):\n",
    "\tprint(word)\n",
    "\n",
    "# Swap the keys and values \n",
    "resultRDD_swap = resultRDD.map(lambda x: (x[1], x[0]))\n",
    "\n",
    "# Sort the keys in descending order\n",
    "resultRDD_swap_sort = resultRDD_swap.sortByKey(ascending=False)\n",
    "\n",
    "# Show the top 10 most frequent words and their frequencies\n",
    "for word in resultRDD_swap_sort.take(10):\n",
    "\tprint(\"{} has {} counts\". format(word[1], word[0]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "3 - PySpark SQL & DataFrames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a list of tuples\n",
    "sample_list = [('Mona',20), ('Jennifer',34), ('John',20), ('Jim',26)]\n",
    "\n",
    "# Create a RDD from the list\n",
    "rdd = sc.parallelize(sample_list)\n",
    "\n",
    "# Create a PySpark DataFrame\n",
    "names_df = spark.createDataFrame(rdd, schema=['Name', 'Age'])\n",
    "\n",
    "# Check the type of names_df\n",
    "print(\"The type of names_df is\", type(names_df))\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Loading CSV into DataFrame\n",
    "\n",
    "# Create an DataFrame from file_path\n",
    "people_df = spark.read.csv(file_path, header=True, inferSchema=True)\n",
    "\n",
    "# Check the type of people_df\n",
    "print(\"The type of people_df is\", type(people_df))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Inspecting data in PySpark DataFrame\n",
    "\n",
    "# Print the first 10 observations \n",
    "people_df.show(10)\n",
    "\n",
    "# Count the number of rows \n",
    "print(\"There are {} rows in the people_df DataFrame.\".format(people_df.count()))\n",
    "\n",
    "# Count the number of columns and their names\n",
    "print(\"There are {} columns in the people_df \\\n",
    "        DataFrame and their names are {}\".format(len(people_df.columns), people_df.columns))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## PySpark DataFrame subsetting and cleaning\n",
    "\n",
    "# Select name, sex and date of birth columns\n",
    "people_df_sub = people_df.select('name', 'sex', 'date of birth')\n",
    "\n",
    "# Print the first 10 observations from people_df_sub\n",
    "people_df_sub.show(10)\n",
    "\n",
    "# Remove duplicate entries from people_df_sub\n",
    "people_df_sub_nodup = people_df_sub.dropDuplicates()\n",
    "\n",
    "# Count the number of rows\n",
    "print(\"There were {} rows before removing duplicates,\\\n",
    "       and {} rows after removing duplicates\".format(people_df_sub.count(),\n",
    "                                                     people_df_sub_nodup.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Running SQL Queries Programmatically\n",
    "\n",
    "# Create a temporary table \"people\"\n",
    "people_df.createOrReplaceTempView(\"people\")\n",
    "\n",
    "# Construct a query to select the names of the people from the temporary table \"people\"\n",
    "query = '''SELECT name FROM people'''\n",
    "\n",
    "# Assign the result of Spark's query to people_df_names\n",
    "people_df_names = spark.sql(query)\n",
    "\n",
    "# Print the top 10 names of the people\n",
    "people_df_names.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## SQL queries for filtering Table\n",
    "\n",
    "# Filter the people table to select female sex \n",
    "people_female_df = spark.sql('SELECT * FROM people WHERE sex==\"female\"')\n",
    "\n",
    "# Filter the people table DataFrame to select male sex\n",
    "people_male_df = spark.sql('SELECT * FROM people WHERE sex==\"male\"')\n",
    "\n",
    "# Count the number of rows in both DataFrames\n",
    "print(\"There are {} rows in the people_female_df and {} rows \\\n",
    "       in the people_male_df DataFrames\".format(people_female_df.count(), \n",
    "                                                people_male_df.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## PySpark DataFrame visualization\n",
    "\n",
    "# Check the column names of names_df\n",
    "print(\"The column names of names_df are\", names_df.columns)\n",
    "\n",
    "# Convert to Pandas DataFrame  \n",
    "df_pandas = names_df.toPandas()\n",
    "\n",
    "# Create a horizontal bar plot\n",
    "df_pandas.plot(kind='barh', x='Name', y='Age', colormap='winter_r')\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Part 1: Create a DataFrame from CSV file\n",
    "\n",
    "# Load the Dataframe\n",
    "fifa_df = spark.read.csv(file_path, \n",
    "                         header=True,\n",
    "                         inferSchema=True)\n",
    "\n",
    "# Check the schema of columns\n",
    "fifa_df.printSchema()\n",
    "\n",
    "# Show the first 10 observations\n",
    "fifa_df.show(10)\n",
    "\n",
    "# Print the total number of rows\n",
    "print(\"There are {} rows in the fifa_df DataFrame\".format(fifa_df.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Part 2: SQL Queries on DataFrame\n",
    "\n",
    "# Create a temporary view of fifa_df\n",
    "fifa_df.createOrReplaceTempView('fifa_df_table')\n",
    "\n",
    "# Construct the \"query\"\n",
    "query = '''SELECT Age FROM fifa_df_table WHERE Nationality == \"Germany\"'''\n",
    "\n",
    "# Apply the SQL \"query\"\n",
    "fifa_df_germany_age = spark.sql(query)\n",
    "\n",
    "# Generate basic statistics\n",
    "fifa_df_germany_age.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Part 3: Data visualization\n",
    "\n",
    "# Convert fifa_df to fifa_df_germany_age_pandas DataFrame\n",
    "fifa_df_germany_age_pandas = fifa_df_germany_age.toPandas()\n",
    "\n",
    "# Plot the 'Age' density of Germany Players\n",
    "fifa_df_germany_age_pandas.plot(kind='density')\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "4 - Machine Learning with PySpark MLlib"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## PySpark MLlib algorithms\n",
    "\n",
    "# Import the library for ALS\n",
    "from pyspark.mllib.recommendation import ALS\n",
    "\n",
    "# Import the library for Logistic Regression\n",
    "from pyspark.mllib.classification import LogisticRegressionWithLBFGS\n",
    "\n",
    "# Import the library for Kmeans\n",
    "from pyspark.mllib.clustering import KMeans"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Loading Movie Lens dataset into RDDs\n",
    "\n",
    "# Load the data into RDD\n",
    "data = sc.textFile(file_path)\n",
    "\n",
    "# Split the RDD \n",
    "ratings = data.map(lambda l: l.split(','))\n",
    "\n",
    "# Transform the ratings RDD \n",
    "ratings_final = ratings.map(lambda line: Rating(int(line[0]), int(line[1]), float(line[2])))\n",
    "\n",
    "# Split the data into training and test\n",
    "training_data, test_data = ratings_final.randomSplit([0.8, 0.2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Model training and predictions\n",
    "\n",
    "# Create the ALS model on the training data\n",
    "model = ALS.train(training_data, rank=10, iterations=10)\n",
    "\n",
    "# Drop the ratings column \n",
    "testdata_no_rating = test_data.map(lambda p: (p[0], p[1]))\n",
    "\n",
    "# Predict the model  \n",
    "predictions = model.predictAll(testdata_no_rating)\n",
    "\n",
    "# Print the first rows of the RDD\n",
    "predictions.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Model evaluation using MSE\n",
    "\n",
    "# Prepare ratings data\n",
    "rates = ratings_final.map(lambda r: ((r[0], r[1]),\n",
    "                                     r[2]))\n",
    "\n",
    "# Prepare predictions data\n",
    "preds = predictions.map(lambda r: ((r[0], r[1]),\n",
    "                                   r[2]))\n",
    "\n",
    "# Join the ratings data with predictions data\n",
    "rates_and_preds = rates.join(preds)\n",
    "\n",
    "# Calculate and print MSE\n",
    "MSE = rates_and_preds.map(lambda r: (r[1][0] - r[1][1])**2).mean()\n",
    "print(\"Mean Squared Error of the model for the test data = {:.2f}\".format(MSE))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Loading spam and non-spam data\n",
    "\n",
    "# Load the datasets into RDDs\n",
    "spam_rdd = sc.textFile(file_path_spam)\n",
    "non_spam_rdd = sc.textFile(file_path_non_spam)\n",
    "\n",
    "# Split the email messages into words\n",
    "spam_words = spam_rdd.flatMap(lambda email: email.split(' '))\n",
    "non_spam_words = non_spam_rdd.flatMap(lambda email: email.split(' '))\n",
    "\n",
    "# Print the first element in the split RDD\n",
    "print(\"The first element in spam_words is\", spam_words.first())\n",
    "print(\"The first element in non_spam_words is\", non_spam_words.first())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Feathre Hashing and LabelPoint\n",
    "\n",
    "# Create a HashingTf instance with 200 features\n",
    "tf = HashingTF(numFeatures=200)\n",
    "\n",
    "# Map each word to one feature\n",
    "spam_features = tf.transform(spam_words)\n",
    "non_spam_features = tf.transform(non_spam_words)\n",
    "\n",
    "# Label the features: 1 for spam, 0 for non-spam\n",
    "spam_samples = spam_features.map(lambda features: LabeledPoint(1, features))\n",
    "non_spam_samples = non_spam_features.map(lambda features: LabeledPoint(0, features))\n",
    "\n",
    "# Combine the two datasets\n",
    "samples = spam_samples.join(non_spam_samples)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Logistic Regression model training\n",
    "\n",
    "# Split the data into training and testing\n",
    "train_samples,test_samples = samples.randomSplit([0.8, 0.2])\n",
    "\n",
    "# Train the model\n",
    "model = LogisticRegressionWithLBFGS.train(train_samples)\n",
    "\n",
    "# Create a prediction label from the test data\n",
    "predictions = model.predict(test_samples.map(lambda x: x.features))\n",
    "\n",
    "# Combine original labels with the predicted labels\n",
    "labels_and_preds = test_samples.map(lambda x: x.label).zip(predictions)\n",
    "\n",
    "# Check the accuracy of the model on the test data\n",
    "accuracy = labels_and_preds.filter(lambda x: x[0] == x[1]).count() / float(test_samples.count())\n",
    "print(\"Model accuracy : {:.2f}\".format(accuracy))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Loading and parsing the 5000 points data\n",
    "\n",
    "# Load the dataset into a RDD\n",
    "clusterRDD = sc.textFile(file_path)\n",
    "\n",
    "# Split the RDD based on tab\n",
    "rdd_split = clusterRDD.map(lambda x: x.split(\"\\t\"))\n",
    "\n",
    "# Transform the split RDD by creating a list of integers\n",
    "rdd_split_int = rdd_split.map(lambda x: [int(x[0]), int(x[1])])\n",
    "\n",
    "# Count the number of rows in RDD \n",
    "print(\"There are {} rows in the rdd_split_int dataset\".format(rdd_split_int.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## K-means training\n",
    "\n",
    "# Train the model with clusters from 13 to 16 and compute WSSSE \n",
    "for clst in range(13, 17):\n",
    "    model = KMeans.train(rdd_split_int, clst, seed=1)\n",
    "    WSSSE = rdd_split_int.map(lambda point: error(point)).reduce(lambda x, y: x + y)\n",
    "    print(\"The cluster {} has Within Set Sum of Squared Error {}\".format(clst, WSSSE))\n",
    "\n",
    "# Train the model again with the best k \n",
    "model = KMeans.train(rdd_split_int, k=15, seed=1)\n",
    "\n",
    "# Get cluster centers\n",
    "cluster_centers = model.clusterCenters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Visualizing clusters\n",
    "\n",
    "# Convert rdd_split_int RDD into Spark DataFrame\n",
    "rdd_split_int_df = spark.createDataFrame(rdd_split_int, schema=[\"col1\", \"col2\"])\n",
    "\n",
    "# Convert Spark DataFrame into Pandas DataFrame\n",
    "rdd_split_int_df_pandas = rdd_split_int_df.toPandas()\n",
    "\n",
    "# Convert \"cluster_centers\" that you generated earlier into Pandas DataFrame\n",
    "cluster_centers_pandas = pd.DataFrame(cluster_centers, columns=[\"col1\", \"col2\"])\n",
    "\n",
    "# Create an overlaid scatter plot\n",
    "plt.scatter(rdd_split_int_df_pandas[\"col1\"], rdd_split_int_df_pandas[\"col2\"])\n",
    "plt.scatter(cluster_centers_pandas[\"col1\"], cluster_centers_pandas[\"col2\"], color=\"red\", marker=\"x\")\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "tf-gpu-2.0",
   "language": "python",
   "name": "tf_gpu2"
  },
  "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.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}