notebooks/IEEE-CIS-Fraud-Detection-Score-Spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3c41202a-c923-4e52-856e-85a31334221e",
   "metadata": {},
   "outputs": [],
   "source": [
    "!sudo wget -O /usr/local/spark/jars/hadoop-aws-3.3.1.jar https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.1/hadoop-aws-3.3.1.jar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8290e603-e0e5-4cc3-9271-aa789fa92768",
   "metadata": {},
   "outputs": [],
   "source": [
    "!sudo wget -O /usr/local/spark/jars/aws-java-sdk-bundle-1.11.901.jar https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.11.901/aws-java-sdk-bundle-1.11.901.jar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "8e32052f-a67d-4c0e-94a9-c9dff6b2434e",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.conf import SparkConf\n",
    "from pyspark.ml.feature import Imputer, StringIndexer, IndexToString\n",
    "from pyspark.sql import SparkSession, DataFrame\n",
    "from pyspark.sql.functions import when, count, col, lit, udf, isnan\n",
    "from pyspark.sql.types import *\n",
    "from pyspark import SparkContext\n",
    "import os\n",
    "import requests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 154,
   "id": "99b53836-3c74-441b-ab87-68d748c67981",
   "metadata": {},
   "outputs": [],
   "source": [
    "AWS_S3_CUSTOM_ENDPOINT = \"http://storage:9000\"\n",
    "AWS_ACCESS_KEY_ID = \"minioadmin\"\n",
    "AWS_SECRET_ACCESS_KEY = \"minioadmin\"\n",
    "\n",
    "FRAUD_DETECTION_DB=\"ml_fraud_detection_db\"\n",
    "FRAUD_DETECTION_SRC_TBL=f\"{FRAUD_DETECTION_DB}.tb_fraud_score\"\n",
    "\n",
    "PREDICT_URL = \"http://model_server:8501/v1/models/fraud-model:predict\"\n",
    "ID_COLUMN = \"TransactionID\"\n",
    "\n",
    "\n",
    "OUTPUT_DIR = \"s3a://ml-data/fraud_detection_output/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "d45242ea-5dcd-4703-9915-f00fa250866b",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.conf.SparkConf at 0x7f4ddf6f1e50>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf = SparkConf()\n",
    "conf.setMaster(\"local[6]\")\n",
    "conf.set(\"spark.driver.memory\", \"10g\")\n",
    "conf.set(\"spark.executor.memory\", \"4g\")\n",
    "conf.set(\"spark.executor.cores\", \"1\")\n",
    "conf.set(\"spark.dynamicAllocation.enabled\", \"true\")\n",
    "conf.set(\"spark.hadoop.parquet.enable.summary-metadata\", \"false\")\n",
    "conf.set(\"spark.hadoop.mapreduce.fileoutputcommitter.marksuccessfuljobs\", \"false\")\n",
    "conf.set(\"hive.metastore.uris\", \"thrift://hive-metastore:9083\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "ef1d20cf-c567-4f1f-a943-e6ca78efb2c0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.conf.SparkConf at 0x7f4ddf6f1e50>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf.set(\"spark.hadoop.fs.s3a.access.key\", AWS_ACCESS_KEY_ID)\n",
    "conf.set(\"spark.hadoop.fs.s3a.secret.key\", AWS_SECRET_ACCESS_KEY)\n",
    "conf.set(\"spark.hadoop.fs.s3a.endpoint\", AWS_S3_CUSTOM_ENDPOINT)\n",
    "conf.set(\"spark.hadoop.fs.s3a.path.style.access\", True)\n",
    "conf.set(\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "ffca674b-708f-4f1e-9d19-39087904134b",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = SparkSession.builder.appName(\"ScoreProcessor\").config(conf=conf).enableHiveSupport().getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "id": "c2750264-2b8b-4a50-980a-fde8e9be1023",
   "metadata": {},
   "outputs": [],
   "source": [
    "source_df = spark.sql(f\"select {ID_COLUMN}, ARRAY(c1) as c1, ARRAY(c13) as c13, ARRAY(c5) as c5, ARRAY(card1) as card1, ARRAY(card2) as card2, ARRAY(card3) as card3, ARRAY(card4) as card4, ARRAY(card5) as card5, ARRAY(card6) as card6, ARRAY(d1) as d1, ARRAY(d10) as d10, ARRAY(d11) as d11, ARRAY(d15) as d15, ARRAY(d3) as d3, ARRAY(d4) as d4, ARRAY(d5) as d5, ARRAY(dist1) as dist1, ARRAY(m2) as m2, ARRAY(m3) as m3, ARRAY(m4) as m4, ARRAY(m5) as m5, ARRAY(m6) as m6, ARRAY(m7) as m7, ARRAY(m8) as m8, ARRAY(m9) as m9, ARRAY(p_emaildomain) as p_emaildomain, ARRAY(productcd) as productcd, ARRAY(r_emaildomain) as r_emaildomain, ARRAY(transactionamt) as transactionamt, ARRAY(v10) as v10, ARRAY(v12) as v12, ARRAY(v15) as v15, ARRAY(v19) as v19, ARRAY(v2) as v2, ARRAY(v22) as v22, ARRAY(v23) as v23, ARRAY(v25) as v25, ARRAY(v29) as v29, ARRAY(v3) as v3, ARRAY(v35) as v35, ARRAY(v4) as v4, ARRAY(v6) as v6, ARRAY(v8) as v8 from {FRAUD_DETECTION_SRC_TBL}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "id": "68b78bc5-f946-4536-b784-22d299a48410",
   "metadata": {},
   "outputs": [],
   "source": [
    "source_df.createOrReplaceTempView(\"score_source_tbl\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fa3b21de-15d1-4ab8-84f7-8153abb99683",
   "metadata": {},
   "source": [
    "## First Transformation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "id": "09962b51-7185-4589-ad3a-b7073dec5b37",
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_columns = [c for c in source_df.columns if c != ID_COLUMN ]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "id": "a2c5dc10-553d-4222-a4ad-377650edf5c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "feature_columns_str = \",\".join(feature_columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "id": "229d12ea-ca9c-4668-95a9-9ec23c94bdd2",
   "metadata": {},
   "outputs": [],
   "source": [
    "first_transformation_df = spark.sql(f\"select {ID_COLUMN}, ARRAY(struct({feature_columns_str})) as instances, 'json_serving' as signature_name from score_source_tbl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "id": "0bcacf08-f474-4f6a-ad13-0f7496d0461d",
   "metadata": {},
   "outputs": [],
   "source": [
    "first_transformation_df.createOrReplaceTempView(\"first_transform_tbl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "id": "f61d6756-e821-4946-b2a1-eeaae7d4cfdc",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------------------+--------------+\n",
      "|TransactionID|           instances|signature_name|\n",
      "+-------------+--------------------+--------------+\n",
      "|      3663549|[{[6.0], [115.0],...|  json_serving|\n",
      "|      3663550|[{[3.0], [12.0], ...|  json_serving|\n",
      "|      3663551|[{[2.0], [22.0], ...|  json_serving|\n",
      "|      3663552|[{[5.0], [7.0], [...|  json_serving|\n",
      "|      3663553|[{[6.0], [14.0], ...|  json_serving|\n",
      "|      3663554|[{[5.0], [10.0], ...|  json_serving|\n",
      "|      3663555|[{[1.0], [2.0], [...|  json_serving|\n",
      "|      3663556|[{[3.0], [11.0], ...|  json_serving|\n",
      "|      3663557|[{[152.0], [407.0...|  json_serving|\n",
      "|      3663558|[{[2.0], [8.0], [...|  json_serving|\n",
      "|      3663559|[{[2.0], [3.0], [...|  json_serving|\n",
      "|      3663560|[{[4.0], [17.0], ...|  json_serving|\n",
      "|      3663561|[{[3.0], [1.0], [...|  json_serving|\n",
      "|      3663562|[{[3.0], [9.0], [...|  json_serving|\n",
      "|      3663563|[{[1.0], [3.0], [...|  json_serving|\n",
      "|      3663564|[{[6.0], [13.0], ...|  json_serving|\n",
      "|      3663565|[{[173.0], [510.0...|  json_serving|\n",
      "|      3663566|[{[1.0], [2.0], [...|  json_serving|\n",
      "|      3663567|[{[1.0], [0.0], [...|  json_serving|\n",
      "|      3663568|[{[5.0], [12.0], ...|  json_serving|\n",
      "+-------------+--------------------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select * from first_transform_tbl\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2a611235-c799-4725-be7d-937dda0066a3",
   "metadata": {},
   "source": [
    "## Second Transformation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "id": "048dd27f-607a-47dd-9cc7-db9fe0a76c38",
   "metadata": {},
   "outputs": [],
   "source": [
    "second_transformation_df = spark.sql(f\"select {ID_COLUMN}, to_json(struct(instances, signature_name)) as data, '{PREDICT_URL}' as url from first_transform_tbl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "id": "1d31e1b2-cce5-4852-942a-3d9bef4475bf",
   "metadata": {},
   "outputs": [],
   "source": [
    "second_transformation_df.createOrReplaceTempView(\"second_transform_tbl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "id": "d74452b5-dc32-4390-8108-270d2fd46449",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------+\n",
      "|TransactionID|data                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |url                                                   |\n",
      "+-------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------+\n",
      "|3663549      |{\"instances\":[{\"c1\":[6.0],\"c13\":[115.0],\"c5\":[3.0],\"card1\":[10409.0],\"card2\":[111.0],\"card3\":[150.0],\"card4\":[\"visa\"],\"card5\":[226.0],\"card6\":[\"debit\"],\"d1\":[419.0],\"d10\":[418.0],\"d11\":[203.0],\"d15\":[409.0],\"d3\":[27.0],\"d4\":[398.0],\"d5\":[27.0],\"dist1\":[1.0],\"m2\":[\"T\"],\"m3\":[\"F\"],\"m4\":[\"M1\"],\"m5\":[\"T\"],\"m6\":[\"F\"],\"m7\":[\"T\"],\"m8\":[\"T\"],\"m9\":[\"T\"],\"p_emaildomain\":[\"gmail.com\"],\"productcd\":[\"W\"],\"r_emaildomain\":[\"scranton.edu\"],\"transactionamt\":[31.95],\"v10\":[1.0],\"v12\":[0.0],\"v15\":[0.0],\"v19\":[0.0],\"v2\":[1.0],\"v22\":[0.0],\"v23\":[1.0],\"v25\":[0.0],\"v29\":[0.0],\"v3\":[1.0],\"v35\":[1.0],\"v4\":[1.0],\"v6\":[1.0],\"v8\":[1.0]}],\"signature_name\":\"json_serving\"}|http://model_server:8501/v1/models/fraud-model:predict|\n",
      "+-------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------------------------------------------------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select * from second_transform_tbl\").show(truncate=False, n=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "83c3dd15-f58c-4fb5-b218-6e5c62b5867f",
   "metadata": {},
   "source": [
    "## Predict UDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 139,
   "id": "758db761-190d-4b53-b21e-80f73b526dd8",
   "metadata": {},
   "outputs": [],
   "source": [
    "def predict_function(data, url):\n",
    "    prediction = requests.post(url, data).json()\n",
    "    value = prediction['predictions'][0][0]\n",
    "    # print(value)\n",
    "    return round(value, 2)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c18943e0-567d-4f9d-a51e-760ced2a9351",
   "metadata": {},
   "source": [
    "### Sample Data to test UDF"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 143,
   "id": "33face95-9b4f-4e8a-bf8f-3f7726ff9b1e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.0"
      ]
     },
     "execution_count": 143,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "req_data = '{\"instances\":[{\"c1\":[6.0],\"c13\":[115.0],\"c5\":[3.0],\"card1\":[10409.0],\"card2\":[111.0],\"card3\":[150.0],\"card4\":[\"visa\"],\"card5\":[226.0],\"card6\":[\"debit\"],\"d1\":[419.0],\"d10\":[418.0],\"d11\":[203.0],\"d15\":[409.0],\"d3\":[27.0],\"d4\":[398.0],\"d5\":[27.0],\"dist1\":[1.0],\"m2\":[\"T\"],\"m3\":[\"F\"],\"m4\":[\"M1\"],\"m5\":[\"T\"],\"m6\":[\"F\"],\"m7\":[\"T\"],\"m8\":[\"T\"],\"m9\":[\"T\"],\"p_emaildomain\":[\"gmail.com\"],\"productcd\":[\"W\"],\"r_emaildomain\":[\"scranton.edu\"],\"transactionamt\":[31.95],\"v10\":[1.0],\"v12\":[0.0],\"v15\":[0.0],\"v19\":[0.0],\"v2\":[1.0],\"v22\":[0.0],\"v23\":[1.0],\"v25\":[0.0],\"v29\":[0.0],\"v3\":[1.0],\"v35\":[1.0],\"v4\":[1.0],\"v6\":[1.0],\"v8\":[1.0]}],\"signature_name\":\"json_serving\"}'\n",
    "predict_function(req_data, PREDICT_URL)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 144,
   "id": "91f62001-8fd1-447a-8cbb-9531294e47f9",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<function __main__.predict_function(data, url)>"
      ]
     },
     "execution_count": 144,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import col, udf\n",
    "from pyspark.sql.types import IntegerType, FloatType\n",
    "\n",
    "predict_udf = udf(lambda data:predict_function(data))  \n",
    "spark.udf.register(\"predict\", predict_function,FloatType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 145,
   "id": "1df15424-4033-4642-9f84-5fd8310b65b2",
   "metadata": {},
   "outputs": [],
   "source": [
    "predict_df = spark.sql(f\"select {ID_COLUMN}, predict(data, url) as isfraud from second_transform_tbl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 146,
   "id": "8b6ab4b2-d7db-4a4d-8e25-748d1b0a702b",
   "metadata": {},
   "outputs": [],
   "source": [
    "predict_df.cache()\n",
    "predict_df.createOrReplaceTempView(\"predict_tbl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "id": "7505b583-2cde-4684-8e32-53363763d848",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+\n",
      "|total_records|\n",
      "+-------------+\n",
      "|506691       |\n",
      "+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"select count(*) as total_records from predict_tbl\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 159,
   "id": "a659f91c-6df6-4370-a970-e98cdb94a337",
   "metadata": {},
   "outputs": [],
   "source": [
    "final_df = spark.sql(f\"select {ID_COLUMN}, isfraud as isFraud from predict_tbl\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 160,
   "id": "7fa7b755-1ff1-4005-9abd-90243a8bd5c7",
   "metadata": {},
   "outputs": [],
   "source": [
    "final_df.coalesce(1).write.mode(\"overwrite\").option(\"header\",True).csv(OUTPUT_DIR)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e9f9db3c-0ecb-4fcc-b2f9-671fa92d6c3a",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}