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

# # Projeto Big Data
#
# <NAME> <br/>
# <NAME><br/>
# <NAME>
#
# Dataset: https://www.kaggle.com/mlg-ulb/creditcardfraud

import findspark
import pyspark
from pyspark.conf import SparkConf
from pyspark.sql.functions import monotonically_increasing_id
from pyspark.mllib.linalg import Vectors
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.tree import RandomForest as  cls_rf
from time import *
from pyspark.mllib.evaluation import BinaryClassificationMetrics

# # Utilizando o contexto do spark

findspark.init()
sc = pyspark.SparkContext(appName="projeto")
session = pyspark.sql.SparkSession.builder.config(conf=SparkConf())
spark = session.getOrCreate()

# # Carregando o dataset do hdfs do hadoop

df = spark.read.options(header = "true", inferschema = "true").csv('hdfs://cluster-001-m/user/victor_outtes/creditcard.csv')
df.printSchema()
print("Total number of rows:", df.count())

# # Split dataset 70-30

# +
TRAIN_DATA_RATIO = 0.7

# A ultima coluna contém o target
transformed_df = df.rdd.map(lambda row: LabeledPoint(row[-1], Vectors.dense(row[0:-1])))

# Dividindo o dataset
splits = [TRAIN_DATA_RATIO, 1.0 - TRAIN_DATA_RATIO]
train_data, test_data = transformed_df.randomSplit(splits, RANDOM_SEED)
print("Quantidade de linhas do dataset de treinamento: %d" % train_data.count())
print("Quantidade de linhas do dataset de testes: %d" % test_data.count())
# -

# # Treino e parametrização do modelo random forest

# +
RANDOM_SEED = 13579
NUM_TREES = 3
MAX_DEPTH = 4
MAX_BINS = 32
t1 = time()
# Treinando a random forest
model = cls_rf.trainClassifier(train_data, numClasses=2, categoricalFeaturesInfo={}, numTrees=NUM_TREES, maxDepth=MAX_DEPTH, maxBins=MAX_BINS, seed=RANDOM_SEED)

t2 = time()
t_end = t2 - t1
print("Tempo do treinamento: %.3f s" % t_end)
# -

# # Predição e medição da acurácia

predictions = model.predict(test_data.map(lambda x: x.features))
labels_and_predictions = test_data.map(lambda x: x.label).zip(predictions)
accuracy = labels_and_predictions.filter(lambda x: x[0] == x[1]).count() / float(test_data.count())
print("Model accuracy: %.3f%%" % (accuracy * 100))

# # Métricas: precision recall / curva roc

metrics = BinaryClassificationMetrics(labels_and_predictions)
print("Area PR curve: %.f" % (metrics.areaUnderPR * 100))
print("Area ROC curve: %.3f" % (metrics.areaUnderROC * 100))

StarcoderdataJupyterScriptsDedupFiltered