<jupyter_start><jupyter_text>Task 1 - Install Spark, download datasets, create final dataframe. If you get an error regarding tar or wget, it is probably due to the Spark file being removed from the repository. Go to https://downloads.apache.org/spark/ and choose an equivalent version of Spark and Hadoop to download. So if 2.4.7 is not available, download the next version.<jupyter_code>!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q https://downloads.apache.org/spark/spark-3.1.2/spark-3.1.2-bin-hadoop3.2.tgz
!tar xf spark-3.1.2-bin-hadoop3.2.tgz
!pip install -q findspark
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.1.2-bin-hadoop3.2"
import findspark
findspark.init()
from google.colab import files
from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import isnan, when, count, col, lit, trim, avg, ceil
from pyspark.sql.types import StringType
import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns
!wget https://s3.amazonaws.com/drivendata/data/7/public/4910797b-ee55-40a7-8668-10efd5c1b960.csv -O features.csv
!wget https://s3.amazonaws.com/drivendata/data/7/public/0bf8bc6e-30d0-4c50-956a-603fc693d966.csv -O labels.csv
!ls
# Starting a spark session and loading these CSV files into spark
sc = SparkSession.builder.master("local[*]").getOrCreate() # This will set spark to run locally on all available cores<jupyter_output><empty_output><jupyter_text>Here, sc is the spark connection variable and you're telling spark to infer the schema of the files automatically. By setting inferSchema=true, Spark will automatically go through the csv file and infer the schema of each column.<jupyter_code>feature = sc.read.csv("features.csv", inferSchema=True, header=True)
label = sc.read.csv("labels.csv", inferSchema=True, header=True)
feature_shape = (feature.count(), len(feature.columns))
print(feature_shape)
print(feature.columns)<jupyter_output>(59400, 40)
['id', 'amount_tsh', 'date_recorded', 'funder', 'gps_height', 'installer', 'longitude', 'latitude', 'wpt_name', 'num_private', 'basin', 'subvillage', 'region', 'region_code', 'district_code', 'lga', 'ward', 'population', 'public_meeting', 'recorded_by', 'scheme_management', 'scheme_name', 'permit', 'construction_year', 'extraction_type', 'extraction_type_group', 'extraction_type_class', 'management', 'management_group', 'payment', 'payment_type', 'water_quality', 'quality_group', 'quantity', 'quantity_group', 'source', 'source_type', 'source_class', 'waterpoint_type', 'waterpoint_type_group']<jupyter_text>**Features.csv****amount_tsh** : Total static head (amount water available to waterpoint)**date_recorded** : The date the row was entered**funder** : Who funded the well**gps_height** : Altitude of the well**installer** : Organization that installed the well**longitude** : GPS coordinate**latitude** : GPS coordinate**wpt_name** : Name of the waterpoint if there is one**num_private** :**basin** : Geographic water basin**subvillage** : Geographic location**region** : Geographic location**region_code** : Geographic location (coded)**district_code** : Geographic location (coded)**lga** : Geographic location**ward** : Geographic location**population** : Population around the well**public_meeting** : True/False**recorded_by** : Group entering this row of data**scheme_management** : Who operates the waterpoint**scheme_name** : Who operates the waterpoint**permit** : If the waterpoint is permitted**construction_year** : Year the waterpoint was constructed**extraction_type** : The kind of extraction the waterpoint uses**extraction_type_group** : The kind of extraction the waterpoint uses**extraction_type_class** : The kind of extraction the waterpoint uses**management** : How the waterpoint is managed**management_group** : How the waterpoint is managed**payment** : What the water costs**payment_type** : What the water costs**water_quality** : The quality of the water**quality_group** : The quality of the water**quantity** : The quantity of water**quantity_group** : The quantity of water**source** : The source of the water**source_type** : The source of the water**source_class** : The source of the water**waterpoint_type** : The kind of waterpoint**waterpoint_type_group** : The kind of waterpoint***Labels***: There are three possible values:**functional** : the waterpoint is operational and there are no repairs needed**functional needs repair** : the waterpoint is operational, but needs repairs**non functional** : the waterpoint is not operational<jupyter_code>label_shape = (label.count(), len(label.columns))
print(label_shape)
print(label.columns)
data = feature.join(label, on=('id'))
data_shape = (data.count(), len(data.columns))
print(data_shape)
print(data.columns)<jupyter_output>(59400, 41)
['id', 'amount_tsh', 'date_recorded', 'funder', 'gps_height', 'installer', 'longitude', 'latitude', 'wpt_name', 'num_private', 'basin', 'subvillage', 'region', 'region_code', 'district_code', 'lga', 'ward', 'population', 'public_meeting', 'recorded_by', 'scheme_management', 'scheme_name', 'permit', 'construction_year', 'extraction_type', 'extraction_type_group', 'extraction_type_class', 'management', 'management_group', 'payment', 'payment_type', 'water_quality', 'quality_group', 'quantity', 'quantity_group', 'source', 'source_type', 'source_class', 'waterpoint_type', 'waterpoint_type_group', 'status_group']<jupyter_text>Task 2 - Change column type, drop duplicated rows, remove whitespacs.<jupyter_code>print(data.printSchema())
print(data.show(10))
data = data.withColumn('region_code', col('region_code').cast(StringType())).withColumn('district_code', col('district_code').cast(StringType()))
data.printSchema()
data = data.drop_duplicates(['id'])
data.count()
# Hence, no duplicates.
#Removing whitespaces in each of these columns
str_cols = [item[0] for item in data.dtypes if item[1].startswith('string')]
for cols in str_cols:
  data = data.withColumn(cols, trim(data[cols]))<jupyter_output><empty_output><jupyter_text>Task 3 - Remove columns with avg. number of null values more than a threshold.Columns with too many null values are uninformative and can cause issues in any ML model. Using a default threshold value beyond which you can drop these columns is a good strategy to deal with columns with high number of null values.<jupyter_code>data.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in data.columns if c not in {'date_recorded', 'public_meeting', 'permit'}]).show()<jupyter_output>+---+----------+------+----------+---------+---------+--------+--------+-----------+-----+----------+------+-----------+-------------+---+----+----------+-----------+-----------------+-----------+-----------------+---------------+---------------------+---------------------+----------+----------------+-------+------------+-------------+-------------+--------+--------------+------+-----------+------------+---------------+---------------------+------------+
| id|amount_tsh|funder|gps_height|installer|longitude|latitude|wpt_name|num_private|basin|subvillage|region|region_code|district_code|lga|ward|population|recorded_by|scheme_management|scheme_name|construction_year|extraction_type|extraction_type_group|extraction_type_class|management|management_group|payment|payment_type|water_quality|quality_group|quantity|quantity_group|source|source_type|source_class|waterpoint_type|waterpoint_type_group|status_group|
+---+----------+------+----------+---------+---------+--------+--------+----------[...]<jupyter_text>Notice that the column name "scheme_name" has 28,000 null values, which is more than 40% of the total values in that column. But instead of manually going through each value and deciding whether to keep or drop the column, we can set a threshold value beyond which a column can be dropped. We will modify the current cell and get the average number of null values in a variable.<jupyter_code>agg_row = data.select([(count(when(isnan(c) | col(c).isNull(), c))/data.count()).alias(c) for c in data.columns if c not in {'date_recorded', 'public_meeting', 'permit'}]).collect()
# Convert the result into a dictionary for easy iteration later on.
agg_dict_list = [row.asDict() for row in agg_row]
agg_dict = agg_dict_list[0]
col_null = list({i for i in agg_dict if agg_dict[i] > 0.4})
print(agg_dict)
print(col_null)
data = data.drop(*col_null)<jupyter_output>{'id': 0.0, 'amount_tsh': 0.0, 'funder': 0.0611952861952862, 'gps_height': 0.0, 'installer': 0.061531986531986535, 'longitude': 0.0, 'latitude': 0.0, 'wpt_name': 0.0, 'num_private': 0.0, 'basin': 0.0, 'subvillage': 0.0062457912457912455, 'region': 0.0, 'region_code': 0.0, 'district_code': 0.0, 'lga': 0.0, 'ward': 0.0, 'population': 0.0, 'recorded_by': 0.0, 'scheme_management': 0.06526936026936027, 'scheme_name': 0.4741750841750842, 'construction_year': 0.0, 'extraction_type': 0.0, 'extraction_type_group': 0.0, 'extraction_type_class': 0.0, 'management': 0.0, 'management_group': 0.0, 'payment': 0.0, 'payment_type': 0.0, 'water_quality': 0.0, 'quality_group': 0.0, 'quantity': 0.0, 'quantity_group': 0.0, 'source': 0.0, 'source_type': 0.0, 'source_class': 0.0, 'waterpoint_type': 0.0, 'waterpoint_type_group': 0.0, 'status_group': 0.0}
['scheme_name']<jupyter_text>Task 4 - Group, aggregate, create pivot table.Creating aggregations and pivot tables over a grouping can help you answer questions such as number of customers over time, or total amount of sales per each product.<jupyter_code># Counting the frequencies of each category in a column
data.groupBy('recorded_by').count().show() #not-informative, hence drop
data.groupBy('water_quality').count().orderBy('count', ascending=False).show()
data = data.drop('recorded_by')
# Creating a pivot table grouped by the status of the well in each region and calculate the sum of the total amount of water in the wells
data.groupBy('status_group').pivot('region').sum('amount_tsh').show()<jupyter_output>+--------------------+---------+-------------+------+---------+------+--------+------------------+-------+--------+--------+-----+--------+-------+-------+---------+--------+---------+---------+--------+------+--------+
|        status_group|   Arusha|Dar es Salaam|Dodoma|   Iringa|Kagera|  Kigoma|       Kilimanjaro|  Lindi| Manyara|    Mara|Mbeya|Morogoro| Mtwara| Mwanza|    Pwani|   Rukwa|   Ruvuma|Shinyanga| Singida|Tabora|   Tanga|
+--------------------+---------+-------------+------+---------+------+--------+------------------+-------+--------+--------+-----+--------+-------+-------+---------+--------+---------+---------+--------+------+--------+
|functional needs ...|  10280.0|          0.0|   0.0| 118400.0|   0.0|270770.0|          129248.0| 3625.0| 71315.0|   900.0|  0.0| 70005.0|36240.0|10000.0|    520.0|292980.0| 125650.0|   2200.0|  6505.0|   0.0|  4310.0|
|          functional|1125485.0|     181175.0|   0.0|5515710.0|   0.0|530725.0|         746942.25|24560.0|312745.0| 9678[...]<jupyter_text>Generally speaking, functional wells tend to have higher water quantity than non functional wells or wells that needs repair. Task 5 - Convert categories with low frequency to Others, impute missing values.In this task, we will rename categories with lower frequencies in character columns and impute missing values in numeric columns. In character columns, categories which are lowly represented are not informative and can be grouped together as a single category. In numeric columns, missing values should be imputed or given a default value. Rather than giving a default value of 0 or the mean of the full column, it is better to give the average value over some grouping.<jupyter_code>print(str_cols)<jupyter_output>['date_recorded', 'funder', 'installer', 'wpt_name', 'basin', 'subvillage', 'region', 'region_code', 'district_code', 'lga', 'ward', 'recorded_by', 'scheme_management', 'scheme_name', 'extraction_type', 'extraction_type_group', 'extraction_type_class', 'management', 'management_group', 'payment', 'payment_type', 'water_quality', 'quality_group', 'quantity', 'quantity_group', 'source', 'source_type', 'source_class', 'waterpoint_type', 'waterpoint_type_group', 'status_group']<jupyter_text>Within each column, you would want to convert categories which occurred less than, say, a 1000 times to a category called "Others". This will reduce the cardinality of columns, which have high number of unique categories but occuring less frequently. This will help in creating better visualizations and reduce the number of one hot encoding required in machine learning algorithms.Trying this for the first two character columns. First, print the current categories and their frequency in descending order. And store this pair of categories and frequencies in a variable. Iterate over each pair and store the category's name if its frequency is below 1000, in a new variable called lessthan. Lastly, using a case when clause, if a category occurs in the variable "lessthan", replace the category with a new category called "Others" else, keep the category.<jupyter_code>for column in str_cols[:2]:
  print(data.groupBy(column).count().orderBy('count', ascending=False).show())
  values_cat = data.groupBy(column).count().collect()
  lessthan = [x[0] for x in values_cat if x[1] < 1000] # Threshold of 1000 is arbitrary & should be changed according to the need
  data = data.withColumn(column, when(col(column).isin(lessthan), 'Others').otherwise(col(column)))
  data.groupBy(column).count().orderBy('count', ascending=False).show()<jupyter_output>+-------------+-----+
|date_recorded|count|
+-------------+-----+
|   2011-03-15|  572|
|   2011-03-17|  558|
|   2013-02-03|  546|
|   2011-03-14|  520|
|   2011-03-16|  513|
|   2011-03-18|  497|
|   2011-03-19|  466|
|   2013-02-04|  464|
|   2013-01-29|  459|
|   2011-03-04|  458|
|   2013-02-14|  444|
|   2013-01-24|  435|
|   2011-03-05|  434|
|   2013-02-15|  429|
|   2013-03-15|  428|
|   2011-03-11|  426|
|   2013-01-30|  421|
|   2013-02-16|  418|
|   2011-03-23|  417|
|   2011-03-09|  416|
+-------------+-----+
only showing top 20 rows

None
+-------------+-----+
|date_recorded|count|
+-------------+-----+
|       Others|59400|
+-------------+-----+

+--------------------+-----+
|              funder|count|
+--------------------+-----+
|Government Of Tan...| 9084|
|                null| 3635|
|              Danida| 3114|
|              Hesawa| 2202|
|               Rwssp| 1374|
|          World Bank| 1349|
|                Kkkt| 1287|
|        World Vision| 1246|
|          [...]<jupyter_text>We can see that for the "funder" column, initially, there were several categories with low frequencies and these low frequencies categories were grouped together as Others, and it reduces the number of categories to 10. Similar results were seen for the next column, "installer".<jupyter_code># Printing out the frequency of each population recorded and order by the population rather than the frequency.
data.groupBy('population').count().orderBy('population').show()<jupyter_output>+----------+-----+
|population|count|
+----------+-----+
|         0|21381|
|         1| 7025|
|         2|    4|
|         3|    4|
|         4|   13|
|         5|   44|
|         6|   19|
|         7|    3|
|         8|   23|
|         9|   11|
|        10|  163|
|        11|    7|
|        12|   43|
|        13|   12|
|        14|   18|
|        15|  193|
|        16|    8|
|        17|    6|
|        18|   18|
|        19|    3|
+----------+-----+
only showing top 20 rows<jupyter_text>We can see that, there are 28,000 wells which have population either zero or one. This is an error and should be fixed. One way would be to replace this population by the average population in a district where the well is located. To do so, first, impute this population values with Null values. Next, partition your data on the column "district code". Now replace all the null values in the column with the average population over the partition window. Lastly, we should round up the imputed population values using the ceil() function. Ceil() rounds a number UP to the nearest integer. Now print out the frequency of each population recorded and ordered by the population again.<jupyter_code>data = data.withColumn('population', when(col('population') < 2, lit(None)).otherwise(col('population')))
w = Window.partitionBy(data['district_code'])
data = data.withColumn('population', when(col('population').isNull(), avg(data['population']).over(w)).otherwise(col('population')))
data = data.withColumn('population', ceil(data['population']))
data.groupBy('population').count().orderBy('population').show()<jupyter_output>+----------+-----+
|population|count|
+----------+-----+
|      null|   35|
|         2|    4|
|         3|    4|
|         4|   13|
|         5|   44|
|         6|   19|
|         7|    3|
|         8|   23|
|         9|   11|
|        10|  163|
|        11|    7|
|        12|   43|
|        13|   12|
|        14|   18|
|        15|  193|
|        16|    8|
|        17|    6|
|        18|   18|
|        19|    3|
|        20|  462|
+----------+-----+
only showing top 20 rows<jupyter_text>You'll notice that the population value 0 and 1 don't exist anymore. Although there are 35 null values, as some district do not have population recorded. One can impute these null values by the average population in the region. Task 6 - Make visualizations In this final task, we will create visualizations using Seaborn Library to gather insights into ours big data set. Unfortunately, Spark doesn't come with any visualization library, so one has to convert the Spark data frame to pandas data frame before creating visualizations. As big data can be extremely large one trick that can be utilized is to select the required columns for plotting and then convert this subsetted data set to Pandas data frame.<jupyter_code>color_status = {'functional': 'green', 'non functional': 'red', 'functional needs repair': 'blue'}
cols = ['status_group', 'payment_type', 'longitude', 'latitude', 'gps_height']
df = data.select(cols).toPandas()
fig, ax = plt.subplots(figsize=(12,8))
sns.countplot(x='payment_type', hue='status_group', data=df, ax=ax, palette=color_status)
plt.xticks(rotation=45)<jupyter_output><empty_output><jupyter_text>We can notice that wells that are paid annually, monthly or by bucket fare better than ones that are never paid.<jupyter_code>fig, ax = plt.subplots(figsize=(12,8))
sns.scatterplot(x='longitude', y='latitude', data=df, hue='status_group', ax=ax, palette=color_status)
# Here, we don't seem any pattern here, but you do see an error of latitude and longitude being zero.<jupyter_output><empty_output><jupyter_text>Now, creating a histogram with kernel density estimates of the column GPS height using the distplot function. This column tells you the altitude of the well. To create such a plot first, you need to separate the data frame into three subsets, one for each type of status.<jupyter_code>row_functional = (df['status_group'] == 'functional')
row_non_functional = (df['status_group'] == 'non functional')
row_repair = (df['status_group'] == 'functional needs replair')
col = 'gps_height'
fig, ax = plt.subplots(figsize=(12,8))
sns.distplot(df[col][row_functional], color='green', label='functional', ax=ax)
sns.distplot(df[col][row_non_functional], color='red', label='non functional', ax=ax)
sns.distplot(df[col][row_repair], color='blue', label='functional needs repair', ax=ax)
plt.legend()<jupyter_output>/usr/local/lib/python3.7/dist-packages/seaborn/distributions.py:2557: FutureWarning: `distplot` is a deprecated function and will be removed in a future version. Please adapt your code to use either `displot` (a figure-level function with similar flexibility) or `histplot` (an axes-level function for histograms).
  warnings.warn(msg, FutureWarning)
/usr/local/lib/python3.7/dist-packages/seaborn/distributions.py:2557: FutureWarning: `distplot` is a deprecated function and will be removed in a future version. Please adapt your code to use either `displot` (a figure-level function with similar flexibility) or `histplot` (an axes-level function for histograms).
  warnings.warn(msg, FutureWarning)
/usr/local/lib/python3.7/dist-packages/seaborn/distributions.py:2557: FutureWarning: `distplot` is a deprecated function and will be removed in a future version. Please adapt your code to use either `displot` (a figure-level function with similar flexibility) or `histplot` (an axes-level function [...]<jupyter_text>We can notice that in general, wells at lower altitude are more functional than wells at higher altitude. Additionally, certain wells have negative heights and these should be fixed. One can fix these negative values by imputing the negative values with the average height of the region or district.<jupyter_code>data.groupBy(col).count().orderBy(col).show()<jupyter_output>+----------+-----+
|gps_height|count|
+----------+-----+
|       -90|    1|
|       -63|    2|
|       -59|    1|
|       -57|    1|
|       -55|    1|
|       -54|    1|
|       -53|    1|
|       -52|    2|
|       -51|    2|
|       -50|    5|
|       -49|    4|
|       -48|    6|
|       -47|    8|
|       -46|    6|
|       -45|    9|
|       -44|    6|
|       -43|   11|
|       -42|   13|
|       -41|   11|
|       -40|   26|
+----------+-----+
only showing top 20 rows
{'fraction_non_alphanumeric': 0.1773304688959462, 'fraction_numerical': 0.044461049878572766, 'mean_word_length': 3.531851851851852, 'pattern_counts': {'":': 0, '<': 52, '<?xml version=': 0, '>': 52, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '15', 'dir': 'jupyter-structured-clean-dedup', 'id': '660950', 'license': 'MIT', 'n_tokens_mistral': 7136, 'n_tokens_neox': 6417, 'n_words': 2042, 'path': 'Cleaning and Exploring Big Data using PySpark/Cleaning_and_Exploring_Big_Data_using_PySpark.ipynb', 'repo_name': 'padam56/Data-Analysis-and-Visualization-Projects', 'size': '21412'}