{"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":"# ML with Pyspark (In Progress)\nThis is still in progress actually, yet I wanna show you how to tackle the competition with pyspark. In this noteboook I only use categoric columns for training and predictions, because the number of features get bloated when adding numeric features with mean, std, min and max. I haven't regarded the evaluation metric and just doing training -> predictions -> submissions. ","metadata":{}},{"cell_type":"markdown","source":"## Setup and Install","metadata":{}},{"cell_type":"code","source":"!pip install -q pyspark","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2022-06-19T20:02:29.909798Z","iopub.execute_input":"2022-06-19T20:02:29.910269Z","iopub.status.idle":"2022-06-19T20:03:18.250040Z","shell.execute_reply.started":"2022-06-19T20:02:29.910180Z","shell.execute_reply":"2022-06-19T20:03:18.248152Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Import Libraries","metadata":{}},{"cell_type":"code","source":"import itertools\nimport multiprocessing\nimport re\nfrom IPython import display\n\nimport numpy as np\nimport pandas as pd\nimport matplotlib.pyplot as plt\n\nimport pyspark.pandas as ps\nfrom pyspark import StorageLevel\nfrom pyspark.sql import SparkSession, types\nfrom pyspark.sql import functions as F\nfrom pyspark.ml.feature import OneHotEncoder, StringIndexer, VectorAssembler\nfrom pyspark.ml import Pipeline\nfrom pyspark.ml.classification import LogisticRegression\n","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:18.253555Z","iopub.execute_input":"2022-06-19T20:03:18.254259Z","iopub.status.idle":"2022-06-19T20:03:18.733212Z","shell.execute_reply.started":"2022-06-19T20:03:18.254199Z","shell.execute_reply":"2022-06-19T20:03:18.732566Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Spark Session\nSpark session is an entry point into all Spark functionality. We will establish it and tune some configurations.","metadata":{}},{"cell_type":"code","source":"# SESSION PARAMETER\nCORES = multiprocessing.cpu_count()\nMAX_PARTITION_SIZE = \"134217728b\"","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:18.734211Z","iopub.execute_input":"2022-06-19T20:03:18.734602Z","iopub.status.idle":"2022-06-19T20:03:18.739603Z","shell.execute_reply.started":"2022-06-19T20:03:18.734566Z","shell.execute_reply":"2022-06-19T20:03:18.738574Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"The additional configurations are based on these refs:\n1. [Configure OffHeap][1] (else get OutOfMemoryError when training model)\n2. [Spark job tuning tips][2]\n\n\n[1]: https://stackoverflow.com/questions/21138751/spark-java-lang-outofmemoryerror-java-heap-space\n[2]: https://cloud.google.com/dataproc/docs/support/spark-job-tuning","metadata":{}},{"cell_type":"code","source":"spark = (SparkSession.builder.master(f\"local[{CORES}]\")\n                             .config(\"spark.memory.offHeap.enabled\", \"true\")\n                             .config(\"spark.memory.offHeap.size\",\"5g\")\n                             .config(\"spark.sql.shuffle.partitions\", CORES * 3)\n                             .config(\"spark.default.parallelism\", CORES * 3)\n                             .config(\"spark.sql.adaptive.advisoryPartitionSizeInBytes\", MAX_PARTITION_SIZE)\n                             .appName(\"ML_spark\")\n                             .getOrCreate())\nspark","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:18.741551Z","iopub.execute_input":"2022-06-19T20:03:18.741870Z","iopub.status.idle":"2022-06-19T20:03:25.684865Z","shell.execute_reply.started":"2022-06-19T20:03:18.741836Z","shell.execute_reply":"2022-06-19T20:03:25.683775Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Read Data","metadata":{}},{"cell_type":"code","source":"train_path = \"../input/amex-pyspark-parquet/train_amex\"\ntest_path = \"../input/amex-pyspark-parquet/test_amex\"\nlabel_path = \"../input/amex-pyspark-parquet/label_amex\"","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:25.687121Z","iopub.execute_input":"2022-06-19T20:03:25.688018Z","iopub.status.idle":"2022-06-19T20:03:25.694756Z","shell.execute_reply.started":"2022-06-19T20:03:25.687962Z","shell.execute_reply":"2022-06-19T20:03:25.693166Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"The data reading speed of pyspark is quick because it didn't read the data as a whole.","metadata":{}},{"cell_type":"code","source":"%%time\ntrain_df = spark.read.parquet(train_path)\ntest_df = spark.read.parquet(test_path)\nlabel_df = spark.read.parquet(label_path)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:25.696997Z","iopub.execute_input":"2022-06-19T20:03:25.697904Z","iopub.status.idle":"2022-06-19T20:03:31.501236Z","shell.execute_reply.started":"2022-06-19T20:03:25.697850Z","shell.execute_reply":"2022-06-19T20:03:31.500190Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Checks on Spark and Data","metadata":{}},{"cell_type":"markdown","source":"### Physical plan of spark","metadata":{}},{"cell_type":"code","source":"train_df.select(\"customer_ID\").explain()  # select one column to simplify the output","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:31.502390Z","iopub.execute_input":"2022-06-19T20:03:31.502817Z","iopub.status.idle":"2022-06-19T20:03:32.189013Z","shell.execute_reply.started":"2022-06-19T20:03:31.502769Z","shell.execute_reply":"2022-06-19T20:03:32.187305Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Missing Customer\nWe need to make sure that any customer isn't missing","metadata":{}},{"cell_type":"code","source":"def get_null_count(sql_df, colname):\n    count = (sql_df.select(colname)\n                   .filter(F.col(colname).isNull())\n                   .count())\n    return count","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:32.190400Z","iopub.execute_input":"2022-06-19T20:03:32.190777Z","iopub.status.idle":"2022-06-19T20:03:32.387311Z","shell.execute_reply.started":"2022-06-19T20:03:32.190746Z","shell.execute_reply":"2022-06-19T20:03:32.386150Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"missing_customer_train = get_null_count(train_df, \"customer_ID\") \nmissing_customer_test = get_null_count(test_df, \"customer_ID\")\n\ntotal_miss = missing_customer_train + missing_customer_test\nprint(f\"Missing customer_ID: {total_miss}\")","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:32.389106Z","iopub.execute_input":"2022-06-19T20:03:32.389482Z","iopub.status.idle":"2022-06-19T20:03:38.822030Z","shell.execute_reply.started":"2022-06-19T20:03:32.389451Z","shell.execute_reply":"2022-06-19T20:03:38.820972Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Sample Data","metadata":{}},{"cell_type":"code","source":"train_df.show(1, vertical=True)\nlabel_df.show(1, vertical=True)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:38.826087Z","iopub.execute_input":"2022-06-19T20:03:38.826841Z","iopub.status.idle":"2022-06-19T20:03:41.414745Z","shell.execute_reply.started":"2022-06-19T20:03:38.826793Z","shell.execute_reply":"2022-06-19T20:03:41.413686Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Preprocessing\nLong story short, in `pyspark.sql`, we can transform or apply function in the columns by using `.withColumn()` method. Almost every functions we need are lies inside `pyspark.sql.functions` namespace (we are gonna use `F` as alias). ","metadata":{}},{"cell_type":"markdown","source":"### Here is how to deal with columns in `pyspark.sql`:\n\n**1. We can select columns by using `.select` method (return only selected columns)**\n\n```\ndf.select([\"col1\", \"col2\"])\n```\n\n<br>\n\n\n**2. Apply function from the selected columns**\n\n```\ndf.select([F.func1(\"col1\"), F.func2(\"col2\")])\n```\n\n<br>\n\n\n**3. Make an alias (the colnames change as you apply a functions)**\n\n```\ndf.select([F.func1(\"col1\").alias(\"col1\"), F.func2(\"col2\").alias(\"col2\")])\n```\n\n\n<br>\n\n\n\n**4. Transform specific column (still retaining df)**\n\n```\ndf.withColumn(\"colname\", F.func(col))\n```\n\n\n<br>\n\n\n**5. Transform multiple columns**\n\n```\ndf.withColumns({\"colname1\", F.func(col1), \"colname2\": F.func(col2)})\n```\n\n<br>\n\n\n\n**6. Rename columns**\n\n```\ndf.withColumnRename(\"colname\", \"newcolname\")\n```","metadata":{}},{"cell_type":"markdown","source":"### Don't quite familiar with pyspark.sql?\nIf you don't quite familiar with pyspark.sql API you can still use pandas API by converting the current dataframe/rdd object into pandas.\n\nTo convert pyspark dataframe to pandas you can use `.toPandas` and `.to_pandas_on_spark` methods.\n```\nspark_df = spark_df.toPandas()\n```\nor\n```\nspark_df = spark_df.to_pandas_on_spark(index_col=\"id\")  # index_col for unlocking multiple partitions\n```\n\nThe drawback is, when the data is in large volume, it will affect the performance and lengthen the computation time. I suggest you to stick with `pyspark.sql` or `pyspark.rdd` until you reduce your data small enough.","metadata":{}},{"cell_type":"markdown","source":"### Utility Functions","metadata":{}},{"cell_type":"code","source":"def add_suffix(names, suffix):\n    return [name + suffix for name in names]","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:41.417517Z","iopub.execute_input":"2022-06-19T20:03:41.418005Z","iopub.status.idle":"2022-06-19T20:03:41.426572Z","shell.execute_reply.started":"2022-06-19T20:03:41.417961Z","shell.execute_reply":"2022-06-19T20:03:41.425615Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Define Columns\nThe data is consist of multiple group of columns, there are:\n- Individual information (id, date)\n- Features (representation of individual)\n- Target (label of individual)","metadata":{}},{"cell_type":"code","source":"# Known Columns\ninfo_cols = ['customer_ID', 'S_2']\ntarget_cols = ['target']\ncat_cols = [\n    'B_30', 'B_38', \n    'D_114', 'D_116', 'D_117', 'D_120', 'D_126', 'D_63', 'D_64', 'D_66', 'D_68']\n\n\n# Define Numeric Columns\nexcluded = info_cols + cat_cols\nnum_cols = [col for col in train_df.columns if col not in excluded]\n\n# Define Feature Columns\nfeatures_cols =  cat_cols + num_cols\n\nprint(f\"Number of categoric cols: {len(cat_cols)}\")\nprint(f\"Number of numeric cols: {len(num_cols)}\")","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:41.434508Z","iopub.execute_input":"2022-06-19T20:03:41.435757Z","iopub.status.idle":"2022-06-19T20:03:41.501599Z","shell.execute_reply.started":"2022-06-19T20:03:41.435700Z","shell.execute_reply":"2022-06-19T20:03:41.500680Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Fill Missing Values\nThere are some columns in this dataframe that have two or more `null` value, our base strategies are:\n- Fill null in numeric columns with 0\n- Fill null in categoric columns with \"null\"","metadata":{}},{"cell_type":"code","source":"train_df = (train_df.fillna(0, subset=num_cols)\n                    .fillna(\"null\", subset=cat_cols))\n\ntest_df = (test_df.fillna(0, subset=num_cols)\n                  .fillna(\"null\", subset=cat_cols))","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:41.503308Z","iopub.execute_input":"2022-06-19T20:03:41.504051Z","iopub.status.idle":"2022-06-19T20:03:42.159282Z","shell.execute_reply.started":"2022-06-19T20:03:41.504007Z","shell.execute_reply":"2022-06-19T20:03:42.158151Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Feature Transform and Engineer\n\n**Data Structure**  \nLet's see how the data is structured, the data is collections of individual customer records. Every customer have multiple records of their profile. The default status is labeled for each individual customer, it means we need to account the features as combination.\n\nExample:\n\n| id         \t| pay \t| remain \t| last       \t| due        \t| label \t|\n|------------\t|-----\t|--------\t|------------\t|------------\t|-------\t|\n| customer_1 \t| 100 \t| 200    \t| 2022-01-01 \t| 2022-02-01 \t|       \t|\n| customer_1 \t| 100 \t| 100    \t| 2022-02-01 \t| 2022-03-01 \t|       \t|\n| customer_1 \t| 100 \t| 0      \t| 2022-02-03 \t| -          \t| 0     \t|\n| customer_2 \t| 10  \t| 290    \t| 2022-01-01 \t| 2022-02-01 \t|       \t|\n| customer_2 \t| 0   \t| 290    \t| 2022-01-01 \t| -           \t| 1     \t|\n\n<br>\n","metadata":{}},{"cell_type":"markdown","source":"### Feature Engineer\ncoming soon","metadata":{}},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Feature Transform\nThe data is a combination of numeric and categoric types. Machine learning model can only take numeric input, therefore we need to convert our categoric columns into numeric type. For this specific task, we will use `OneHotEncoder`.","metadata":{}},{"cell_type":"markdown","source":"#### Index String\nBefore using OneHotEncoding, we need to index our categoric columns into integer. We can use `StringIndexer` for this task.","metadata":{}},{"cell_type":"code","source":"# Create columns aliases\ncat_index_cols = add_suffix(cat_cols, \"_index\")\n\n# Fit StringIndexer\nindexers = StringIndexer(inputCols=cat_cols, outputCols=cat_index_cols)\nindexers_model = indexers.fit(train_df)\n\n# Transform to data\ntrain_df_indexed = indexers_model.transform(train_df)\ntest_df_indexed = indexers_model.transform(test_df)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:42.160661Z","iopub.execute_input":"2022-06-19T20:03:42.161108Z","iopub.status.idle":"2022-06-19T20:03:54.478682Z","shell.execute_reply.started":"2022-06-19T20:03:42.161063Z","shell.execute_reply":"2022-06-19T20:03:54.477638Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# See what columns the indexer handle\nindexers.getInputCols()\n\n# See the indexed columns\ntrain_df_indexed.select(\"B_30_index\").show(5)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:54.479890Z","iopub.execute_input":"2022-06-19T20:03:54.480326Z","iopub.status.idle":"2022-06-19T20:03:54.823673Z","shell.execute_reply.started":"2022-06-19T20:03:54.480284Z","shell.execute_reply":"2022-06-19T20:03:54.822628Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"#### One Hot Encoding\nAfter that, we can apply OneHotEncoder on the indexed categoric columns.","metadata":{}},{"cell_type":"code","source":"# Create columns aliases\ncat_ohe_cols = add_suffix(cat_cols, \"_ohe\")\n\n# Fit OneHotEncoder\nohe = OneHotEncoder(inputCols=cat_index_cols, outputCols=cat_ohe_cols)\nohe_model = ohe.fit(train_df_indexed)\n\n# Transform to data\ntrain_df_ohed = ohe_model.transform(train_df_indexed)\ntest_df_ohed = ohe_model.transform(test_df_indexed)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:54.824954Z","iopub.execute_input":"2022-06-19T20:03:54.825400Z","iopub.status.idle":"2022-06-19T20:03:55.360032Z","shell.execute_reply.started":"2022-06-19T20:03:54.825359Z","shell.execute_reply":"2022-06-19T20:03:55.358943Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"We know that `B_30` has 4 unique values (3+1 after fillna), but the result of `OneHotEncoder` shows only 3 categories. It is caused by `dropLast=True` default argument from `OneHotEncoder`. The doc said, the last category isn't included in the vector because it will make linear dependent vector. \n\nrefer to: [here](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.feature.OneHotEncoder.html?highlight=onehot#pyspark.ml.feature.OneHotEncoder)","metadata":{}},{"cell_type":"code","source":"train_df_ohed.select(\"B_30_ohe\").show(5)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:55.361315Z","iopub.execute_input":"2022-06-19T20:03:55.361799Z","iopub.status.idle":"2022-06-19T20:03:55.839971Z","shell.execute_reply.started":"2022-06-19T20:03:55.361754Z","shell.execute_reply":"2022-06-19T20:03:55.838580Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Group Customer\nAfter the features is fit for machine learning model we need to make a few adjustment about the input. Let's see the data structure:","metadata":{}},{"cell_type":"markdown","source":"**Data Structure**  \nLet's see how the data is structured, the data is collections of individual customer records. Every customer have multiple records of their profile. The default status is labeled for each individual customer, it means that we need to account the features as group.\n\nExample:\n\n| id         \t| pay \t| remain \t| last       \t| due        \t| label \t|\n|------------\t|-----\t|--------\t|------------\t|------------\t|-------\t|\n| customer_1 \t| 100 \t| 200    \t| 2022-01-01 \t| 2022-02-01 \t|       \t|\n| customer_1 \t| 100 \t| 100    \t| 2022-02-01 \t| 2022-03-01 \t|       \t|\n| customer_1 \t| 100 \t| 0      \t| 2022-02-03 \t| -          \t| 0     \t|\n| customer_2 \t| 10  \t| 290    \t| 2022-01-01 \t| 2022-02-01 \t|       \t|\n| customer_2 \t| 0   \t| 290    \t| 2022-01-01 \t| -           \t| 1     \t|\n\n<br>\n\nFrom above explanation, we will group every customer using some reduce functions like: mean, min and max.","metadata":{}},{"cell_type":"markdown","source":"**Aggregate Data**  \nWe need to aggregate features columns with suitable functions for specific data types:\n- Numeric: mean, std, min, max\n- Categoric: count, last, nunique","metadata":{}},{"cell_type":"code","source":"# Functions for each type\n# each tuple consist of: (function, column's suffix)\nnum_funcs = [\n    (F.mean, \"_mean\"),\n    (F.stddev, \"_std\"),\n    (F.min, \"_min\"),\n    (F.max, \"_max\"),\n]\n\ncat_funcs = [\n    (F.count, \"_count\"),\n    (F.last, \"_last\"),\n    (F.countDistinct, \"_nunique\"),\n]","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:55.841309Z","iopub.execute_input":"2022-06-19T20:03:55.842358Z","iopub.status.idle":"2022-06-19T20:03:55.851595Z","shell.execute_reply.started":"2022-06-19T20:03:55.842304Z","shell.execute_reply":"2022-06-19T20:03:55.850806Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Arguments for .agg method\n# each arg consist of: func(colname).alias(colname + suffix)\nagg_num_args = [\n    func(col).alias(col + suffix) \n    for col, (func, suffix) in itertools.product(num_cols, num_funcs)]\n\nagg_cols_args = [\n    func(col).alias(col + suffix) \n    for col, (func, suffix) in itertools.product(cat_ohe_cols, cat_funcs)]\n\n# Combine numeric and categoric agg arguments\nagg_args = agg_num_args + agg_cols_args\nagg_args[0]","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:55.852942Z","iopub.execute_input":"2022-06-19T20:03:55.853681Z","iopub.status.idle":"2022-06-19T20:03:57.880853Z","shell.execute_reply.started":"2022-06-19T20:03:55.853635Z","shell.execute_reply":"2022-06-19T20:03:57.879939Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Columns that we won't use\nunused_cols = cat_cols + num_cols + cat_index_cols + cat_ohe_cols\nprint(f\"Unused columns {len(unused_cols)}\")","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:57.882407Z","iopub.execute_input":"2022-06-19T20:03:57.882780Z","iopub.status.idle":"2022-06-19T20:03:57.888465Z","shell.execute_reply.started":"2022-06-19T20:03:57.882748Z","shell.execute_reply":"2022-06-19T20:03:57.887489Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Apply the agg while also dropping unused columns\ntrain_df_grouped = (train_df_ohed.groupBy(\"customer_ID\")\n                                 .agg(*agg_cols_args)\n                                 .drop(*unused_cols))\n\ntest_df_grouped = (test_df_ohed.groupBy(\"customer_ID\")\n                                .agg(*agg_cols_args)\n                                .drop(*unused_cols))","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:57.889892Z","iopub.execute_input":"2022-06-19T20:03:57.890900Z","iopub.status.idle":"2022-06-19T20:03:58.100499Z","shell.execute_reply.started":"2022-06-19T20:03:57.890861Z","shell.execute_reply":"2022-06-19T20:03:58.099446Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Modelling","metadata":{}},{"cell_type":"markdown","source":"### Join label and train data\nUse broadcast function","metadata":{}},{"cell_type":"code","source":"train_joined_df = train_df_grouped.join(F.broadcast(label_df), on=\"customer_ID\")","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:58.101774Z","iopub.execute_input":"2022-06-19T20:03:58.102116Z","iopub.status.idle":"2022-06-19T20:03:58.168666Z","shell.execute_reply.started":"2022-06-19T20:03:58.102085Z","shell.execute_reply":"2022-06-19T20:03:58.167604Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"dim = len(train_joined_df.columns)\nprint(f\"Total features: {dim}\")","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:58.170179Z","iopub.execute_input":"2022-06-19T20:03:58.170764Z","iopub.status.idle":"2022-06-19T20:03:58.189350Z","shell.execute_reply.started":"2022-06-19T20:03:58.170718Z","shell.execute_reply":"2022-06-19T20:03:58.188616Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Assemble Vector","metadata":{}},{"cell_type":"code","source":"va = VectorAssembler(\n    inputCols=train_joined_df.drop(\"customer_ID\", \"target\").columns,\n    outputCol=\"features\",\n    handleInvalid=\"error\",\n)\n\ntrain_ready_df = (va.transform(train_joined_df)\n                    .select([\"customer_ID\", \"features\", \"target\"])\n                    .persist(StorageLevel.DISK_ONLY))\n\ntest_ready_df = (va.transform(test_df_grouped)\n                   .select([\"customer_ID\", \"features\"])\n                   .persist(StorageLevel.DISK_ONLY))","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:03:58.192334Z","iopub.execute_input":"2022-06-19T20:03:58.193491Z","iopub.status.idle":"2022-06-19T20:05:45.626661Z","shell.execute_reply.started":"2022-06-19T20:03:58.193433Z","shell.execute_reply":"2022-06-19T20:05:45.625288Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Training","metadata":{}},{"cell_type":"code","source":"logres = LogisticRegression(featuresCol=\"features\", labelCol=\"target\")\nlogres_model = logres.fit(train_ready_df)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:14:24.454322Z","iopub.execute_input":"2022-06-19T20:14:24.454900Z","iopub.status.idle":"2022-06-19T20:14:47.707748Z","shell.execute_reply.started":"2022-06-19T20:14:24.454849Z","shell.execute_reply":"2022-06-19T20:14:47.706616Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Prediction","metadata":{}},{"cell_type":"code","source":"test_predictions = logres_model.transform(test_ready_df)\ntest_predictions","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:54:08.256334Z","iopub.execute_input":"2022-06-19T20:54:08.257234Z","iopub.status.idle":"2022-06-19T20:54:08.335407Z","shell.execute_reply.started":"2022-06-19T20:54:08.257182Z","shell.execute_reply":"2022-06-19T20:54:08.334202Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Submission","metadata":{}},{"cell_type":"code","source":"submission = (test_predictions.select([\"customer_ID\", \"probability\"])\n                              .toPandas())","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:55:29.100331Z","iopub.execute_input":"2022-06-19T20:55:29.100757Z","iopub.status.idle":"2022-06-19T20:56:21.861892Z","shell.execute_reply.started":"2022-06-19T20:55:29.100725Z","shell.execute_reply":"2022-06-19T20:56:21.860638Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Get probability of target 1\nsubmission[\"probability\"] = submission[\"probability\"].apply(lambda x: x[-1])\n\n# Rename columns\nsubmission = submission.rename(columns={\"probability\": \"prediction\"})\n\n# Save File\nsubmission.to_csv(\"submission.csv\", index=False)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T21:07:30.633568Z","iopub.execute_input":"2022-06-19T21:07:30.634073Z","iopub.status.idle":"2022-06-19T21:07:35.009189Z","shell.execute_reply.started":"2022-06-19T21:07:30.634039Z","shell.execute_reply":"2022-06-19T21:07:35.008195Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"pd.read_csv(\"submission.csv\").head()","metadata":{"execution":{"iopub.status.busy":"2022-06-19T21:07:59.895344Z","iopub.execute_input":"2022-06-19T21:07:59.895862Z","iopub.status.idle":"2022-06-19T21:08:01.430319Z","shell.execute_reply.started":"2022-06-19T21:07:59.895818Z","shell.execute_reply":"2022-06-19T21:08:01.429603Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# training_summary = logres_model.summary\n\n# # Obtain the objective per iteration\n# objective_history = training_summary.objectiveHistory\n# print(\"objective history:\")\n# for objective in objective_history:\n#     print(objective)\n\n# # Obtain the receiver-operating characteristic as a dataframe and areaUnderROC.\n# training_summary.roc.show()\n# print(\"areaUnderROC: \" + str(training_summary.areaUnderROC))\n\n# # Set the model threshold to maximize F-Measure\n# f_measure = training_summary.fMeasureByThreshold\n# maxf_measure = (f_measure.groupBy()\n#                          .max('F-Measure')\n#                          .select('max(F-Measure)')\n#                          .head())\n\n# best_threshold = (f_measure.where(f_measure['F-Measure'] == maxf_measure['max(F-Measure)'])\n#                            .select('threshold')\n#                            .head()['threshold'])\n\n# print(f\"Best threshold: {best_threshold}\")\n\n# logres.setThreshold(best_threshold)","metadata":{"execution":{"iopub.status.busy":"2022-06-19T20:20:39.794476Z","iopub.execute_input":"2022-06-19T20:20:39.794969Z","iopub.status.idle":"2022-06-19T20:20:40.046377Z","shell.execute_reply.started":"2022-06-19T20:20:39.794934Z","shell.execute_reply":"2022-06-19T20:20:40.045448Z"},"collapsed":true,"jupyter":{"outputs_hidden":true},"trusted":true},"execution_count":null,"outputs":[]}]}