<jupyter_start><jupyter_text>Random Forest and Gradient Boosted Trees used for Fraud detection<jupyter_code>from pyspark.ml import Pipeline
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, countDistinct
from pyspark.sql.functions import col, explode, array, lit
# Import VectorAssembler and Vectors
from pyspark.ml.linalg import Vectors
from pyspark.ml.feature import VectorAssembler
spark = SparkSession.builder.appName('FraudTreeMethods').getOrCreate()<jupyter_output><empty_output><jupyter_text>Read data<jupyter_code># Load and parse the data file, converting it to a DataFrame.
#data = sqlContext.sql("SELECT * FROM fraud_train_sample")
data = spark.read.csv('train.csv', inferSchema=True, header=True)
data.show(5)<jupyter_output>+------+---+------+---+-------+-------------------+---------------+-------------+
|    ip|app|device| os|channel|         click_time|attributed_time|is_attributed|
+------+---+------+---+-------+-------------------+---------------+-------------+
| 83230|  3|     1| 13|    379|2017-11-06 14:32:21|           null|            0|
| 17357|  3|     1| 19|    379|2017-11-06 14:33:34|           null|            0|
| 35810|  3|     1| 13|    379|2017-11-06 14:34:12|           null|            0|
| 45745| 14|     1| 13|    478|2017-11-06 14:34:52|           null|            0|
|161007|  3|     1| 13|    379|2017-11-06 14:35:08|           null|            0|
+------+---+------+---+-------+-------------------+---------------+-------------+
only showing top 5 rows<jupyter_text>Count distinct values<jupyter_code>data.agg(*(countDistinct(col(c)).alias(c) for c in data.columns)).show()
data.groupBy('is_attributed').count().show()<jupyter_output>+-------------+---------+
|is_attributed|    count|
+-------------+---------+
|            1|   456846|
|            0|184447044|
+-------------+---------+<jupyter_text>Over sampling The imbalance ratio of 0 and 1 are calculated at the following and it is used for oversampling procedure.<jupyter_code>major_df = data.filter(col("is_attributed") == 0)
minor_df = data.filter(col("is_attributed") == 1)
ratio = int(major_df.count()/minor_df.count())
print("ratio: {}".format(ratio))<jupyter_output>ratio: 403<jupyter_text>The ratio is applied for oversampling at the following.<jupyter_code>a = range(ratio)
# duplicate the minority rows
oversampled_df = minor_df.withColumn("dummy", explode(array([lit(x) for x in a]))).drop('dummy')
# combine both oversampled minority rows and previous majority rows combined_df = major_df.unionAll(oversampled_df)
combined_df = major_df.unionAll(oversampled_df)
combined_df.show(3)<jupyter_output>+-----+---+------+---+-------+-------------------+---------------+-------------+
|   ip|app|device| os|channel|         click_time|attributed_time|is_attributed|
+-----+---+------+---+-------+-------------------+---------------+-------------+
|83230|  3|     1| 13|    379|2017-11-06 14:32:21|           null|            0|
|17357|  3|     1| 19|    379|2017-11-06 14:33:34|           null|            0|
|35810|  3|     1| 13|    379|2017-11-06 14:34:12|           null|            0|
+-----+---+------+---+-------+-------------------+---------------+-------------+
only showing top 3 rows<jupyter_text>The data is combined at the following and is presented.<jupyter_code>combined_df.groupBy('is_attributed').count().show()<jupyter_output>+-------------+---------+
|is_attributed|    count|
+-------------+---------+
|            1|184108938|
|            0|184447044|
+-------------+---------+<jupyter_text>Replace the new data (combined_df) into the old data.<jupyter_code>data = combined_df
data.groupBy('is_attributed').count().show()
data.agg(*(countDistinct(col(c)).alias(c) for c in data.columns)).show()<jupyter_output>+------+---+------+---+-------+----------+---------------+-------------+
|    ip|app|device| os|channel|click_time|attributed_time|is_attributed|
+------+---+------+---+-------+----------+---------------+-------------+
|277396|706|  3475|800|    202|    259620|         182057|            2|
+------+---+------+---+-------+----------+---------------+-------------+<jupyter_text>Drop two columns including the click time and attributed time. This is done for sack of simplicity in this toturial.<jupyter_code>data = data.drop('click_time','attributed_time')
data.show(3)
data.head()<jupyter_output><empty_output><jupyter_text>Split data.<jupyter_code># Split the data into training and test sets (30% held out for testing)
(trainingData, testData) = data.randomSplit([0.7, 0.3])
trainingData.printSchema()
trainingData.show(3)
data.columns<jupyter_output><empty_output><jupyter_text>Vectorize the data before feeding into the ML algorithm. Determine the columns name as follow.<jupyter_code>assembler = VectorAssembler(inputCols=['ip', 'app', 'device', 'os', 'channel'],outputCol="features")
trainingData = assembler.transform(trainingData)
testData = assembler.transform(testData)
trainingData.show(3)<jupyter_output>+---+---+------+---+-------+-------------+--------------------+
| ip|app|device| os|channel|is_attributed|            features|
+---+---+------+---+-------+-------------+--------------------+
|  9|  3|     1| 13|    115|            0|[9.0,3.0,1.0,13.0...|
|  9|  3|     1| 13|    135|            0|[9.0,3.0,1.0,13.0...|
|  9|  3|     1| 13|    280|            0|[9.0,3.0,1.0,13.0...|
+---+---+------+---+-------+-------------+--------------------+
only showing top 3 rows<jupyter_text>Asign the model, train the model, and predict.<jupyter_code>#https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#module-pyspark.ml.classification
# Train a RandomForest model.
#rf = RandomForestClassifier(labelCol="is_attributed", featuresCol="features", numTrees=20)
rf = RandomForestClassifier(labelCol="is_attributed", featuresCol="features", numTrees=30, maxDepth=3)
# Train model.  This also runs the indexers.
model = rf.fit(trainingData)
# Make predictions.
predictions = model.transform(testData)
predictions.printSchema()
# Select example rows to display.
predictions.select("prediction", "is_attributed", "features").show(5)<jupyter_output>+----------+-------------+--------------------+
|prediction|is_attributed|            features|
+----------+-------------+--------------------+
|       0.0|            0|[9.0,2.0,1.0,19.0...|
|       0.0|            0|[9.0,3.0,1.0,13.0...|
|       0.0|            0|[9.0,3.0,1.0,13.0...|
|       0.0|            0|[9.0,3.0,1.0,13.0...|
|       0.0|            0|[9.0,9.0,1.0,13.0...|
+----------+-------------+--------------------+
only showing top 5 rows<jupyter_text>Evaluate the model performance.<jupyter_code># Select (prediction, true label) and compute test error
evaluator = MulticlassClassificationEvaluator(labelCol="is_attributed", predictionCol="prediction", metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print("Test Error = %g" % (1.0 - accuracy))
print("Test accuracy = %g" % (accuracy))
# Not a very good example to show this!
model.featureImportances<jupyter_output><empty_output><jupyter_text>Apply the model to test dataset and predict. Read the test data first and prepare the format so it can be fed into the model.<jupyter_code>data = spark.read.csv('test.csv', inferSchema=True, header=True)
data.show(5)
#data = data.drop('click_time','attributed_time')
test = data
#test = data.select(['ip','app','device','os','channel'])
test.show(3)<jupyter_output>+--------+------+---+------+---+-------+-------------------+
|click_id|    ip|app|device| os|channel|         click_time|
+--------+------+---+------+---+-------+-------------------+
|       0|  5744|  9|     1|  3|    107|2017-11-10 04:00:00|
|       1|119901|  9|     1|  3|    466|2017-11-10 04:00:00|
|       2| 72287| 21|     1| 19|    128|2017-11-10 04:00:00|
+--------+------+---+------+---+-------+-------------------+
only showing top 3 rows<jupyter_text>Vecorize the test data and transfer it as follow.<jupyter_code>assembler = VectorAssembler(inputCols=['ip', 'app', 'device', 'os', 'channel'],outputCol="features")
test = assembler.transform(test)
test.show(3)<jupyter_output>+--------+------+---+------+---+-------+-------------------+--------------------+
|click_id|    ip|app|device| os|channel|         click_time|            features|
+--------+------+---+------+---+-------+-------------------+--------------------+
|       0|  5744|  9|     1|  3|    107|2017-11-10 04:00:00|[5744.0,9.0,1.0,3...|
|       1|119901|  9|     1|  3|    466|2017-11-10 04:00:00|[119901.0,9.0,1.0...|
|       2| 72287| 21|     1| 19|    128|2017-11-10 04:00:00|[72287.0,21.0,1.0...|
+--------+------+---+------+---+-------+-------------------+--------------------+
only showing top 3 rows<jupyter_text>Fed the test data set into the model and predict.<jupyter_code>predictions = model.transform(test)
predictions.show(2)<jupyter_output>+--------+------+---+------+---+-------+-------------------+--------------------+--------------------+--------------------+----------+
|click_id|    ip|app|device| os|channel|         click_time|            features|       rawPrediction|         probability|prediction|
+--------+------+---+------+---+-------+-------------------+--------------------+--------------------+--------------------+----------+
|       0|  5744|  9|     1|  3|    107|2017-11-10 04:00:00|[5744.0,9.0,1.0,3...|[20.6742278268919...|[0.68914092756306...|       0.0|
|       1|119901|  9|     1|  3|    466|2017-11-10 04:00:00|[119901.0,9.0,1.0...|[24.8633981858914...|[0.82877993952971...|       0.0|
+--------+------+---+------+---+-------+-------------------+--------------------+--------------------+--------------------+----------+
only showing top 2 rows<jupyter_text>Prepare a submiting file as follow and determine the columns name.<jupyter_code>data_to_submit = predictions.select(['click_id','prediction'])
data_to_submit.show(3)
data_to_submit = data_to_submit.withColumnRenamed('prediction','is_attributed')
data_to_submit.show(3)<jupyter_output>+--------+-------------+
|click_id|is_attributed|
+--------+-------------+
|       0|          0.0|
|       1|          0.0|
|       2|          1.0|
+--------+-------------+
only showing top 3 rows<jupyter_text>Lets before submiting the results, look inside and see the model performance.<jupyter_code>data_to_submit.groupBy('is_attributed').count().show()<jupyter_output>+-------------+--------+
|is_attributed|   count|
+-------------+--------+
|          0.0|15921222|
|          1.0| 2869247|
+-------------+--------+<jupyter_text>Not very balance results. We need to improve the results by tuning the model hyperparameters plus some data feathering are suggested.  Save the output file and submit it.<jupyter_code>#data_to_submit.to_csv('RFmodeling.csv', index = False)
#data_to_submit.coalesce(1).write.option("header","true").option("sep",",").mode("overwrite").csv("output/path")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.3074832397173401, 'fraction_numerical': 0.07238630186628012, 'mean_word_length': 1.9547644539614561, 'pattern_counts': {'":': 0, '<': 60, '<?xml version=': 0, '>': 60, 'https://': 1, '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': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '647921', 'license': 'MIT', 'n_tokens_mistral': 4039, 'n_tokens_neox': 3576, 'n_words': 676, 'path': 'Code/RandomForest/RF_FraudDetection.ipynb', 'repo_name': 'Atashnezhad/Fraud_detection', 'size': '11038'}