<jupyter_start><jupyter_text><jupyter_code>from google.colab import drive
drive.mount('/content/drive')
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q https://www-us.apache.org/dist/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()
# import library wiche we need
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from pyspark.ml.classification import NaiveBayes, DecisionTreeClassifier, LogisticRegression, RandomForestClassifier
from pyspark.sql import SparkSession
from pyspark.ml import feature, evaluation
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator
from pyspark.mllib.evaluation import BinaryClassificationMetrics
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.util import MLWriter
from sklearn.preprocessing import StandardScaler 
import statsmodels.api as sm
from pyspark.sql.functions import col, when
import re
from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit
from pyspark.ml import Pipeline

from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.regression import LinearRegression
from pyspark.ml.tuning import ParamGridBuilder, TrainValidationSplit

from pyspark import SparkContext, SparkConf

spark = SparkSession \
    .builder \
    .appName("pass") \
    .getOrCreate()
    
df_train = spark.read.csv("/content/drive/My Drive/data_sets/train.csv",header=True, inferSchema=True)

df_train = df_train.drop('PassengerId', 'Name', 'Ticket', 'Cabin')

df_train = df_train.dropna(subset = ['Age', 'Embarked'])

convert_cols = ['Sex', 'Embarked']
for co in convert_cols:
    label_indexer = feature.StringIndexer(inputCol=co, outputCol=co + 'C').fit(df_train)
    df_train = label_indexer.transform(df_train).drop(co)

feature_cols_train = [df_train.columns[2], df_train.columns[3], df_train.columns[5], df_train.columns[6]]

assembler = feature.VectorAssembler(inputCols=feature_cols_train, outputCol='features')
df_train = assembler.setHandleInvalid("skip").transform(df_train)
df_train = df_train.withColumnRenamed('Survived', 'label')
df_train = df_train.select('features', 'label')

scaler = feature.StandardScaler(inputCol="features", outputCol="scaledFeatures",
                        withStd=True, withMean=False)
scaler = scaler.fit(df_train)
df_train =scaler.transform(df_train)
df_train = df_train.drop('features').withColumnRenamed('scaledFeatures','features')

dt = DecisionTreeClassifier(labelCol="label", featuresCol="features")
pipeline = Pipeline(stages=[dt])  
model = pipeline.fit(df_train)
outpath = "./model2"
model.save(outpath)



df_test = spark.read.csv("/content/drive/My Drive/data_sets/kafka_test.csv",header=True, inferSchema=True)
df_test = df_test.drop('PassengerId', 'Name', 'Ticket', 'Cabin')
df_test = df_test.dropna(subset = ['Age', 'Embarked'])

convert_cols = ['Sex', 'Embarked']
for co in convert_cols:
    label_indexer = feature.StringIndexer(inputCol=co, outputCol=co + 'C').fit(df_test)
    df_test = label_indexer.transform(df_test).drop(co)

feature_cols_train = [df_test.columns[2], df_test.columns[3], df_test.columns[5], df_test.columns[6]]

assembler = feature.VectorAssembler(inputCols=feature_cols_train, outputCol='features')
df_test = assembler.setHandleInvalid("skip").transform(df_test)
df_test = df_test.withColumnRenamed('Survived', 'label')
df_test = df_test.select('features', 'label')

scaler = feature.StandardScaler(inputCol="features", outputCol="scaledFeatures",
                        withStd=True, withMean=False)
scaler = scaler.fit(df_test)
df_test =scaler.transform(df_test)
df_test = df_test.drop('features').withColumnRenamed('scaledFeatures','features')
from pyspark.ml import PipelineModel
persistedModel = PipelineModel.load('/content/model2')

# predict
pred = persistedModel.transform(df_test)
evaloator = evaluation.MulticlassClassificationEvaluator(metricName='accuracy')
pred.show(10)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12458313482610767, 'fraction_numerical': 0.008099094807050978, 'mean_word_length': 6.47153024911032, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 6, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '593086', 'license': 'MIT', 'n_tokens_mistral': 1445, 'n_tokens_neox': 1408, 'n_words': 255, 'path': 'HW_Save&Load_8.ipynb', 'repo_name': 'amirsartipi13/intro-python-kafka', 'size': '4198'}