{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"I forked the majority of the spark code in this notebook from https://www.kaggle.com/code/rayanaay/pyspark-is-all-you-need-simple-decision-tree. I added onto it some memory enhancements and added some random forest code. The memory enhancements  allowed me to create a submission by partitioning the output in different csv files and putting them all into one csv file. the random forest created too many memory issues to even export the submission. Will potentially try to solve that later. Hope you find this useful and thanks to Rayan-Aay for putting together the original notebook!","metadata":{}},{"cell_type":"markdown","source":"## Disclaimer :\nthis notebook is not published with good performance, the notebook aims to  propose an approach based on Spark to introduce people to Distributed Computing and to encourage this type of implementation, that is representative of the industry reality.\n\nIn this notebook, we'll try to introduce you  pyspark which is an interface for Apache Spark ( dedicated for distributed computation ) in Python.","metadata":{}},{"cell_type":"code","source":"import numpy as np\nimport pandas as pd \nimport random\nimport os\nimport gc","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2022-08-10T03:28:07.803917Z","iopub.execute_input":"2022-08-10T03:28:07.804405Z","iopub.status.idle":"2022-08-10T03:28:07.809987Z","shell.execute_reply.started":"2022-08-10T03:28:07.804360Z","shell.execute_reply":"2022-08-10T03:28:07.809019Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"\n\nThe power of Pyspark is its simplicity, because we don't need to manipulate / use Resilient Distributed Dataset ( RDD ) which are the basic unit of Spark that are distributed on the cluster, pyspark handle all of this. However, if you want to program at this level you can easily do it.\n\n## Resilient Distributed Dataset\n-  An RDD (Resilient Distributed Dataset) is the basic abstraction of Spark representing an unchanging set of elements partitioned across cluster nodes, allowing parallel computation.\n- It has 3 advantages:\n    - Performance.\n    - Consistency.\n    - Fault Tolerance\n\nPyspark supports most of Spark features such as Dataframe / SparkSQL / MLlib. We will not introduce Spark Streaming in this tutorial.\n\n## SparkSQL \n- Simply said, sparkSQL allows  to query structured data inside Spark programs. It acts as a distributed SQL query engine for fast data retrieving and processing. We can  also use SQL's like code creating a temporary view as we will see at the end of the tutorial\n- It provides a programming abstraction called DataFrame:\n    -  Just as Python Dataframe, pyspark allows us to use this structure which is distributable across multiple machine of the cluster\n\n## MLlib\n- MLlib is a scalable machine learning library that provides a uniform set of high-level APIs that help users create and tune practical machine learning pipelines. For this competition, we'll use trees algorithms.\n\n","metadata":{}},{"cell_type":"code","source":"# Install pyspark\n!pip3 install pyspark","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:28:07.816406Z","iopub.execute_input":"2022-08-10T03:28:07.817413Z","iopub.status.idle":"2022-08-10T03:29:06.612258Z","shell.execute_reply.started":"2022-08-10T03:28:07.817357Z","shell.execute_reply":"2022-08-10T03:29:06.610363Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import pyspark\nimport pyspark.sql.functions as F\nimport pyspark.sql.types\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import SQLContext","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:06.616369Z","iopub.execute_input":"2022-08-10T03:29:06.616846Z","iopub.status.idle":"2022-08-10T03:29:06.708016Z","shell.execute_reply.started":"2022-08-10T03:29:06.616807Z","shell.execute_reply":"2022-08-10T03:29:06.706916Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"SparkSession is the entry point to Spark/Pyspark to work with RDD/Dataframe. We can also configure properties like :\n\n - `spark.driver.memory` : Amount of memory to use for the driver process where the Session is initialized. I initialize it by default at 16gb         \n - `spark.sql.shuffle.partitions` :  number of partitions to use when shuffling data for joins or aggregations. In this example we partitioned the data into 150 partitions.","metadata":{}},{"cell_type":"code","source":"spark = (SparkSession.builder.master(\"local[*]\")\n                    .config('spark.driver.maxResultSize', 0)\n                    .config(\"spark.driver.memory\",\"16g\")\n                    .config(\"spark.sql.shuffle.partitions\",20)\n                    .appName('PysparkIsAllYouNeed')\n                    .getOrCreate())\n","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:06.709445Z","iopub.execute_input":"2022-08-10T03:29:06.711067Z","iopub.status.idle":"2022-08-10T03:29:13.459617Z","shell.execute_reply.started":"2022-08-10T03:29:06.711003Z","shell.execute_reply":"2022-08-10T03:29:13.458530Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def sample_df(n_cols,fraction_rows,path):\n    print(\"using PySpark ...\")\n    parDF1=spark.read.parquet(path)\n    cols_list = [\"customer_ID\"] + random.sample(parDF1.columns[1:-1],n_cols) + ['target']\n    \n    parDF1 = parDF1.select(*cols_list).sample(fraction=fraction_rows)\n    \n    \n    return parDF1","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:13.462962Z","iopub.execute_input":"2022-08-10T03:29:13.463713Z","iopub.status.idle":"2022-08-10T03:29:13.470973Z","shell.execute_reply.started":"2022-08-10T03:29:13.463673Z","shell.execute_reply":"2022-08-10T03:29:13.470071Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"I observe that using all the columns make the kernel crash, so I decide to randomly select a subset of columns while looking for future solution.","metadata":{}},{"cell_type":"code","source":"train_sample = sample_df(n_cols=40,fraction_rows=0.2,path=\"../input/amex-parquet/train_data.parquet\")\ntrain_sample.count(),len(train_sample.columns)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:13.472837Z","iopub.execute_input":"2022-08-10T03:29:13.473446Z","iopub.status.idle":"2022-08-10T03:29:22.924021Z","shell.execute_reply.started":"2022-08-10T03:29:13.473413Z","shell.execute_reply":"2022-08-10T03:29:22.922636Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"cols_used = train_sample.columns","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:22.930280Z","iopub.execute_input":"2022-08-10T03:29:22.933995Z","iopub.status.idle":"2022-08-10T03:29:22.942311Z","shell.execute_reply.started":"2022-08-10T03:29:22.933905Z","shell.execute_reply":"2022-08-10T03:29:22.940553Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Preprocessing","metadata":{}},{"cell_type":"code","source":"from pyspark.ml.classification import DecisionTreeClassifier,RandomForestClassifier\nfrom pyspark.ml.feature import VectorAssembler\nfrom pyspark.ml.evaluation import BinaryClassificationEvaluator\nfrom sklearn.metrics import confusion_matrix","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:22.945107Z","iopub.execute_input":"2022-08-10T03:29:22.948990Z","iopub.status.idle":"2022-08-10T03:29:24.251255Z","shell.execute_reply.started":"2022-08-10T03:29:22.948861Z","shell.execute_reply":"2022-08-10T03:29:24.249832Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"For the sake of the tutorial, I decide to apply a very simple preprocessing and aggregation. You can easily perform better with further preprocessing","metadata":{}},{"cell_type":"markdown","source":"### Aggregation function","metadata":{}},{"cell_type":"code","source":"def agg_data(train_sample,option=\"train\"):\n    # Select only numerical columns for aggregation\n    if option ==\"train\":\n        numerical_list = [item[0] for item in train_sample.dtypes[1:-1] if not  item[1].startswith('string') ]\n\n\n    else:\n        numerical_list = [item[0] for item in train_sample.dtypes[1:] if not  item[1].startswith('string') ]\n    print(numerical_list)\n    sum_agg = {x: \"sum\" for x in numerical_list   if x is not train_sample.columns[0] }\n    min_agg = {x: \"min\" for x in numerical_list if x is not train_sample.columns[0] }\n    std_agg = {x: \"stddev_samp\" for x in numerical_list if x is not train_sample.columns[0] }\n    #variance_agg = {x: \"variance\" for x in numerical_list if x is not train_sample.columns[0] }\n    avg_agg = {x: \"mean\" for x in numerical_list if x is not train_sample.columns[0] }\n    if option ==\"train\":\n        max_agg = {x: \"max\" for x in numerical_list + ['target'] if x is not train_sample.columns[0] }\n\n    else:\n        max_agg = {x: \"max\" for x in numerical_list if x is not train_sample.columns[0] }\n\n    sum_df = train_sample.groupBy(\"customer_ID\").agg(sum_agg).withColumnRenamed(\"customer_ID\",\"customer_ID_1\")\n    min_df = train_sample.groupBy(\"customer_ID\").agg(min_agg).withColumnRenamed(\"customer_ID\",\"customer_ID_2\")\n    std_df = train_sample.groupBy(\"customer_ID\").agg(std_agg).withColumnRenamed(\"customer_ID\",\"customer_ID_3\")\n    #variance_df = train_sample.groupBy(\"customer_ID\").agg(variance_agg).withColumnRenamed(\"customer_ID\",\"customer_ID_4\")\n    avg_df = train_sample.groupBy(\"customer_ID\").agg(avg_agg).withColumnRenamed(\"customer_ID\",\"customer_ID_5\")\n    max_df = train_sample.groupBy(\"customer_ID\").agg(max_agg).withColumnRenamed(\"customer_ID\",\"customer_ID_6\")\n\n    join1 = sum_df.join(min_df, sum_df.customer_ID_1 == min_df.customer_ID_2, 'inner').select(\"*\")\n    join2 = join1.join(std_df, join1.customer_ID_1 == std_df.customer_ID_3, 'inner').select(\"*\")\n    #join3 = join2.join(variance_df, join2.customer_ID_1 == variance_df.customer_ID_4, 'inner').select(\"*\")\n\n    join4 = join2.join(avg_df, join2.customer_ID_1 == avg_df.customer_ID_5, 'inner').select(\"*\")\n\n    join_final = join4.join(max_df, join4.customer_ID_1 == max_df.customer_ID_6, 'inner').select(\"*\")\n\n    join_final = join_final.drop(*[\"customer_ID_{}\".format(x) for x in range(2,7)]).withColumnRenamed(\"customer_ID_1\",\"customer_ID\")\n\n    return join_final","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:24.253777Z","iopub.execute_input":"2022-08-10T03:29:24.254768Z","iopub.status.idle":"2022-08-10T03:29:24.274496Z","shell.execute_reply.started":"2022-08-10T03:29:24.254708Z","shell.execute_reply":"2022-08-10T03:29:24.272757Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Preprocessing ","metadata":{}},{"cell_type":"markdown","source":"It is important to note that MLlib models inputs should be in this form ( X,Y ) where X represent the vectorized inputs. For this, we simply need to use `VectorAssembler`.","metadata":{}},{"cell_type":"code","source":"def preprocess_data(df,option=\"train\"):\n\n    if option ==\"train\":\n        agg_df = agg_data(df,option=\"train\")\n        print(\"training ...\")\n        agg_df = agg_df.withColumnRenamed(\"max(target)\", \"target\")\n        input_cols = agg_df.select(\"*\").drop(\"customer_ID\",\"target\").columns\n        va = VectorAssembler(inputCols = input_cols, outputCol='features',handleInvalid = \"keep\")\n        va_df = va.transform(agg_df)\n        va_df = va_df.select(['features', 'target'])  \n    else:\n        print(\"testing ...\")\n        agg_df = agg_data(df,option=\"test\")\n        input_cols = agg_df.select(\"*\").drop(\"customer_ID\").columns\n        va = VectorAssembler(inputCols = input_cols, outputCol='features',handleInvalid = \"keep\")\n        va_df = va.transform(agg_df)\n        va_df = va_df.select(['customer_ID','features']) \n    \n          \n    return va_df","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:24.276387Z","iopub.execute_input":"2022-08-10T03:29:24.276762Z","iopub.status.idle":"2022-08-10T03:29:24.294117Z","shell.execute_reply.started":"2022-08-10T03:29:24.276730Z","shell.execute_reply":"2022-08-10T03:29:24.292442Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_data_agg_vectorized  = preprocess_data(train_sample,\"train\")\ntrain_data_agg_vectorized.count(),len(train_data_agg_vectorized.columns)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:24.298736Z","iopub.execute_input":"2022-08-10T03:29:24.299282Z","iopub.status.idle":"2022-08-10T03:29:37.244949Z","shell.execute_reply.started":"2022-08-10T03:29:24.299237Z","shell.execute_reply":"2022-08-10T03:29:37.243549Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Split the data randomly\n(train, valid) = train_data_agg_vectorized.randomSplit([0.7, 0.3])\n","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:37.250284Z","iopub.execute_input":"2022-08-10T03:29:37.251651Z","iopub.status.idle":"2022-08-10T03:29:37.357506Z","shell.execute_reply.started":"2022-08-10T03:29:37.251597Z","shell.execute_reply":"2022-08-10T03:29:37.356215Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Model & evaluation","metadata":{}},{"cell_type":"code","source":"\n\"\"\"\n## Random Forest \ndtc = RandomForestClassifier(numTrees=20,\n      maxDepth=8,subsamplingRate=0.8,\n      featureSubsetStrategy= \"sqrt\",\n      featuresCol=\"features\", labelCol=\"target\")\n\"\"\"\n\n\ndtc = DecisionTreeClassifier(maxDepth=15,featuresCol=\"features\", labelCol=\"target\")\nmodel = dtc.fit(train)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:29:37.359040Z","iopub.execute_input":"2022-08-10T03:29:37.359832Z","iopub.status.idle":"2022-08-10T03:33:59.555767Z","shell.execute_reply.started":"2022-08-10T03:29:37.359787Z","shell.execute_reply":"2022-08-10T03:33:59.553499Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Evaluation","metadata":{}},{"cell_type":"markdown","source":"Because of the unbalanced character of the dataset, we use ``areaUnderROC`` using the `BinaryClassificationEvaluator`.\n\nIts also important to note that `.collect()` will gather/collect all the data that are distributed accross the cluster to this kernel. It's usually not recommended to use this method because local RAM can't handle this amount of data generally.","metadata":{}},{"cell_type":"code","source":"evaluator = BinaryClassificationEvaluator(rawPredictionCol=\"prediction\",labelCol=\"target\",metricName=\"areaUnderROC\")\npred = model.transform(valid)\nacc = evaluator.evaluate(pred)\n \nprint(\"Prediction auc: \", acc)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:33:59.557296Z","iopub.execute_input":"2022-08-10T03:33:59.557723Z","iopub.status.idle":"2022-08-10T03:34:43.397244Z","shell.execute_reply.started":"2022-08-10T03:33:59.557689Z","shell.execute_reply":"2022-08-10T03:34:43.393728Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"'''\ny_pred=pred.select(\"prediction\").collect()\ny_orig=pred.select(\"target\").collect()\n\ncm = confusion_matrix(y_orig, y_pred)\nprint(\"Confusion Matrix:\")\nprint(cm)\n'''","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:34:43.399234Z","iopub.execute_input":"2022-08-10T03:34:43.399847Z","iopub.status.idle":"2022-08-10T03:34:43.411564Z","shell.execute_reply.started":"2022-08-10T03:34:43.399798Z","shell.execute_reply":"2022-08-10T03:34:43.410214Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Free memory\ntrain.unpersist()\nvalid.unpersist()\ntrain_sample.unpersist()\ntrain_data_agg_vectorized.unpersist()\n\n\ndel  train,valid, evaluator, acc, train_sample\ntry :\n    del train_data_agg_vectorized, cm, y_pred, y_orig, pred,\nexcept:\n    pass\n\ngc.collect()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:34:43.413370Z","iopub.execute_input":"2022-08-10T03:34:43.415017Z","iopub.status.idle":"2022-08-10T03:34:43.620602Z","shell.execute_reply.started":"2022-08-10T03:34:43.414940Z","shell.execute_reply":"2022-08-10T03:34:43.619419Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Testing","metadata":{}},{"cell_type":"code","source":"cols_used","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:34:43.622185Z","iopub.execute_input":"2022-08-10T03:34:43.622904Z","iopub.status.idle":"2022-08-10T03:34:43.633338Z","shell.execute_reply.started":"2022-08-10T03:34:43.622861Z","shell.execute_reply":"2022-08-10T03:34:43.631895Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"test_path = \"../input/amex-parquet/test_data.parquet\"\ncols_used.sort()\ntest_cols = cols_used[:-2] # Use the same columns but without the target\ntest_data =spark.read.parquet(test_path).select(*test_cols)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:34:43.635243Z","iopub.execute_input":"2022-08-10T03:34:43.636039Z","iopub.status.idle":"2022-08-10T03:34:43.844232Z","shell.execute_reply.started":"2022-08-10T03:34:43.635986Z","shell.execute_reply":"2022-08-10T03:34:43.843100Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"test_data_agg_vectorized  = preprocess_data(test_data,\"test\")\n\ntest_data_agg_vectorized.count(),len(test_data_agg_vectorized.columns)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:34:43.845342Z","iopub.execute_input":"2022-08-10T03:34:43.845830Z","iopub.status.idle":"2022-08-10T03:35:21.507024Z","shell.execute_reply.started":"2022-08-10T03:34:43.845782Z","shell.execute_reply":"2022-08-10T03:35:21.505752Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"test_pred = model.transform(test_data_agg_vectorized)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:21.509442Z","iopub.execute_input":"2022-08-10T03:35:21.509932Z","iopub.status.idle":"2022-08-10T03:35:21.693123Z","shell.execute_reply.started":"2022-08-10T03:35:21.509886Z","shell.execute_reply":"2022-08-10T03:35:21.692198Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#test_pred.show()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:21.694607Z","iopub.execute_input":"2022-08-10T03:35:21.695961Z","iopub.status.idle":"2022-08-10T03:35:21.700934Z","shell.execute_reply.started":"2022-08-10T03:35:21.695900Z","shell.execute_reply":"2022-08-10T03:35:21.699541Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"We'll use the famous  User-Defined Functions (UDFs) to  write functions in Python and use them when writing Spark SQL queries. In order to get the probability vector as described in the function, I just did some logical operations to get the desired probability vector.","metadata":{}},{"cell_type":"code","source":"from pyspark.sql.types import DoubleType, IntegerType\nfrom pyspark.sql.functions import udf, col\n\ndef extract_prob_0(v):\n    try:\n        return float(v[0]) # Your VectorUDT is of length 2\n    except ValueError:\n        return None\n\ndef extract_prob_1(v):\n    try:\n        return float(v[1])  # Your VectorUDT is of length 2\n    except ValueError:\n        return None\n\n\nextract_prob0_udf = udf(extract_prob_0, DoubleType())\nextract_prob1_udf = udf(extract_prob_1, DoubleType())\n\n\n\nprobabilities = (test_pred.select(\"customer_ID\",\"probability\").withColumn(\"prob_0\", extract_prob0_udf(col(\"probability\")))\n                                                             .withColumn(\"prob_1\", extract_prob1_udf(col(\"probability\"))))","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:21.702774Z","iopub.execute_input":"2022-08-10T03:35:21.703131Z","iopub.status.idle":"2022-08-10T03:35:21.865581Z","shell.execute_reply.started":"2022-08-10T03:35:21.703099Z","shell.execute_reply":"2022-08-10T03:35:21.864197Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#probabilities.select(\"prob_0\",\"prob_1\").show(5)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:21.867203Z","iopub.execute_input":"2022-08-10T03:35:21.867709Z","iopub.status.idle":"2022-08-10T03:35:21.873511Z","shell.execute_reply.started":"2022-08-10T03:35:21.867662Z","shell.execute_reply":"2022-08-10T03:35:21.872296Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"test_data_agg_vectorized.unpersist()\ntest_data.unpersist()\ntest_pred.unpersist()\n\n\ndel test_data_agg_vectorized, test_data, test_pred, test_cols, model\ngc.collect()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:21.875681Z","iopub.execute_input":"2022-08-10T03:35:21.876607Z","iopub.status.idle":"2022-08-10T03:35:22.038877Z","shell.execute_reply.started":"2022-08-10T03:35:21.876562Z","shell.execute_reply":"2022-08-10T03:35:22.035570Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Some SparkSQL","metadata":{}},{"cell_type":"code","source":"sqlContext = SQLContext(spark)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.040816Z","iopub.execute_input":"2022-08-10T03:35:22.041163Z","iopub.status.idle":"2022-08-10T03:35:22.068870Z","shell.execute_reply.started":"2022-08-10T03:35:22.041132Z","shell.execute_reply":"2022-08-10T03:35:22.067552Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"probabilities.createOrReplaceTempView(\"tmp_probabilities\")\n\nsample_submission = sqlContext.sql(\n\"\"\"\nSELECT \ncustomer_ID,\nCASE WHEN prob_0 > prob_1 THEN 1-prob_0\n          ELSE prob_1\nEND as prediction\nFROM tmp_probabilities\n\"\"\"\n)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.070605Z","iopub.execute_input":"2022-08-10T03:35:22.072805Z","iopub.status.idle":"2022-08-10T03:35:22.303915Z","shell.execute_reply.started":"2022-08-10T03:35:22.072745Z","shell.execute_reply":"2022-08-10T03:35:22.302624Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"del probabilities\ngc.collect()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.305237Z","iopub.execute_input":"2022-08-10T03:35:22.305677Z","iopub.status.idle":"2022-08-10T03:35:22.486307Z","shell.execute_reply.started":"2022-08-10T03:35:22.305635Z","shell.execute_reply":"2022-08-10T03:35:22.485040Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#sample_submission.show(5)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.488231Z","iopub.execute_input":"2022-08-10T03:35:22.490238Z","iopub.status.idle":"2022-08-10T03:35:22.498583Z","shell.execute_reply.started":"2022-08-10T03:35:22.490190Z","shell.execute_reply":"2022-08-10T03:35:22.496841Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"#### Some sanity checks (commented out for issues with RAM they cause)","metadata":{}},{"cell_type":"code","source":"#sample_submission.groupBy(\"customer_ID\").agg(F.count(\"customer_ID\")).count() == sample_submission.count()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.504823Z","iopub.execute_input":"2022-08-10T03:35:22.505736Z","iopub.status.idle":"2022-08-10T03:35:22.510823Z","shell.execute_reply.started":"2022-08-10T03:35:22.505694Z","shell.execute_reply":"2022-08-10T03:35:22.509711Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#sample_submission.select(\"prediction\").agg(F.max(\"prediction\"),F.min(\"prediction\")).show()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.512388Z","iopub.execute_input":"2022-08-10T03:35:22.512917Z","iopub.status.idle":"2022-08-10T03:35:22.526420Z","shell.execute_reply.started":"2022-08-10T03:35:22.512881Z","shell.execute_reply":"2022-08-10T03:35:22.525374Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Convert sparkDataframe to pandas Dataframe ( we collect all the RDD's in the edge node )\n#final_submission_df = sample_submission.toPandas()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.528296Z","iopub.execute_input":"2022-08-10T03:35:22.529012Z","iopub.status.idle":"2022-08-10T03:35:22.537567Z","shell.execute_reply.started":"2022-08-10T03:35:22.528974Z","shell.execute_reply":"2022-08-10T03:35:22.536667Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#import shutil\n#shutil.rmtree(\"/kaggle/working/submission\")","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:44:13.623598Z","iopub.execute_input":"2022-08-10T03:44:13.624187Z","iopub.status.idle":"2022-08-10T03:44:13.652206Z","shell.execute_reply.started":"2022-08-10T03:44:13.624111Z","shell.execute_reply":"2022-08-10T03:44:13.650823Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"sqlContext.clearCache()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.551774Z","iopub.execute_input":"2022-08-10T03:35:22.552413Z","iopub.status.idle":"2022-08-10T03:35:22.568704Z","shell.execute_reply.started":"2022-08-10T03:35:22.552377Z","shell.execute_reply":"2022-08-10T03:35:22.567422Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"gc.collect()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:35:22.570399Z","iopub.execute_input":"2022-08-10T03:35:22.570875Z","iopub.status.idle":"2022-08-10T03:35:22.755782Z","shell.execute_reply.started":"2022-08-10T03:35:22.570796Z","shell.execute_reply":"2022-08-10T03:35:22.754438Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"sample_submission.write.csv(\"/kaggle/working/submission\")","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:44:18.162572Z","iopub.execute_input":"2022-08-10T03:44:18.163235Z","iopub.status.idle":"2022-08-10T03:47:49.287148Z","shell.execute_reply.started":"2022-08-10T03:44:18.163188Z","shell.execute_reply":"2022-08-10T03:47:49.284246Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#n_splits = 700\n  \n# Calculate count of each dataframe rows\n#each_len = sample_submission.count() // n_splits","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":" \"\"\"\n #potential optimization of submission\n     i = 0\nwhile i < n_splits:\n    #sample_submission.limit(each_len).toPandas().to_csv(f\"/kaggle/working/sample_submission/submission_{i}.csv\")\n    sample_submission.limit(each_len).write.csv(f\"/kaggle/working/sample_submission/submission_{i}.csv\")\n    # Truncate the `copy_df` to remove\n    # the contents fetched for `temp_df`\n    sample_submission = sample_submission.subtract(sample_submission.limit(each_len))\n    \n    i+=1\n    \n    gc.collect()\n    sqlContext.clearCache()\nsample_submission.coalesce(1).write.csv(f\"/kaggle/working/sample_submission/submission_final.csv\")\n \"\"\"","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#target_path = \"/kaggle/working/sample_submission\"\n\n#sample_submission.write.format(\"csv\").save(target_path)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#RAM WON\"T ALLOW TO PANDAS (EXTREMELY EXPENSIVE)\n#sample_submission.toPandas().to_csv(\"/kaggle/working/sample_submission/rf_submission1.csv\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"del sample_submission\ngc.collect()\nsqlContext.clearCache()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:49:06.220799Z","iopub.execute_input":"2022-08-10T03:49:06.221268Z","iopub.status.idle":"2022-08-10T03:49:06.402520Z","shell.execute_reply.started":"2022-08-10T03:49:06.221233Z","shell.execute_reply":"2022-08-10T03:49:06.400789Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"final_df = pd.DataFrame(columns=['customer_ID', 'prediction'])","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:49:10.826698Z","iopub.execute_input":"2022-08-10T03:49:10.827411Z","iopub.status.idle":"2022-08-10T03:49:10.862046Z","shell.execute_reply.started":"2022-08-10T03:49:10.827366Z","shell.execute_reply":"2022-08-10T03:49:10.860985Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"final_df.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"file_str = '/kaggle/working/submission/'\nfor dirname, _, filenames in os.walk(file_str):\n    for filename in filenames:\n        if filename.endswith('.csv'):\n            print(filename)\n            csv = pd.read_csv(f'{file_str}{filename}', header=None)\n            csv.columns = ['customer_ID', 'prediction']\n            final_df = final_df.append(csv)\n            del csv\n            gc.collect()","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:49:15.999855Z","iopub.execute_input":"2022-08-10T03:49:16.001248Z","iopub.status.idle":"2022-08-10T03:49:21.065179Z","shell.execute_reply.started":"2022-08-10T03:49:16.001180Z","shell.execute_reply":"2022-08-10T03:49:21.063662Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"final_df.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#export final submission\nfinal_df.to_csv('/kaggle/working/submission/dt_submission.csv', index=False)","metadata":{"execution":{"iopub.status.busy":"2022-08-10T03:49:42.065748Z","iopub.execute_input":"2022-08-10T03:49:42.066251Z","iopub.status.idle":"2022-08-10T03:49:44.375985Z","shell.execute_reply.started":"2022-08-10T03:49:42.066193Z","shell.execute_reply":"2022-08-10T03:49:44.374607Z"},"trusted":true},"execution_count":null,"outputs":[]}]}