<jupyter_start><jupyter_text>Read train and test data<jupyter_code>%%time
train_df = spark.read.csv('data/fraudTrain.csv', header=True)
test_df = spark.read.csv('data/fraudTest.csv', header=True)
train_df.show(5)<jupyter_output>+---+---------------------+----------------+--------------------+-------------+------+---------+-------+------+--------------------+--------------+-----+-----+-------+---------+--------+--------------------+----------+--------------------+----------+------------------+-----------+--------+
|_c0|trans_date_trans_time|          cc_num|            merchant|     category|   amt|    first|   last|gender|              street|          city|state|  zip|    lat|     long|city_pop|                 job|       dob|           trans_num| unix_time|         merch_lat| merch_long|is_fraud|
+---+---------------------+----------------+--------------------+-------------+------+---------+-------+------+--------------------+--------------+-----+-----+-------+---------+--------+--------------------+----------+--------------------+----------+------------------+-----------+--------+
|  0|  2019-01-01 00:00:18|2703186189652095|fraud_Rippin, Kub...|     misc_net|  4.97| Jennifer|  Banks|     F|      561 Perry [...]<jupyter_text>Transform data using transformation pipeline<jupyter_code>get_age_fn = get_age_from_dob(input_col="dob", output_col="age")
get_time_fn = get_time(input_col="trans_date_trans_time", output_col="time")
amt_int_fn = cast_int(input_col="amt",output_col="amt")
city_pop_int_fn = cast_int(input_col="city_pop",output_col="city_pop")
is_fraud_int_fn = cast_int(input_col="is_fraud",output_col="is_fraud")

transformer_pipeline = Pipeline(stages=[get_age_fn,get_time_fn,amt_int_fn,city_pop_int_fn,is_fraud_int_fn])
%%time
transformer_model = transformer_pipeline.fit(train_df)
train_df = transformer_model.transform(train_df)
test_df = transformer_model.transform(test_df)
train_data = train_df.select(col("time"), col("amt"), col("city_pop"), col("age"), col("is_fraud"))
test_data = test_df.select(col("time"), col("amt"), col("city_pop"), col("age"), col("is_fraud"))
train_data.show()
train_data.printSchema()<jupyter_output>+----+---+--------+---+--------+
|time|amt|city_pop|age|is_fraud|
+----+---+--------+---+--------+
|   0|  4|    3495| 33|       0|
|   0|107|     149| 43|       0|
|   0|220|    4154| 59|       0|
|   0| 45|    1939| 54|       0|
|   0| 41|      99| 35|       0|
|   0| 94|    2158| 60|       0|
|   0| 44|    2691| 28|       0|
|   0| 71|    6018| 74|       0|
|   0|  4|    1472| 80|       0|
|   0|198|  151785| 47|       0|
|   0| 24|    7297| 31|       0|
|   0|  7|    1925| 55|       0|
|   0| 71|  341043| 32|       0|
|   0| 96|     589| 76|       0|
|   0|  7|     899| 54|       0|
|   0|  3|    4664| 56|       0|
|   0|327|    1078| 69|       0|
|   0|341|    4081| 83|       0|
|   0| 63|    2518| 75|       0|
|   0| 44|  124967| 41|       0|
+----+---+--------+---+--------+
only showing top 20 rows

root
 |-- time: integer (nullable = true)
 |-- amt: integer (nullable = true)
 |-- city_pop: integer (nullable = true)
 |-- age: integer (nullable = true)
 |-- is_fraud: integer (nul[...]<jupyter_text>PySpark Model  Create features for training model<jupyter_code>feature_columns = ['time', 'amt', 'city_pop', 'age']
assembler = VectorAssembler(inputCols=feature_columns, outputCol='features')
train_data = assembler.transform(train_data)
test_data = assembler.transform(test_data)<jupyter_output><empty_output><jupyter_text>Train<jupyter_code>rf = RandomForestClassifier(maxDepth=3,numTrees=10,labelCol="is_fraud",featuresCol='features', seed=6)
%%time
rf_model_spark = rf.fit(train_data)<jupyter_output><jupyter_text>Predictions<jupyter_code>%%time
predictions = rf_model_spark.transform(test_data)
%%time
correct = predictions.where(col('prediction')==col('is_fraud')).count()
total = predictions.count()

print("Accuracy: ", correct/total)<jupyter_output><jupyter_text>Load and broadcast a previously trained RF model<jupyter_code>model_data = sc.binaryFiles("rf_model.pkl").collect()
rf_model = pickle.loads(model_data[0][1])
rf_model_bc = sc.broadcast(rf_model)

def predict_fn(*cols):
    cols = np.array(cols).reshape((1,-1))
    prediction = rf_model_bc.value.predict_proba((cols))
    return int( float(prediction[0,1]) > 0.5 )

predict = udf(predict_fn, IntegerType())<jupyter_output><empty_output><jupyter_text>Predictions<jupyter_code>%%time
df = test_data.withColumn("score", predict(*feature_columns))
df.show()
%%time
correct = df.filter(col('score')==col('is_fraud')).count()
total = df.count()

print(correct/total)<jupyter_output><jupyter_text>Caching<jupyter_code>train_data.cache()
test_data.cache()<jupyter_output><empty_output><jupyter_text>Re-train Pyspark model<jupyter_code>%%time
rf_model_spark = rf.fit(train_data)<jupyter_output><jupyter_text>Predictions<jupyter_code>%%time
predictions = rf_model_spark.transform(test_data)
%%time
correct = predictions.where(col('prediction')==col('is_fraud')).count()
total = predictions.count()

print("Accuracy: ", correct/total)<jupyter_output><jupyter_text>Predictions for broadcasted model<jupyter_code>%%time
df = test_data.withColumn("score", predict(*feature_columns))
df.show()
%%time
correct = df.filter(col('score')==col('is_fraud')).count()
total = df.count()

print(correct/total)<jupyter_output><jupyter_text>Caching the previously trained model and making predictions on cached data<jupyter_code>model_pkl = sc.binaryFiles("rf_model.pkl")
model_pkl.cache()
model_data = model_pkl.collect()
rf_model = pickle.loads(model_data[0][1])
rf_model_bc = sc.broadcast(rf_model)

def predict_fn(*cols):
    cols = np.array(cols).reshape((1,-1))
    prediction = rf_model_bc.value.predict_proba((cols))
    return int( float(prediction[0,1]) > 0.5 )

predict = udf(predict_fn, IntegerType())
%%time
df = test_data.withColumn("score", predict(*feature_columns))
df.show()
%%time
correct = df.filter(col('score')==col('is_fraud')).count()
total = df.count()

print(correct/total)
# train_data.unpersist()
# test_data.unpersist()<jupyter_output><empty_output><jupyter_text>Evaluation of pyspark model<jupyter_code>from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.mllib.evaluation import MulticlassMetrics
from pyspark.sql.types import FloatType
import pyspark.sql.functions as F
from itertools import chain
predictions = rf_model_spark.transform(test_data)
evaluator = BinaryClassificationEvaluator(labelCol="is_fraud", rawPredictionCol="rawPrediction")
accuracy = evaluator.evaluate(predictions)
print("Accuracy = %s" % (accuracy))
print("Test Error = %s" % (1.0 - accuracy))


preds_and_labels = predictions.select(['prediction','is_fraud']).withColumn('is_fraud', F.col('is_fraud').cast(FloatType())).orderBy('prediction')
preds_and_labels = preds_and_labels.select(['prediction','is_fraud'])
metrics = MulticlassMetrics(preds_and_labels.rdd.map(tuple))
print(metrics.confusionMatrix().toArray())<jupyter_output>Accuracy = 0.9119739259417061
Test Error = 0.08802607405829388<jupyter_text>Weighted training<jupyter_code>y_collect = train_data.select("is_fraud").groupBy("is_fraud").count().collect()
unique_y = [x["is_fraud"] for x in y_collect]
total_y = sum([x["count"] for x in y_collect])
unique_y_count = len(y_collect)
bin_count = [x["count"] for x in y_collect]

class_weights_spark = {i: ii for i, ii in zip(unique_y, total_y / (unique_y_count * np.array(bin_count)))}
print(class_weights_spark) 

mapping_expr = F.create_map([F.lit(x) for x in chain(*class_weights_spark.items())])
train_data = train_data.withColumn("weight", mapping_expr.getItem(F.col("is_fraud")))
%%time
rf_weight = RandomForestClassifier(maxDepth=3,numTrees=10,labelCol="is_fraud",featuresCol='features',weightCol="weight", seed=6)
rf_model_weight = rf_weight.fit(train_data)
predictions = rf_model_weight.transform(test_data)
evaluator = BinaryClassificationEvaluator(labelCol="is_fraud", rawPredictionCol="rawPrediction")
accuracy = evaluator.evaluate(predictions)
print("Accuracy = %s" % (accuracy))
print("Test Error = %s" % (1.0 - accuracy))


preds_and_labels = predictions.select(['prediction','is_fraud']).withColumn('is_fraud', F.col('is_fraud').cast(FloatType())).orderBy('prediction')
preds_and_labels = preds_and_labels.select(['prediction','is_fraud'])
metrics = MulticlassMetrics(preds_and_labels.rdd.map(tuple))
print(metrics.confusionMatrix().toArray())
import matplotlib.pyplot as plt
N=3
ind = np.arange(N) 
width = 0.35 

non_cache = [9.88, 0.0452, 29.8]
cache = [2.51, 0.0436, 30.3]

plt.figure(figsize=(10,6))
plt.bar(ind, non_cache, width, label='Non-Cache')
plt.bar(ind + width, cache, width,
    label='Cache')
plt.yscale("log")
plt.xticks(ind + width / 2, ('Pyspark train','Pyspark pred','Broadcast pred'))
plt.ylabel("Time (s)")
plt.legend(loc='best')
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.21900686859587884, 'fraction_numerical': 0.03772097736741358, 'mean_word_length': 3.9180509413067552, 'pattern_counts': {'":': 0, '<': 48, '<?xml version=': 0, '>': 50, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '664200', 'license': 'MIT', 'n_tokens_mistral': 3433, 'n_tokens_neox': 3190, 'n_words': 498, 'path': 'Spark_model.ipynb', 'repo_name': 'sharriy/spark-201-project', 'size': '8881'}