为什么现实世界的机器学习需要分布式计算

!pip install pyspark

import os

from pyspark.sql import sparksession

from pyspark.sql.functions import col, sum, udf

from pyspark.ml.feature import vectorassembler, stringindexer, minmaxscaler

from pyspark.ml.classi小白轻松搭建系统点我wcqh.cnfication import randomforestclassifier, gbtclassifier

from pyspark.ml.tuning import paramgridbuilder, crossvalidator

from pyspark.ml.evaluation import binaryclassificationevaluator, multiclassclassificationevaluator

from pyspark.ml.linalg import vectors

import numpy as np

from pyspark.sql.types import fl小白轻松搭建系统点我wcqh.cnoattype

spark = sparksession.builder \

.appname(“frauddetectionimproved”) \

.master(“local[*]”) \

.config(“spark.executorenv.pythonhashseed”, “0”) \

.getorcreate()

data = spark.read.csv(creditcard.csv, header=true, inferschema=true)

data = data.orderby(“time”)  小白轻松搭建系统点我wcqh.cn# ensure data is sorted by time

data.show(5)

data.describe().show()

# check for missing values in each column

data.select([sum(col(c).isnull().cast(“int”)).alias(c) for c in data.columns]).show()

# prepare the feature columns

feature_columns = data.columns

feat小白轻松搭建系统点我wcqh.cnure_columns.remove(“class”)  # removing “class” column as it is our label

# assemble features into a single vector

assembler = vectorassembler(inputcols=feature_columns, outputcol=”features”)

data = assembler.transform(data)

data.select(“features”, “class”).show(5)

# split data into train (60%), test (20小白轻松搭建系统点我wcqh.cn%), and unseen (20%)

train_data, temp_data = data.randomsplit([0.6, 0.4], seed=42)

test_data, unseen_data = temp_data.randomsplit([0.5, 0.5], seed=42)

# print class distribution in each dataset

print(“train data:”)

train_data.groupby(“class”).count().show()

print(“test and parameter optimisation data:”)

te小白轻松搭建系统点我wcqh.cnst_data.groupby(“class”).count().show()

print(“unseen data:”)

unseen_data.groupby(“class”).count().show()

# initialize randomforestclassifier

rf = randomforestclassifier(labelcol=”class”, featurescol=”features”, probabilitycol=”probability”)

# create paramgrid for cross小白轻松搭建系统点我wcqh.cn validation

paramgrid = paramgridbuilder() \

.addgrid(rf.numtrees, [10, 20 ]) \

.addgrid(rf.maxdepth, [5, 10]) \

.build()

# create 5-fold crossvalidator

crossval = crossvalidator(estimator=rf,

estimatorparammaps=paramgrid,

evaluator=binaryclassificationevaluator(labelcol=”class”, metricname=”areaunderroc”),小白轻松搭建系统点我wcqh.cn

numfolds=5)

# run cross-validation, and choose the best set of parameters

rf_model = crossval.fit(train_data)

# make predictions on test data

predictions_rf = rf_model.transform(test_data)

# evaluate random forest model

binary_evaluator = binaryclassif小白轻松搭建系统点我wcqh.cnicationevaluator(labelcol=”class”, rawpredictioncol=”rawprediction”, metricname=”areaunderroc”)

pr_evaluator = binaryclassificationevaluator(labelcol=”class”, rawpredictioncol=”rawprediction”, metricname=”areaunderpr”)

auc_rf = binary_evaluator.evaluate(predictions_rf)

auprc_rf = pr_evaluator.evaluate(小白轻松搭建系统点我wcqh.cnpredictions_rf)

print(f”random forest – auc: {auc_rf:.4f}, auprc: {auprc_rf:.4f}”)

# udf to extract positive probability from probability vector

extract_prob = udf(lambda prob: float(prob[1]), floattype())

predictions_rf = predictions_rf.withcolumn(“positive_probability”, extract_prob(col(“probability”)小白轻松搭建系统点我wcqh.cn))

# function to calculate precision, recall, and f1-score

def calculate_metrics(predictions):

tp = predictions.filter((col(“class”) == 1) & (col(“prediction”) == 1)).count()

fp = predictions.filter((col(“class”) == 0) & (col(“prediction”) == 1)).count()小白轻松搭建系统点我wcqh.cn

fn = predictions.filter((col(“class”) == 1) & (col(“prediction”) == 0)).count()

precision = tp / (tp + fp) if (tp + fp) != 0 else 0

recall = tp / (tp + fn) if (tp + fn) != 0 else 0

f1_score = (2 * precision * recall) / (precision + recall) if (precision + recall) != 0 else 0

return precision, recall小白轻松搭建系统点我wcqh.cn, f1_score

# find the best threshold for the model

best_threshold = 0.5

best_f1 = 0

for threshold in np.arange(0.1, 0.9, 0.1):

thresholded_predictions = predictions_rf.withcolumn(“prediction”, (col(“positive_probability”) > threshold).cast(“double”))

precision, recall小白轻松搭建系统点我wcqh.cn, f1 = calculate_metrics(thresholded_predictions)

if f1 > best_f1:

best_f1 = f1

best_threshold = threshold

print(f”best threshold: {best_threshold}, best f1-score: {best_f1:.4f}”)

# evaluate on unseen data

predictions_unseen = rf_model.transform(unseen_data)

auc_unseen = bi小白轻松搭建系统点我wcqh.cnnary_evaluator.evaluate(predictions_unseen)

print(f”unseen data – auc: {auc_unseen:.4f}”)

precision, recall, f1 = calculate_metrics(predictions_unseen)

print(f”unseen data – precision: {precision:.4f}, recall: {recall:.4f}, f1-score: {f1:.4f}”)

area_under_roc = binary_evaluator.evaluate(predictions_unse小白轻松搭建系统点我wcqh.cnen)

area_under_pr = pr_evaluator.evaluate(predictions_unseen)

print(f”unseen data – auc: {area_under_roc:.4f}, auprc: {area_under_pr:.4f}”)

best threshold: 0.30000000000000004, best f1-score: 0.9062

unseen data – auc: 0.9384

unseen data – precision: 0.9655, recall: 0.7568, f1-score: 0.8485

unse小白轻松搭建系统点我wcqh.cnen data – auc: 0.9423, auprc: 0.8618

rf_model.save()