# ---
# 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="zH9NLWD13EjB"
# Install the dependencies
# !apt-get install openjdk-8-jdk-headless -qq > /dev/null
# !wget -q https://archive.apache.org/dist/spark/spark-3.0.1/spark-3.0.1-bin-hadoop3.2.tgz
# !tar xf spark-3.0.1-bin-hadoop3.2.tgz
# !pip install -q findspark

# + id="c6Cokct83h5R"
# Set the environment variables for running PySpark in the collaboration environmentimport os
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.0.1-bin-hadoop3.2"

# + id="WaccJdNb4hol"
# Run the local session to test the installation
import findspark
findspark.init('spark-3.0.1-bin-hadoop3.2')
from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local[*]').getOrCreate()

# + colab={"base_uri": "https://localhost:8080/", "height": 216} id="Uq4KARAb0_AR" outputId="34ca97ea-b4b9-4fdf-a860-6c6117361232"
spark

# + id="N6ymCYUi2kO_"
df = (spark.read
          .format("csv")
          .option('header', 'true')
          .load("drive/MyDrive/db/iris_frame.csv"))

# + colab={"base_uri": "https://localhost:8080/"} id="HD4TXz5f5XQG" outputId="ac8cb1bc-f59a-493b-e9af-c1e2438cd2ae"
df.show(5)

# + colab={"base_uri": "https://localhost:8080/", "height": 419} id="Z6fRlGq65ky2" outputId="b0c59f19-aa21-4bd2-85c1-125a664f669c"
df.toPandas()

# + colab={"base_uri": "https://localhost:8080/"} id="PdyoYZH_504N" outputId="ab8cd114-343f-4bad-c1e7-4077ce1b5df4"
# How many rows we have
df.count()

# + colab={"base_uri": "https://localhost:8080/"} id="RaiphMRh55sF" outputId="e4744f99-3b27-44e4-b790-d91d0a9f0bec"
# The names of our columns
df.columns

# + colab={"base_uri": "https://localhost:8080/"} id="pRthBq4y6Dki" outputId="d0aa39b8-5573-42da-b3cc-8f0c9ab66793"
# Types of our columns
df.dtypes

# + colab={"base_uri": "https://localhost:8080/", "height": 204} id="TBUk7rJM6OAO" outputId="8ef04431-54f5-4f96-b25e-ca9fa123da15"
# Basics stats from our columns
df.describe().toPandas()

# + colab={"base_uri": "https://localhost:8080/"} id="-O0ZCXh16S3W" outputId="3ec987f0-2ad7-47c1-d113-218e4d7527cf"
from pyspark.sql.functions import col
dataset = df.select(col('sepal length (cm)').cast('float'),
                         col('sepal width (cm)').cast('float'),
                         col('petal length (cm)').cast('float'),
                         col('petal width (cm)').cast('float'),
                         col('target').cast('integer')            
                        )
dataset.show()

# + colab={"base_uri": "https://localhost:8080/"} id="zJc0dsuD7U6u" outputId="ed0fed49-a69c-4879-8b56-33581758854a"
dataset.dtypes

# + id="p3uVLZBa8PIl"
# Assemble all the features with VectorAssembler
required_features = ['sepal length (cm)',
                    'sepal width (cm)',
                    'petal length (cm)',
                    'petal width (cm)'
                   ]
from pyspark.ml.feature import VectorAssembler
assembler = VectorAssembler(inputCols=required_features, outputCol='features')
transformed_data = assembler.transform(dataset)

# + colab={"base_uri": "https://localhost:8080/"} id="Hes21iNV82XW" outputId="4ea76411-3494-47a8-e0f7-30d99a22748e"
transformed_data.show(10)

# + id="a49bETgf9Im1"
# Modeling
(training_data, test_data) = transformed_data.randomSplit([0.8,0.2])

# + id="FgIjr9dX9Yaz"
from pyspark.ml.classification import RandomForestClassifier
rf = RandomForestClassifier(labelCol='target', featuresCol='features', maxDepth=5)

# + id="CKf-wAM89iib"
model = rf.fit(training_data)

# + id="qq43-X289nC8"
predictions = model.transform(test_data)

# + id="TAmO5iSJ9xJf"
# Evaluate our model
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluator_accuracy = MulticlassClassificationEvaluator(
    labelCol='target', 
    predictionCol='prediction', 
    metricName='accuracy')

# + colab={"base_uri": "https://localhost:8080/"} id="pghaVdy493et" outputId="55a633dc-3dad-4736-a83b-d2cdce4ccf8e"
accuracy = evaluator_accuracy.evaluate(predictions)
print('Test Accuracy = ', accuracy)

# + id="adJn93diBoVr"
evaluator_f1 = MulticlassClassificationEvaluator(
    labelCol='target', 
    predictionCol='prediction', 
    metricName='f1')

# + colab={"base_uri": "https://localhost:8080/"} id="38028aHPCMKc" outputId="13a16cf6-9bde-4ad2-be2e-9a8760226b15"
f1 = evaluator_f1.evaluate(predictions)
print('Test f1 = ', f1)

StarcoderdataJupyterScriptsDedupFiltered