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

# # Reference doc
# [PySpark을 이용한 머신러닝 튜토리얼 예제](https://www.sqler.com/board_MachineLearning_AI_tip_lecture/1102607)

# Create a SparkContext
import pyspark
from pyspark import SparkContext
sc = SparkContext()

# RDD, Resilient Distributed Dataset
# Computation in an RDD is automatically parallelized across the cluster
nums = sc.parallelize([1,2,3,4])
nums.take(1)

squared = nums.map(lambda x: x*x).collect()
for num in squared:
    print('%i ' % (num))

# +
# SQLContext
from pyspark.sql import Row
from pyspark.sql import SQLContext

sqlContext = SQLContext(sc)
# -

list_p = [('John',19),('Smith',29),('Adam',35),('Henry',50)]
rdd = sc.parallelize(list_p)
ppl = rdd.map(lambda x: Row(name=x[0], age=int(x[1])))
DF_ppl = sqlContext.createDataFrame(ppl)
DF_ppl.printSchema()

# +
# Machine Learning Example with PySpark

# +
from pyspark import SparkFiles

url = "https://raw.githubusercontent.com/guru99-edu/R-Programming/master/adult_data.csv"
sc.addFile(url)
sqlContext = SQLContext(sc)
# -

df = sqlContext.read.csv(SparkFiles.get("adult_data.csv"), header=True, inferSchema= True)
df.printSchema()

df.show(5, truncate = False)

df_string = sqlContext.read.csv(SparkFiles.get("adult_data.csv"), header=True, inferSchema=  False)
df_string.printSchema()

# +
# Import all from `sql.types`
# from pyspark.sql.types import *
from pyspark.sql.types import FloatType

# Write a custom function to convert the data type of DataFrame columns
def convertColumn(df, names, newType):
    for name in names: 
        df = df.withColumn(name, df[name].cast(newType))
    return df 
# List of continuous features
CONTI_FEATURES  = ['age', 'fnlwgt','capital-gain', 'educational-num', 'capital-loss', 'hours-per-week']
# Convert the type
df_string = convertColumn(df_string, CONTI_FEATURES, FloatType())
# Check the dataset
df_string.printSchema()
# -

df.select('age','fnlwgt').show(5)

df.groupBy("education").count().sort("count",ascending=True).show()

df.describe().show()

df.describe('capital-gain').show()

df.crosstab('age', 'income').sort("age_income").show()

df.drop('education_num').columns

df.filter(df.age > 40).count()

df.groupby('marital-status').agg({'capital-gain': 'mean'}).show()

# +
# Step 2) Data preprocessing
from pyspark.sql.functions import col, asc

# 1 Select the column
age_square = df.select(col("age")**2)

# 2 Apply the transformation and add it to the DataFrame
df = df.withColumn("age_square", col("age")**2)
df.printSchema()
# -

df.first()



df.filter(df['native-country'] == 'Holand-Netherlands').count()


df.groupby('native-country').agg({'native-country': 'count'}).sort(asc("count(native-country)")).show()

df_remove = df.filter(df['native-country'] != 'Holand-Netherlands')
df_remove.groupby('native-country').agg({'native-country': 'count'}).sort(asc("count(native-country)")).show()

# +
# Step 3) Build a data processing pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler

stringIndexer = StringIndexer(inputCol="workclass", outputCol="workclass_encoded")
model = stringIndexer.fit(df)
indexed = model.transform(df)
encoder = OneHotEncoder(dropLast=False, inputCol="workclass_encoded", outputCol="workclass_vec")
# encoded = encoder.transform(indexed)  # error
ohe = encoder.fit(indexed)
encoded = ohe.transform(indexed)
encoded.show(2)

# error 
# AttributeError: 'OneHotEncoder' object has no attribute 'transform'
# https://stackoverflow.com/questions/64011674/onehotencoder-object-has-no-attribute-transform

# +
from pyspark.ml import Pipeline
# from pyspark.ml.feature import OneHotEncoderEstimator  # error
from pyspark.ml.feature import OneHotEncoder  # error

CATE_FEATURES = ['workclass', 'education', 'marital-status', 'occupation', 'relationship', 'race', 'gender', 'native-country']
stages = [] # stages in our Pipeline
for categoricalCol in CATE_FEATURES:
    stringIndexer = StringIndexer(inputCol=categoricalCol, outputCol=categoricalCol + "Index")
    encoder = OneHotEncoder(inputCols=[stringIndexer.getOutputCol()],
                                     outputCols=[categoricalCol + "classVec"])
    stages += [stringIndexer, encoder]

# error
# ImportError: cannot import name 'OneHotEncoderEstimator' from 'pyspark.ml.feature'
# https://stackoverflow.com/questions/59926511/pyspark-cannot-import-name-onehotencoderestimator
# -

# Convert label into label indices using the StringIndexer
label_stringIdx =  StringIndexer(inputCol="income", outputCol="newlabel")
stages += [label_stringIdx]

assemblerInputs = [c + "classVec" for c in CATE_FEATURES] + CONTI_FEATURES

assembler = VectorAssembler(inputCols=assemblerInputs, outputCol="features")
stages += [assembler]

# Create a Pipeline.
pipeline = Pipeline(stages=stages)
pipelineModel = pipeline.fit(df_remove)
model = pipelineModel.transform(df_remove)
model.take(1)

# Step 4) Build the classifier: logistic
from pyspark.ml.linalg import DenseVector
input_data = model.rdd.map(lambda x: (x["newlabel"], DenseVector(x["features"])))

df_train = sqlContext.createDataFrame(input_data, ["income", "features"])
df_train.show(5)

train_data, test_data = df_train.randomSplit([.8,.2],seed=1234)

train_data.groupby('income').agg({'income': 'count'}).show()

test_data.groupby('income').agg({'income': 'count'}).show()

# +
# Build the logistic regressor
# Import `LinearRegression`
from pyspark.ml.classification import LogisticRegression

# Initialize `lr`
lr = LogisticRegression(labelCol="income",
                        featuresCol="features",
                        maxIter=10,
                        regParam=0.3)

# Fit the data to the model
linearModel = lr.fit(train_data)

# -

# Print the coefficients and intercept for logistic regression
print("Coefficients: " + str(linearModel.coefficients))
print("Intercept: " + str(linearModel.intercept))

# Step 5) Train and evaluate the model
# Make predictions on test data using the transform() method.
predictions = linearModel.transform(test_data)

predictions.printSchema()

selected = predictions.select("income", "prediction", "probability")
selected.show(20)

# Evaluate the model
cm = predictions.select("income", "prediction")
cm.groupby('income').agg({'income': 'count'}).show()

cm.groupby('prediction').agg({'prediction': 'count'}).show()

cm.filter(cm.income == cm.prediction).count() / cm.count()


# +
def accuracy_m(model): 
    predictions = model.transform(test_data)
    cm = predictions.select("income", "prediction")
    acc = cm.filter(cm.income == cm.prediction).count() / cm.count()
    print("Model accuracy: %.3f%%" % (acc * 100)) 

accuracy_m(model = linearModel)

# +
# ROC metrics
### Use ROC 
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# Evaluate model
evaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="income")
print(evaluator.evaluate(predictions))
print(evaluator.getMetricName())

# error 
# IllegalArgumentException: label does not exist. Available: income, features, rawPrediction, probability, prediction
# https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.evaluation.BinaryClassificationEvaluator.html
# set label column

# +
# Step 6) Tune the hyperparameter
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator

# Create ParamGrid for Cross Validation
paramGrid = (ParamGridBuilder()
             .addGrid(lr.regParam, [0.01, 0.5])
             .build())

# +
from time import time
start_time = time()

# Create 5-fold CrossValidator
cv = CrossValidator(estimator=lr,
                    estimatorParamMaps=paramGrid,
                    evaluator=evaluator, numFolds=5)

# Run cross validations
cvModel = cv.fit(train_data)
# likely take a fair amount of time
end_time = time()
elapsed_time = end_time - start_time
print("Time to train model: %.3f seconds" % elapsed_time)
# -

accuracy_m(model = cvModel)

bestModel = cvModel.bestModel
bestModel.extractParamMap()



StarcoderdataJupyterScriptsDedupFiltered