# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     name: python3
# ---

# + id="_cS9tEuBejIP" colab_type="code" colab={}
# Setting up Spark environment

# !apt-get install openjdk-8-jdk-headless -qq > /dev/null
# !wget -q https://www-us.apache.org/dist/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz
# !tar xf spark-2.4.5-bin-hadoop2.7.tgz
# !pip install -q findspark

# + id="pSKJdE-ahPiK" colab_type="code" colab={}
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.4.5-bin-hadoop2.7"

# + id="0kvsRYRoiFam" colab_type="code" outputId="ffb6c638-1448-49b9-995d-6baab73db502" colab={"base_uri": "https://localhost:8080/", "height": 34}
# Mounting on Drive

from google.colab import drive
drive.mount('/content/drive')

# + id="Dt5j0Fj3igmK" colab_type="code" outputId="c44cbb70-d038-4b09-bcdf-0ec83a312fa5" colab={"base_uri": "https://localhost:8080/", "height": 51}
pip install pyspark

# + id="kidXCfMiitRn" colab_type="code" colab={}
# Importing libraries

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
import matplotlib.pyplot as plt
import seaborn as sns
import pandas as pd
import numpy as np

# + [markdown] id="PUAkV6T3hKmO" colab_type="text"
# **1. Explore configuration options on SparkSession**

# + id="-3YVwjSzi7ji" colab_type="code" colab={}
# Building Spark session with config details

spark = SparkSession.builder \
    .master("local") \
    .appName("Assignment_6") \
    .config("spark.some.config.option", "some-value") \
    .config("driver.cores", "1") \
    .config("broadcast.compress", "true") \
    .config("checkpoint.compress", "false") \
    .config("streaming.backpressure.enabled", "false") \
    .getOrCreate()

# + [markdown] id="Igbz7kt72dI3" colab_type="text"
#  **2. READING DATA**

# + [markdown] id="5YCz2Lf108my" colab_type="text"
# READ DATA AS DATAFRAME

# + [markdown] id="TyLpLB0lkbNL" colab_type="text"
# Read a csv file into spark dataframe, considering first row as headers. A csv file is a comma seperated file

# + id="WGUa4VbLpRl7" colab_type="code" colab={}
# Read data into dataframe

df = spark.read.format("csv").option("header", "true").load("/content/sample_data/california_housing_test.csv")

# + id="Kxx9SBGemtOL" colab_type="code" outputId="0d6622d9-121e-4b0c-89e7-1ecfbd759062" colab={"base_uri": "https://localhost:8080/", "height": 459}
# Look at the dataframe

df.show()

# + [markdown] id="xvPN3LkKqL2I" colab_type="text"
# READING DATA AS RDD

# + [markdown] id="x-KlSoEpdqqQ" colab_type="text"
# Resilient Distributed Datasets (RDD) is a fundamental data structure of Spark. It is an immutable distributed collection of objects

# + id="j5rsZAnKqOWk" colab_type="code" colab={}
# Read csv to RDD

rdd_1 = spark.sparkContext.textFile("/content/sample_data/california_housing_test.csv")

# + id="vIwojaRkq4Xl" colab_type="code" outputId="adacf5e1-0f21-4a1d-da5c-b8591162164f" colab={"base_uri": "https://localhost:8080/", "height": 102}
# View RDD top 5 records

rdd_1.take(5)

# + [markdown] id="n5GcKEsPxWuk" colab_type="text"
# CONVERT RDD TO SPARK DATAFRAME

# + id="6CTeyTsCx8h3" colab_type="code" colab={}
# Importing libraries

from pyspark.sql.functions import *
from pyspark.sql import Row
from pyspark.sql.types import *

# + id="SKVz1q9QxYfL" colab_type="code" outputId="c6c97009-6996-47a0-a32a-8aef12b3f530" colab={"base_uri": "https://localhost:8080/", "height": 289}
#Converting to dataframe

get_data = rdd_1.map(lambda x : x.split(","))
dataframe = get_data.map(lambda p: Row(longitude=p[0],latitude=p[1],housing_median_age=p[2],total_rooms=p[3],total_bedrooms=p[4],population=p[5],households=p[6],median_income=p[7],median_house_value=p[8]))
rdd_df = spark.createDataFrame(dataframe)
rdd_df.show(10)

# + [markdown] id="GKsEW3Hb0i1u" colab_type="text"
# CONVERT SPARK DATAFRAME TO RDD 

# + id="5O-whp0b0lc_" colab_type="code" outputId="126c22f1-48ec-47ba-bac6-8081aa9fa7e4" colab={"base_uri": "https://localhost:8080/", "height": 34}
# Creating dataframe from the previously generated RDD

df_rdd1 = df.rdd
df_rdd1

# + [markdown] id="BqRSbj6M1q3c" colab_type="text"
# CONVERT SPARK DATAFRAME TO PANDAS DATAFRAME

# + id="r1EtwozlziAU" colab_type="code" outputId="64505934-0e59-449a-f571-321bf53598c8" colab={"base_uri": "https://localhost:8080/", "height": 419}
# Converting the spark dataframe into pandas dataframe

pandas_df = df.toPandas()
pandas_df

# + [markdown] id="PCmoTOJQ2n9m" colab_type="text"
#  **3. TASKS**

# + [markdown] id="Nz7A9PJNl26i" colab_type="text"
# Here, I performed 8 taskes on my dataset using pyspark.sql.dataframe functions and sql queries. In order to implement SQL queries, we either need to connect to our SQL server or use a temporary view. Here, I created a temporary view for SQL purposes.

# + [markdown] id="Q6Gl-j_M3HCR" colab_type="text"
# 1. Select first 10 rows of dataset - pyspark.sql.dataframe

# + id="uLqTjQqH2Omx" colab_type="code" outputId="a63e6025-d2ad-494a-9474-ebc210cdc5c3" colab={"base_uri": "https://localhost:8080/", "height": 289}
# Looking at the dataframe's first 10 rows

df.show(10)

# + [markdown] id="22cIFoeXC93U" colab_type="text"
# 1. Select first 10 rows of dataset - SQL

# + id="1dZHF5KqCU0A" colab_type="code" colab={}
# Creating a temperary view to implement SQL queries in Spark

df.createOrReplaceTempView("df_sql")

# + id="tmDYHh8r_WMI" colab_type="code" outputId="96a3b6fc-8411-4c27-cbee-be133b2ad238" colab={"base_uri": "https://localhost:8080/", "height": 272}
# First 10 rows using SQL

task_1 = 'select * from df_sql LIMIT 10'
spark.sql(task_1).show()

# + [markdown] id="a9dpSIs23U-_" colab_type="text"
# 2. Show the schema of of the dataset - pyspark.sql.dataframe

# + [markdown] id="iN7ZJ4mmmnTS" colab_type="text"
# Looking at the datatypes and nullable constraints for each column

# + id="cqx60kgL3Kyy" colab_type="code" outputId="c2b7ef98-b04e-435d-88d9-22330f227b46" colab={"base_uri": "https://localhost:8080/", "height": 204}
# Schema of our data using pyspark

df.printSchema()

# + [markdown] id="V6u8TZFIDf_J" colab_type="text"
# 2. Show the schema of of the dataset - SQL

# + id="vrgq7JIdDjZE" colab_type="code" outputId="4011fbed-62e9-4110-a5f9-a3b69671b16d" colab={"base_uri": "https://localhost:8080/", "height": 255}
# Schema using Sql's describe method using SQL
 
task_2 = "DESCRIBE FORMATTED df_sql"
spark.sql(task_2).show()

# + [markdown] id="pzLkThDu5UGs" colab_type="text"
# 3. Group by and get max, min, count of a column in the dataset - pyspark.sql.dataframe

# + [markdown] id="B0-GeufPxJHa" colab_type="text"
# The GROUP BY statement groups rows that have the same values into summary rows

# + id="4aTwyDS73Zf0" colab_type="code" outputId="51516973-33d2-4aa6-acfb-4afd87630d69" colab={"base_uri": "https://localhost:8080/", "height": 561}
# count of each income class and it's min and max values using pyspark

df.groupBy('median_income').count().show()
df.select(max('median_income'),min('median_income')).show()

# + [markdown] id="pi5Zh_B9Ftqz" colab_type="text"
# 3. Group by and get max, min, count of a column in the dataset - SQL

# + id="wn7NSonVF00G" colab_type="code" outputId="0b3404fa-1d8e-4cf2-b384-20f620d81be5" colab={"base_uri": "https://localhost:8080/", "height": 663}
# Using groupby for median_income and max and min functions of SQL

task_3 = "select median_income, count(median_income) from df_sql group by median_income"
spark.sql(task_3).show()

task_3_1 = "select MAX(median_income) as max_median_income from df_sql"
spark.sql(task_3_1).show()

task_3_2 = "select MIN(median_income) as min_median_income from df_sql"
spark.sql(task_3_2).show()

# + [markdown] id="EuLNqek45Zpe" colab_type="text"
# 4. Filter your dataset by some conditions based on your column - pyspark.sql.dataframe

# + [markdown] id="Oe7oqiwgwrYL" colab_type="text"
# Analysis on a dataset based on condition is very helpful to extract specific information from the data. Let's have a look at one such conditional extraction.

# + id="VGQkqXo94mmu" colab_type="code" outputId="c3baf57e-ac05-4597-e53c-9123ceced592" colab={"base_uri": "https://localhost:8080/", "height": 459}
# As seen above, the lowest is 0.49 and highest is 9.8. Let's look at incomes > 3 using pyspark

df.filter(df['median_income'] > 3).show()

# + [markdown] id="osjAfXZqG6Yg" colab_type="text"
# 4. Filter your dataset by some conditions based on your column - SQL

# + id="vhd6yen3HBHA" colab_type="code" outputId="03461fbb-2ee3-449a-fadd-2754090ac8d6" colab={"base_uri": "https://localhost:8080/", "height": 459}
# Looking at the data with income > 3.0 using SQL

task_4 = "select * from df_sql where median_income > 3"
spark.sql(task_4).show()

# + [markdown] id="eLTyOX4M5glS" colab_type="text"
# 5. Apply order by - pyspark.sql.dataframe

# + [markdown] id="zPU-4SO5vGcG" colab_type="text"
# Ordering data based on a column values

# + id="qPAm1ZKJ5z57" colab_type="code" outputId="22dbfe82-c05d-41b4-f5db-2246054c7728" colab={"base_uri": "https://localhost:8080/", "height": 459}
# Ordering the data based on the house values using pyspark

df.select('total_rooms', 'total_bedrooms', 'median_house_value').orderBy('median_house_value').show()

# + [markdown] id="on2QuFJGHX85" colab_type="text"
# 5. Apply order by - SQL

# + id="eOnSmDmCHa8q" colab_type="code" outputId="65bb2675-2bef-4b46-f4ee-1f40c5290aef" colab={"base_uri": "https://localhost:8080/", "height": 459}
# Ordering the data by mdeian house values using SQL's order by

task_5 = "select total_rooms, total_bedrooms, median_house_value from df_sql order by median_house_value"
spark.sql(task_5).show()

# + [markdown] id="AgRqUlIf6Y6b" colab_type="text"
# 6. Select distinct records by a column - pyspark.sql.dataframe

# + [markdown] id="qNCrDKXKwSJJ" colab_type="text"
# Getting unique values for a column can be useful to find the non-repeating values. We will explore distinct values of the median incomes for our data.

# + id="YadebFYM6DTw" colab_type="code" outputId="7258abea-5674-4f05-e998-fc04462db136" colab={"base_uri": "https://localhost:8080/", "height": 459}
# All distinct incomes using pyspark

df.select('median_income').distinct().show()

# + [markdown] id="cOUILcCrIvRM" colab_type="text"
# 6. Select distinct records by a column - SQL

# + id="gV9WQ0xOI--C" colab_type="code" outputId="d0b0c632-445d-4d47-e92e-8a49370370e0" colab={"base_uri": "https://localhost:8080/", "height": 459}
# All distinct incomes using SQL

task_6 = "select distinct median_income from df_sql"
spark.sql(task_6).show()

# + [markdown] id="VB3F12pi72V5" colab_type="text"
# 7. Transform the data type of column from string to float - pyspark.sql.dataframe

# + [markdown] id="SjATc792s9z-" colab_type="text"
# In this data, all columns are of string datatype, but as seen in the description, we have float values in various columns as well, so we can cast the type of that column from string to integer or float.

# + id="7dISfeGl7fr3" colab_type="code" outputId="5328f900-3bab-4960-cd1c-4d09a4640b62" colab={"base_uri": "https://localhost:8080/", "height": 204}
# Casting of column datatype from string to float using pyspark

df_n = df.withColumn('median_income', col('median_income').cast((FloatType())))
df_n.printSchema()

# + [markdown] id="LM0cvFi9Jd5A" colab_type="text"
# 7. Transform the data type of column from string to float - SQL

# + id="mAcxU_Z9JkYv" colab_type="code" outputId="85b1c127-380c-4f95-babc-7f87f167df20" colab={"base_uri": "https://localhost:8080/", "height": 34}
# Type casting in SQL

task_7 = "select CAST(median_income as FLOAT) from df_sql"
task_7_column = spark.sql(task_7)

task_7_column.dtypes

# + [markdown] id="JDcZfLPZ-GF3" colab_type="text"
# 8. Apply group by with having clause - pyspark.sql.dataframe

# + [markdown] id="0HEFREgTv3se" colab_type="text"
# Grouping a column value (categorical) and sum of population based on those values

# + id="9qMs-vTB9ChM" colab_type="code" outputId="2c049af1-74cb-4bb2-ae3c-887a02161bf4" colab={"base_uri": "https://localhost:8080/", "height": 459}
# Groupby with having using pyspark

df.groupBy('median_income').agg(sum('population')).alias('pop_income').show()

# + [markdown] id="hSDtY1GeRp8q" colab_type="text"
# 8. Apply group by with having clause - SQL

# + id="kISf2Y6b_HJE" colab_type="code" outputId="29352db1-c930-4549-ee43-76afeb50ed17" colab={"base_uri": "https://localhost:8080/", "height": 459}
# Groupby with having using SQL

task_8 = 'SELECT median_income, sum(population) from df_sql GROUP BY median_income HAVING sum(population) > 660'
spark.sql(task_8).show()

StarcoderdataJupyterScriptsDedupFiltered