etl_pyspark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "4602b805",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.ml.feature import Imputer\n",
    "from pyspark.sql.types import FloatType, ArrayType\n",
    "from pyspark.sql.functions import col, to_date, year, month, regexp_like, \\\n",
    "                                    regexp_extract, isnan, when, count, lit, udf, \\\n",
    "                                    pow as psf_pow, sum as psf_sum"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "2a868e6e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import warnings\n",
    "warnings.filterwarnings('ignore')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "b33dad2e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# spark url: localhost:4040\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"LoanEtl\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "ca30d593",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark \\\n",
    "    .read \\\n",
    "    .option('header', \"true\") \\\n",
    "    .option(\"inferSchema\", \"true\") \\\n",
    "    .csv('data/loan_10%.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0eb4faea",
   "metadata": {},
   "source": [
    "rename some columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "1fb91cd0",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "find = [\n",
    "    'mths_since_rcnt_il', 'mo_sin_rcnt_rev_tl_op', 'mo_sin_rcnt_tl', # rcnt and mo_sin_rcnt\n",
    "    'mths_since_recent_bc_dlq', 'pct_tl_nvr_dlq', # dlq\n",
    "    'mo_sin_old_il_acct', 'mo_sin_old_rev_tl_op', # mo_sin_old\n",
    "]\n",
    "replace = [\n",
    "    'mths_since_recent_il', 'mths_since_recent_rev_tl_op', 'mths_since_recent_tl', # recent and mths_since\n",
    "    'mths_since_recent_bc_delinq', 'percent_tl_nvr_deliq', # delinq\n",
    "    'mths_since_old_il_acct', 'mths_since_old_rev_tl_op' # mths_since_old\n",
    "]\n",
    "\n",
    "for f, r in zip(find, replace):\n",
    "    df = df.withColumnRenamed(f, r)\n",
    "\n",
    "del find, replace"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7319f3ae",
   "metadata": {},
   "source": [
    "convert to datetime"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "921e4ceb",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Row(issue_d='Aug-2015')\n",
      "2015-08-01\n"
     ]
    }
   ],
   "source": [
    "# example\n",
    "\n",
    "# original\n",
    "print(df.select('issue_d').first())\n",
    "# want-to-be\n",
    "print(df.select(to_date(df['issue_d'], \"MMM-yyyy\")).first()[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "5e1b0a8d",
   "metadata": {},
   "outputs": [],
   "source": [
    "datetime_columns = ['issue_d', 'earliest_cr_line', 'last_pymnt_d', 'next_pymnt_d',\n",
    "                    'last_credit_pull_d', 'sec_app_earliest_cr_line', 'hardship_start_date',\n",
    "                    'hardship_end_date', 'payment_plan_start_date']\n",
    "\n",
    "for dt_col in datetime_columns:\n",
    "    df = df.withColumn(dt_col, to_date(df[dt_col], \"MMM-yyyy\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9a6d2f6e",
   "metadata": {},
   "source": [
    "remove loan data issued before 1-1-2013"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "9a0dec8e",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.filter(year('issue_d') > 2012)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba6111b7",
   "metadata": {},
   "source": [
    "modify column int_rate and revol_util: changing from \"xx.xx%\" to xx.xx"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "ba1bcf32",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('int_rate', regexp_extract('int_rate', r'(\\d+\\.\\d+)', 1).cast('float'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "618710b0",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(int_rate=7.889999866485596)"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('int_rate').first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "a75895e7",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('revol_util', regexp_extract('revol_util', r'(\\d+\\.\\d+)', 1).cast(\"float\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "2473e30a",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(revol_util=25.100000381469727)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('revol_util').first()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d8d23a48",
   "metadata": {},
   "source": [
    "create a column for deciding if it is matured loan; empty row = matured"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "4b4048e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('is_matured_loan', col('next_pymnt_d').isNull())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46d09b25",
   "metadata": {},
   "source": [
    "I only analyzed the loan data that is matured"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "7fc92c16",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.filter(df['is_matured_loan'])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "16bc5396",
   "metadata": {},
   "source": [
    "handle na / null data in dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "4b1eb125",
   "metadata": {},
   "outputs": [],
   "source": [
    "# turned into binary: missing = 0; existing = 1\n",
    "mths_since_to_drop = [\n",
    "    col_name for col_name in df.columns \\\n",
    "    if col_name not in ['mths_since_recent_rev_tl_op', 'mths_since_recent_tl'] \\\n",
    "    and 'mths_since_' in col_name\n",
    "]\n",
    "\n",
    "for col_name in mths_since_to_drop:\n",
    "    df = df.withColumn(col_name,\n",
    "                       when(col(col_name).isNull(), 0) \\\n",
    "                       .otherwise(1))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "669b84de",
   "metadata": {},
   "outputs": [],
   "source": [
    "# filled by zero\n",
    "df = df \\\n",
    "    .withColumn('deferral_term', when(col('deferral_term').isNull(), 0) \\\n",
    "    .otherwise(col('deferral_term')).cast(\"integer\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "666a8ac2",
   "metadata": {},
   "outputs": [],
   "source": [
    "# filled by mean\n",
    "col_fill_by_mean = ['il_util', 'all_util', 'total_bal_il', 'max_bal_bc', \n",
    "                    'percent_bc_gt_75', 'bc_util', 'revol_util', 'dti',\n",
    "                    'percent_tl_nvr_deliq', 'avg_cur_bal', 'bc_open_to_buy']\n",
    "\n",
    "imputer = Imputer(strategy='mean', inputCols=col_fill_by_mean, outputCols=col_fill_by_mean)\n",
    "imputer = imputer.fit(df)\n",
    "df = imputer.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "e5af1350",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# filled by mode (numeric)\n",
    "col_fill_by_mode = ['open_acc_6m', 'open_act_il', 'open_il_12m', 'open_il_24m',\n",
    "                    'open_rv_12m', 'open_rv_24m', 'inq_fi', 'total_cu_tl', 'inq_last_12m',\n",
    "                    'num_tl_120dpd_2m', 'num_rev_accts', 'inq_last_6mths']\n",
    "\n",
    "imputer = Imputer(strategy='mode', inputCols=col_fill_by_mode, outputCols=col_fill_by_mode)\n",
    "imputer = imputer.fit(df)\n",
    "df = imputer.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "80edfcad",
   "metadata": {},
   "outputs": [],
   "source": [
    "# filled by mode (string)\n",
    "for col_name in ['emp_length', 'zip_code', 'hardship_flag']:\n",
    "    most_common = df.freqItems([col_name]).first()[0][0]\n",
    "    df = df \\\n",
    "        .withColumn(col_name, when(col(col_name).isNull(), most_common) \\\n",
    "        .otherwise(col(col_name)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "b8df8cf3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# filled by empty\n",
    "col_fill_by_empty_str = ['emp_title', 'title', 'purpose']\n",
    "\n",
    "for col_name in col_fill_by_empty_str:\n",
    "    df = df \\\n",
    "        .withColumn(col_name, when(col(col_name).isNull(), '') \\\n",
    "        .otherwise(col(col_name)))\n",
    "    \n",
    "# filled by df.na.fill('') for string columns (I guess)\n",
    "# https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.fillna.html"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c53f2a9a",
   "metadata": {},
   "source": [
    "create columns for modelling the probablilty of default"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "0c9025d0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# scale some columns by annual_inc\n",
    "for col_name in ['loan_amnt', 'delinq_amnt', 'revol_bal', 'tot_cur_bal', \n",
    "                 'max_bal_bc', 'avg_cur_bal', 'total_bal_ex_mort']:\n",
    "    \n",
    "    df = df.withColumn(col_name + '_to_' + 'annual_inc', \n",
    "                       col(col_name) / (col('annual_inc') + 1))\n",
    "    \n",
    "# ratio of the number of open credit lines to the number of \n",
    "# total credit lines in the borrower's credit file.\n",
    "df = df.withColumn('open_cl_ratio', col('open_acc') / col('total_acc'))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "33358f94",
   "metadata": {},
   "source": [
    "create column: \"expected\"/realized installment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "223c9a3c",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df \\\n",
    "    .withColumn('num_mths_to_pay',\n",
    "                (year(col('last_pymnt_d')) - year(col('issue_d'))) * 12\n",
    "                + month(col('last_pymnt_d')) - month(col('issue_d')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "fc9b1ac6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# logic:\n",
    "# if last_pymnt_d not null => no need to fill => mnths_diff\n",
    "# else\n",
    "#   if total pymnt > 0 => total_pymnt / installment => approx mnths_diff\n",
    "#   else 0 => no payment made\n",
    "\n",
    "df = df.withColumn(\n",
    "    'num_mths_to_pay',\n",
    "    when(\n",
    "        col('last_pymnt_d').isNull(),\n",
    "        when(\n",
    "            col('total_pymnt') > 0, \n",
    "            col('total_pymnt') / col('installment')) \\\n",
    "        .otherwise(0)) \\\n",
    "    .otherwise(col('num_mths_to_pay')).cast(\"integer\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "6d5267b4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a term integer column and drop soon\n",
    "df = df.withColumn('term_int', regexp_extract('term', r'(\\d\\d)', 1).cast('integer'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "5c8fc612",
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_installment(p, r, n):\n",
    "    r = r / 100\n",
    "    return p * (r / 12) * (1 + r / 12) ** n / ((1 + r / 12) ** n - 1)\n",
    "\n",
    "df = df.withColumn(\n",
    "    'calculated_installment',\n",
    "    calculate_installment(col('funded_amnt'), col('int_rate'), col('term_int'))\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "95ee83fb",
   "metadata": {},
   "outputs": [],
   "source": [
    "# scale some installment columns by annual_inc\n",
    "for col_name in ['installment', 'calculated_installment']:\n",
    "    \n",
    "    df = df.withColumn(col_name + '_to_' + 'annual_inc', \n",
    "                       col(col_name) / (col('annual_inc') + 1))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "11f46a17",
   "metadata": {},
   "source": [
    "create column: ROI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "a3af1a32",
   "metadata": {},
   "outputs": [],
   "source": [
    "# assume\n",
    "depreciation_rate = 0.02"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "0355e633",
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_avg_pymnt(total_pymnt, num_mths_to_pay, last_pymnt_amnt):\n",
    "    return when(num_mths_to_pay == 0, last_pymnt_amnt) \\\n",
    "            .otherwise(col('total_pymnt') / col('num_mths_to_pay'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "1ef7e587",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\n",
    "    'avg_pymnt', \n",
    "    calculate_avg_pymnt(\n",
    "        col('total_pymnt'),\n",
    "        col('num_mths_to_pay'), \n",
    "        col('last_pymnt_amnt')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "eda5d8f3",
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_cdf(num, depreciation_rate=0.02):\n",
    "    return sum([(1 + depreciation_rate / 12) ** (-t) for t in range(1, num + 1)])\n",
    "\n",
    "udf_calculate_cdf = udf(calculate_cdf, returnType=FloatType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "f322fd81",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('cdf', udf_calculate_cdf(col('num_mths_to_pay')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "dcdd9854",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "df = df.withColumn('CDP', col('cdf') * col('avg_pymnt'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "f3aeff3c",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\n",
    "    'CDP',\n",
    "    when(\n",
    "        (col('avg_pymnt') > 0)& \\\n",
    "        (col('num_mths_to_pay') == 0), \n",
    "        col('avg_pymnt')\n",
    "    ).otherwise(col('CDP')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "48a04543",
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate_discounted_net_recoveries(depreciation_rate=0.02):\n",
    "    return (\n",
    "        (col('recoveries') - col('collection_recovery_fee')) \\\n",
    "        * 1 / (1 + depreciation_rate / 12)**(col('num_mths_to_pay') + 6)\n",
    "    )\n",
    "    \n",
    "df = df.withColumn('CDP', col('CDP') + calculate_discounted_net_recoveries())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "12c3feff",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\n",
    "    'ROI', \n",
    "    (col('CDP') - col('funded_amnt')) \\\n",
    "    / col('funded_amnt'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "35d3d3a1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(ROI=0.08976702591577829)"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('ROI').first()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e9765583",
   "metadata": {},
   "source": [
    "create column: IRR"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "4657ecfd",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn(\n",
    "    'num_mths_to_pay',\n",
    "    when(\n",
    "        (regexp_like('loan_status', lit(r'(Fully)')))& \\\n",
    "        (col('issue_d') == col('last_pymnt_d')),\n",
    "        1\n",
    "    ).otherwise(col('num_mths_to_pay')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "63de8dc3",
   "metadata": {},
   "outputs": [],
   "source": [
    "def make_list_of_cashflow(f, p, n):\n",
    "    return [-f] + [p] * n\n",
    "\n",
    "udf_make_list_of_cashflow = udf(make_list_of_cashflow, returnType=ArrayType(FloatType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "d42e8b05",
   "metadata": {},
   "outputs": [],
   "source": [
    "# equivalently, it means \n",
    "# df.apply(lambda r: [-r['funded_amnt']] + [r['avg_pymnt']] * r['num_mths_to_pay'], axis=1)\n",
    "\n",
    "df = df.withColumn(\n",
    "    'lst_of_cf', \n",
    "    udf_make_list_of_cashflow(\n",
    "        col('funded_amnt'), \n",
    "        col('avg_pymnt'), \n",
    "        col('num_mths_to_pay')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "7009674c",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(lst_of_cf=[-8500.0, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656, 364.3417053222656])"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select('lst_of_cf').first()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "021683b8",
   "metadata": {},
   "outputs": [],
   "source": [
    "def irr(cash_flow):\n",
    "    \"\"\"\n",
    "    Solving this to get IRR:\n",
    "    p0 + p1 * (1 + IRR)^(-1) + p2 * (1 + IRR)^(-2) + ... + pn * (1 + IRR)^(-n) = 0 _ (*)\n",
    "\n",
    "    Let g = 1 + IRR and multiply (*) by g^n and you will get:\n",
    "    p0 * g^n + p1 * g^(n-1) + p2 * g^(n-2) + ... + pn = 0\n",
    "\n",
    "    Then solve the polynomial and get the positive roots\n",
    "    IRR = g - 1\n",
    "\n",
    "    :param: cash_flow: a stream of cash flow to pay and receive with a NPV = 0\n",
    "    \"\"\"\n",
    "    cash_flow = np.array(cash_flow)\n",
    "    same_sign = np.all(cash_flow > 0) if cash_flow[0] > 0 else np.all(cash_flow < 0)\n",
    "    if same_sign:\n",
    "        if len(cash_flow) == 1 and cash_flow[0] < 0:\n",
    "            return -1\n",
    "        \n",
    "        raise ValueError(\"cash_flow must contain positive and negative values\")\n",
    "\n",
    "    g = np.roots(cash_flow)\n",
    "    IRR = np.real(g[np.isreal(g)]) - 1\n",
    "\n",
    "    # realistic IRR\n",
    "    IRR = IRR[(IRR <= 1)&(IRR >= -1)]\n",
    "\n",
    "    # if no real solution\n",
    "    if len(IRR) == 0:\n",
    "        return np.nan\n",
    "\n",
    "    # if only one real solution\n",
    "    if len(IRR) == 1:\n",
    "        return IRR[0]\n",
    "\n",
    "    # check sign of all IRR solutions\n",
    "    same_sign = np.all(IRR > 0) if IRR[0] > 0 else np.all(IRR < 0)\n",
    "    \n",
    "    # if not the same, first filter potential IRR by comparing the total positive and negative cash flows\n",
    "    if not same_sign:\n",
    "        pos = sum(cash_flow[cash_flow>0])\n",
    "        neg = sum(cash_flow[cash_flow<0])\n",
    "        if pos >= neg:\n",
    "            IRR = IRR[IRR >= 0]\n",
    "        else:\n",
    "            IRR = IRR[IRR < 0]\n",
    "    \n",
    "    # pick the smallest one in magnitude and return\n",
    "    abs_IRR = np.abs(IRR)\n",
    "    return IRR[np.argmin(abs_IRR)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "f66ccdf9",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "48c1ead8",
   "metadata": {},
   "outputs": [],
   "source": [
    "irr_rdd = df.rdd.map(lambda x: (float(irr(x[-1])),))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "id": "0d174e3b",
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = StructType([StructField(\"IRR\", FloatType(), True)])\n",
    "irr_df = irr_rdd.toDF(schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "6fb386ae",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.join(irr_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c5ffc76b",
   "metadata": {},
   "source": [
    "drop irrelevant columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "0556cd13",
   "metadata": {},
   "outputs": [],
   "source": [
    "to_drop = [\n",
    "    col for col in df.columns if \n",
    "        ('hardship' in col and 'hardship_flag' != col) or \n",
    "        ('mths_since_' in col and '_binary' not in col) or \n",
    "        ('sec_app' in col) or \n",
    "        ('joint' in col)\n",
    "        \n",
    "] + [\n",
    "    'orig_projected_additional_accrued_interest', \n",
    "    'payment_plan_start_date', 'pymnt_plan',\n",
    "    'issue_d_year', 'policy_code',\n",
    "    'next_pymnt_d', 'earliest_cr_line',\n",
    "    'url', 'last_credit_pull_d', 'last_pymnt_d'\n",
    "    \n",
    "] + mths_since_to_drop + [\n",
    "    'lst_of_cf', 'cdf', 'CDP', 'is_matured_loan'\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "id": "dbc338fb",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.drop(*to_drop)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d3fefd2b",
   "metadata": {},
   "source": [
    "check na count \n",
    "- tried: count can run so long. so i collect it back work as a chkpt and turn it back to df (failed)\n",
    "    - potential reason: can't handle such a large list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "id": "35d80562",
   "metadata": {},
   "outputs": [],
   "source": [
    "# not going to do now\n",
    "# # Set a checkpoint directory\n",
    "# spark.sparkContext.setCheckpointDir(\"hdfs://your_checkpoint_directory\")\n",
    "\n",
    "# # Perform checkpointing on a DataFrame\n",
    "# df_checkpointed = df.checkpoint()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3f649b9",
   "metadata": {},
   "source": [
    "binning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "id": "5ec38bcc",
   "metadata": {},
   "outputs": [],
   "source": [
    "def binning_int_rate(int_rate):\n",
    "    if int_rate < 10:\n",
    "        return 'less than 10%'\n",
    "    for i in range(2, 6):\n",
    "        if i*5 <= int_rate and (i+1)*5 > int_rate:\n",
    "            return f'within {i*5}% and {i*5+5}%'\n",
    "    return 'greater than or equal to 30%'\n",
    "\n",
    "udf_binning_int_rate = udf(binning_int_rate)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "id": "698c6cb6",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('int_rate_bin', udf_binning_int_rate(col('int_rate')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "id": "a5fe8316",
   "metadata": {},
   "outputs": [],
   "source": [
    "def binning_income(income):\n",
    "    if income < 20000:\n",
    "        return 'less than 20k'\n",
    "    for i in range(2, 16):\n",
    "        if i*10000 <= income and (i+1)*10000 > income:\n",
    "            return f'within {i*10}k and {(i+1)*10}k'\n",
    "    return f'greater than or equal to {(i+1)*10}k'\n",
    "\n",
    "udf_binning_income = udf(binning_income)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "id": "e3bf064b",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('annual_inc_bin', udf_binning_income(col('annual_inc')))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "58e1edcc",
   "metadata": {},
   "source": [
    "load to local and save as csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "id": "23802aa2",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "Project (13)\n",
      "+- BatchEvalPython (12)\n",
      "   +- CartesianProduct Inner (11)\n",
      "      :- Project (9)\n",
      "      :  +- Project (8)\n",
      "      :     +- BatchEvalPython (7)\n",
      "      :        +- Project (6)\n",
      "      :           +- Project (5)\n",
      "      :              +- * Project (4)\n",
      "      :                 +- * Project (3)\n",
      "      :                    +- * Filter (2)\n",
      "      :                       +- Scan csv  (1)\n",
      "      +- * Scan ExistingRDD (10)\n",
      "\n",
      "\n",
      "(1) Scan csv \n",
      "Output [96]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#22, installment#23, grade#24, sub_grade#25, emp_title#26, emp_length#27, home_ownership#28, annual_inc#29, verification_status#30, issue_d#31, loan_status#32, purpose#35, title#36, zip_code#37, addr_state#38, dti#39, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#44, open_acc#47, pub_rec#48, revol_bal#49, revol_util#50, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#62, last_pymnt_amnt#63, next_pymnt_d#64, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#78, open_act_il#79, open_il_12m#80, open_il_24m#81, total_bal_il#83, il_util#84, open_rv_12m#85, open_rv_24m#86, max_bal_bc#87, all_util#88, total_rev_hi_lim#89, inq_fi#90, total_cu_tl#91, inq_last_12m#92, acc_open_past_24mths#93, avg_cur_bal#94, bc_open_to_buy#95, bc_util#96, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#115, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#118, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, pct_tl_nvr_dlq#122, percent_bc_gt_75#123, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#142, deferral_term#146, debt_settlement_flag#157]\n",
      "Batched: false\n",
      "Location: InMemoryFileIndex [file:/G:/lending-club-credit-risk-analysis/data/loan_10%.csv]\n",
      "ReadSchema: struct<id:int,loan_amnt:double,funded_amnt:double,funded_amnt_inv:double,term:string,int_rate:string,installment:double,grade:string,sub_grade:string,emp_title:string,emp_length:string,home_ownership:string,annual_inc:double,verification_status:string,issue_d:string,loan_status:string,purpose:string,title:string,zip_code:string,addr_state:string,dti:double,delinq_2yrs:double,fico_range_low:double,fico_range_high:double,inq_last_6mths:double,open_acc:double,pub_rec:double,revol_bal:double,revol_util:string,total_acc:double,initial_list_status:string,out_prncp:double,out_prncp_inv:double,total_pymnt:double,total_pymnt_inv:double,total_rec_prncp:double,total_rec_int:double,total_rec_late_fee:double,recoveries:double,collection_recovery_fee:double,last_pymnt_d:string,last_pymnt_amnt:double,next_pymnt_d:string,last_fico_range_high:double,last_fico_range_low:double,collections_12_mths_ex_med:double,application_type:string,acc_now_delinq:double,tot_coll_amt:double,tot_cur_bal:double,open_acc_6m:double,open_act_il:double,open_il_12m:double,open_il_24m:double,total_bal_il:double,il_util:double,open_rv_12m:double,open_rv_24m:double,max_bal_bc:double,all_util:double,total_rev_hi_lim:double,inq_fi:double,total_cu_tl:double,inq_last_12m:double,acc_open_past_24mths:double,avg_cur_bal:double,bc_open_to_buy:double,bc_util:double,chargeoff_within_12_mths:double,delinq_amnt:double,mort_acc:double,num_accts_ever_120_pd:double,num_actv_bc_tl:double,num_actv_rev_tl:double,num_bc_sats:double,num_bc_tl:double,num_il_tl:double,num_op_rev_tl:double,num_rev_accts:double,num_rev_tl_bal_gt_0:double,num_sats:double,num_tl_120dpd_2m:double,num_tl_30dpd:double,num_tl_90g_dpd_24m:double,num_tl_op_past_12m:double,pct_tl_nvr_dlq:double,percent_bc_gt_75:double,pub_rec_bankruptcies:double,tax_liens:double,tot_hi_cred_lim:double,total_bal_ex_mort:double,total_bc_limit:double,total_il_high_credit_limit:double,hardship_flag:string,deferral_term:double,debt_settlement_flag:string>\n",
      "\n",
      "(2) Filter [codegen id : 1]\n",
      "Input [96]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#22, installment#23, grade#24, sub_grade#25, emp_title#26, emp_length#27, home_ownership#28, annual_inc#29, verification_status#30, issue_d#31, loan_status#32, purpose#35, title#36, zip_code#37, addr_state#38, dti#39, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#44, open_acc#47, pub_rec#48, revol_bal#49, revol_util#50, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#62, last_pymnt_amnt#63, next_pymnt_d#64, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#78, open_act_il#79, open_il_12m#80, open_il_24m#81, total_bal_il#83, il_util#84, open_rv_12m#85, open_rv_24m#86, max_bal_bc#87, all_util#88, total_rev_hi_lim#89, inq_fi#90, total_cu_tl#91, inq_last_12m#92, acc_open_past_24mths#93, avg_cur_bal#94, bc_open_to_buy#95, bc_util#96, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#115, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#118, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, pct_tl_nvr_dlq#122, percent_bc_gt_75#123, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#142, deferral_term#146, debt_settlement_flag#157]\n",
      "Condition : ((year(cast(gettimestamp(issue_d#31, MMM-yyyy, TimestampType, Some(America/New_York), false) as date)) > 2012) AND isnull(cast(gettimestamp(next_pymnt_d#64, MMM-yyyy, TimestampType, Some(America/New_York), false) as date)))\n",
      "\n",
      "(3) Project [codegen id : 1]\n",
      "Output [95]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, cast(regexp_extract(int_rate#22, (\\d+\\.\\d+), 1) as float) AS int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#26, emp_length#27, home_ownership#28, annual_inc#29, verification_status#30, cast(gettimestamp(issue_d#31, MMM-yyyy, TimestampType, Some(America/New_York), false) as date) AS issue_d#1440, loan_status#32, purpose#35, title#36, zip_code#37, addr_state#38, dti#39, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#44, open_acc#47, pub_rec#48, revol_bal#49, cast(regexp_extract(revol_util#50, (\\d+\\.\\d+), 1) as float) AS revol_util#2863, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, cast(gettimestamp(last_pymnt_d#62, MMM-yyyy, TimestampType, Some(America/New_York), false) as date) AS last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#78, open_act_il#79, open_il_12m#80, open_il_24m#81, total_bal_il#83, il_util#84, open_rv_12m#85, open_rv_24m#86, max_bal_bc#87, all_util#88, total_rev_hi_lim#89, inq_fi#90, total_cu_tl#91, inq_last_12m#92, acc_open_past_24mths#93, avg_cur_bal#94, bc_open_to_buy#95, bc_util#96, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#115, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#118, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, pct_tl_nvr_dlq#122 AS percent_tl_nvr_deliq#1009, percent_bc_gt_75#123, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#142, CASE WHEN isnull(deferral_term#146) THEN 0 ELSE cast(deferral_term#146 as int) END AS deferral_term#4581, debt_settlement_flag#157]\n",
      "Input [96]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#22, installment#23, grade#24, sub_grade#25, emp_title#26, emp_length#27, home_ownership#28, annual_inc#29, verification_status#30, issue_d#31, loan_status#32, purpose#35, title#36, zip_code#37, addr_state#38, dti#39, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#44, open_acc#47, pub_rec#48, revol_bal#49, revol_util#50, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#62, last_pymnt_amnt#63, next_pymnt_d#64, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#78, open_act_il#79, open_il_12m#80, open_il_24m#81, total_bal_il#83, il_util#84, open_rv_12m#85, open_rv_24m#86, max_bal_bc#87, all_util#88, total_rev_hi_lim#89, inq_fi#90, total_cu_tl#91, inq_last_12m#92, acc_open_past_24mths#93, avg_cur_bal#94, bc_open_to_buy#95, bc_util#96, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#115, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#118, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, pct_tl_nvr_dlq#122, percent_bc_gt_75#123, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#142, deferral_term#146, debt_settlement_flag#157]\n",
      "\n",
      "(4) Project [codegen id : 1]\n",
      "Output [99]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, CASE WHEN isnull(emp_title#26) THEN  ELSE emp_title#26 END AS emp_title#6612, CASE WHEN isnull(emp_length#27) THEN < 1 year ELSE emp_length#27 END AS emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, CASE WHEN isnull(purpose#35) THEN  ELSE purpose#35 END AS purpose#6898, CASE WHEN isnull(title#36) THEN  ELSE title#36 END AS title#6755, CASE WHEN isnull(zip_code#37) THEN 773xx ELSE zip_code#37 END AS zip_code#6146, addr_state#38, CASE WHEN isnull(dti#39) THEN 18.77526641018663 WHEN (dti#39 = NaN) THEN 18.77526641018663 ELSE dti#39 END AS dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, CASE WHEN isnull(inq_last_6mths#44) THEN 0.0 WHEN (inq_last_6mths#44 = NaN) THEN 0.0 ELSE inq_last_6mths#44 END AS inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, CASE WHEN isnull(cast(revol_util#2863 as double)) THEN 50.23436 WHEN (revol_util#2863 = NaN) THEN 50.23436 ELSE cast(cast(revol_util#2863 as double) as float) END AS revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, CASE WHEN isnull(open_acc_6m#78) THEN 0.0 WHEN (open_acc_6m#78 = NaN) THEN 0.0 ELSE open_acc_6m#78 END AS open_acc_6m#5348, CASE WHEN isnull(open_act_il#79) THEN 1.0 WHEN (open_act_il#79 = NaN) THEN 1.0 ELSE open_act_il#79 END AS open_act_il#5349, CASE WHEN isnull(open_il_12m#80) THEN 0.0 WHEN (open_il_12m#80 = NaN) THEN 0.0 ELSE open_il_12m#80 END AS open_il_12m#5350, CASE WHEN isnull(open_il_24m#81) THEN 1.0 WHEN (open_il_24m#81 = NaN) THEN 1.0 ELSE open_il_24m#81 END AS open_il_24m#5351, CASE WHEN isnull(total_bal_il#83) THEN 35930.76633081278 WHEN (total_bal_il#83 = NaN) THEN 35930.76633081278 ELSE total_bal_il#83 END AS total_bal_il#4928, CASE WHEN isnull(il_util#84) THEN 70.51851511003233 WHEN (il_util#84 = NaN) THEN 70.51851511003233 ELSE il_util#84 END AS il_util#4929, CASE WHEN isnull(open_rv_12m#85) THEN 0.0 WHEN (open_rv_12m#85 = NaN) THEN 0.0 ELSE open_rv_12m#85 END AS open_rv_12m#5352, CASE WHEN isnull(open_rv_24m#86) THEN 1.0 WHEN (open_rv_24m#86 = NaN) THEN 1.0 ELSE open_rv_24m#86 END AS open_rv_24m#5353, CASE WHEN isnull(max_bal_bc#87) THEN 5599.081456114653 WHEN (max_bal_bc#87 = NaN) THEN 5599.081456114653 ELSE max_bal_bc#87 END AS max_bal_bc#4930, CASE WHEN isnull(all_util#88) THEN 57.361190572817165 WHEN (all_util#88 = NaN) THEN 57.361190572817165 ELSE all_util#88 END AS all_util#4931, total_rev_hi_lim#89, CASE WHEN isnull(inq_fi#90) THEN 0.0 WHEN (inq_fi#90 = NaN) THEN 0.0 ELSE inq_fi#90 END AS inq_fi#5354, CASE WHEN isnull(total_cu_tl#91) THEN 0.0 WHEN (total_cu_tl#91 = NaN) THEN 0.0 ELSE total_cu_tl#91 END AS total_cu_tl#5355, CASE WHEN isnull(inq_last_12m#92) THEN 0.0 WHEN (inq_last_12m#92 = NaN) THEN 0.0 ELSE inq_last_12m#92 END AS inq_last_12m#5356, acc_open_past_24mths#93, CASE WHEN isnull(avg_cur_bal#94) THEN 13604.555559335873 WHEN (avg_cur_bal#94 = NaN) THEN 13604.555559335873 ELSE avg_cur_bal#94 END AS avg_cur_bal#4932, CASE WHEN isnull(bc_open_to_buy#95) THEN 11110.26995201541 WHEN (bc_open_to_buy#95 = NaN) THEN 11110.26995201541 ELSE bc_open_to_buy#95 END AS bc_open_to_buy#4933, CASE WHEN isnull(bc_util#96) THEN 58.035561853186145 WHEN (bc_util#96 = NaN) THEN 58.035561853186145 ELSE bc_util#96 END AS bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, CASE WHEN isnull(num_rev_accts#115) THEN 10.0 WHEN (num_rev_accts#115 = NaN) THEN 10.0 ELSE num_rev_accts#115 END AS num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, CASE WHEN isnull(num_tl_120dpd_2m#118) THEN 0.0 WHEN (num_tl_120dpd_2m#118 = NaN) THEN 0.0 ELSE num_tl_120dpd_2m#118 END AS num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, CASE WHEN isnull(percent_tl_nvr_deliq#1009) THEN 94.13455522916128 WHEN (percent_tl_nvr_deliq#1009 = NaN) THEN 94.13455522916128 ELSE percent_tl_nvr_deliq#1009 END AS percent_tl_nvr_deliq#4935, CASE WHEN isnull(percent_bc_gt_75#123) THEN 42.789030996415775 WHEN (percent_bc_gt_75#123 = NaN) THEN 42.789030996415775 ELSE percent_bc_gt_75#123 END AS percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, CASE WHEN isnull(hardship_flag#142) THEN N ELSE hardship_flag#142 END AS hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, (loan_amnt#18 / (annual_inc#29 + 1.0)) AS loan_amnt_to_annual_inc#7041, (delinq_amnt#98 / (annual_inc#29 + 1.0)) AS delinq_amnt_to_annual_inc#7185, (revol_bal#49 / (annual_inc#29 + 1.0)) AS revol_bal_to_annual_inc#7330, (tot_cur_bal#77 / (annual_inc#29 + 1.0)) AS tot_cur_bal_to_annual_inc#7476]\n",
      "Input [95]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#26, emp_length#27, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#35, title#36, zip_code#37, addr_state#38, dti#39, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#44, open_acc#47, pub_rec#48, revol_bal#49, revol_util#2863, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#78, open_act_il#79, open_il_12m#80, open_il_24m#81, total_bal_il#83, il_util#84, open_rv_12m#85, open_rv_24m#86, max_bal_bc#87, all_util#88, total_rev_hi_lim#89, inq_fi#90, total_cu_tl#91, inq_last_12m#92, acc_open_past_24mths#93, avg_cur_bal#94, bc_open_to_buy#95, bc_util#96, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#115, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#118, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#1009, percent_bc_gt_75#123, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#142, deferral_term#4581, debt_settlement_flag#157]\n",
      "\n",
      "(5) Project\n",
      "Output [105]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, (max_bal_bc#4930 / (annual_inc#29 + 1.0)) AS max_bal_bc_to_annual_inc#7623, (avg_cur_bal#4932 / (annual_inc#29 + 1.0)) AS avg_cur_bal_to_annual_inc#7771, (total_bal_ex_mort#127 / (annual_inc#29 + 1.0)) AS total_bal_ex_mort_to_annual_inc#7920, (open_acc#47 / total_acc#51) AS open_cl_ratio#8070, cast(CASE WHEN isnull(last_pymnt_d#1724) THEN CASE WHEN (total_pymnt#55 > 0.0) THEN (total_pymnt#55 / installment#23) ELSE 0.0 END ELSE cast(((((year(last_pymnt_d#1724) - year(issue_d#1440)) * 12) + month(last_pymnt_d#1724)) - month(issue_d#1440)) as double) END as int) AS num_mths_to_pay#8373, cast(regexp_extract(term#21, (\\d\\d), 1) as int) AS term_int#8525]\n",
      "Input [99]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476]\n",
      "\n",
      "(6) Project\n",
      "Output [107]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#8373, term_int#8525, (((funded_amnt#19 * ((cast(int_rate#2718 as double) / 100.0) / 12.0)) * POWER((((cast(int_rate#2718 as double) / 100.0) / 12.0) + 1.0), cast(term_int#8525 as double))) / (POWER((((cast(int_rate#2718 as double) / 100.0) / 12.0) + 1.0), cast(term_int#8525 as double)) - 1.0)) AS calculated_installment#8678, (installment#23 / (annual_inc#29 + 1.0)) AS installment_to_annual_inc#8832]\n",
      "Input [105]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#8373, term_int#8525]\n",
      "\n",
      "(7) BatchEvalPython\n",
      "Input [107]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#8373, term_int#8525, calculated_installment#8678, installment_to_annual_inc#8832]\n",
      "Arguments: [calculate_cdf(num_mths_to_pay#8373)#9300], [pythonUDF0#12484]\n",
      "\n",
      "(8) Project\n",
      "Output [110]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#8373, term_int#8525, calculated_installment#8678, installment_to_annual_inc#8832, (calculated_installment#8678 / (annual_inc#29 + 1.0)) AS calculated_installment_to_annual_inc#8987, CASE WHEN (num_mths_to_pay#8373 = 0) THEN last_pymnt_amnt#63 ELSE (total_pymnt#55 / cast(num_mths_to_pay#8373 as double)) END AS avg_pymnt#9143, pythonUDF0#12484 AS cdf#9301]\n",
      "Input [108]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#8373, term_int#8525, calculated_installment#8678, installment_to_annual_inc#8832, pythonUDF0#12484]\n",
      "\n",
      "(9) Project\n",
      "Output [109]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, CASE WHEN (RLIKE(loan_status#32, (Fully)) AND (issue_d#1440 = last_pymnt_d#1724)) THEN 1 ELSE num_mths_to_pay#8373 END AS num_mths_to_pay#10107, term_int#8525, calculated_installment#8678, installment_to_annual_inc#8832, calculated_installment_to_annual_inc#8987, avg_pymnt#9143, (((CASE WHEN ((avg_pymnt#9143 > 0.0) AND (num_mths_to_pay#8373 = 0)) THEN avg_pymnt#9143 ELSE (cast(cdf#9301 as double) * avg_pymnt#9143) END + (((recoveries#60 - collection_recovery_fee#61) * 1.0) / POWER(1.0016666666666667, cast((num_mths_to_pay#8373 + 6) as double)))) - funded_amnt#19) / funded_amnt#19) AS ROI#9936]\n",
      "Input [110]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_d#1724, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#8373, term_int#8525, calculated_installment#8678, installment_to_annual_inc#8832, calculated_installment_to_annual_inc#8987, avg_pymnt#9143, cdf#9301]\n",
      "\n",
      "(10) Scan ExistingRDD [codegen id : 2]\n",
      "Output [1]: [IRR#10582]\n",
      "Arguments: [IRR#10582], MapPartitionsRDD[118] at applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD, UnknownPartitioning(0)\n",
      "\n",
      "(11) CartesianProduct\n",
      "Join type: Inner\n",
      "Join condition: None\n",
      "\n",
      "(12) BatchEvalPython\n",
      "Input [110]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#10107, term_int#8525, calculated_installment#8678, installment_to_annual_inc#8832, calculated_installment_to_annual_inc#8987, avg_pymnt#9143, ROI#9936, IRR#10582]\n",
      "Arguments: [binning_int_rate(int_rate#2718)#11720, binning_income(annual_inc#29)#12159], [pythonUDF0#12485, pythonUDF1#12486]\n",
      "\n",
      "(13) Project\n",
      "Output [112]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#10107, term_int#8525, calculated_installment#8678, installment_to_annual_inc#8832, calculated_installment_to_annual_inc#8987, avg_pymnt#9143, ROI#9936, IRR#10582, pythonUDF0#12485 AS int_rate_bin#11721, pythonUDF1#12486 AS annual_inc_bin#12160]\n",
      "Input [112]: [id#17, loan_amnt#18, funded_amnt#19, funded_amnt_inv#20, term#21, int_rate#2718, installment#23, grade#24, sub_grade#25, emp_title#6612, emp_length#5823, home_ownership#28, annual_inc#29, verification_status#30, issue_d#1440, loan_status#32, purpose#6898, title#6755, zip_code#6146, addr_state#38, dti#4926, delinq_2yrs#40, fico_range_low#42, fico_range_high#43, inq_last_6mths#5347, open_acc#47, pub_rec#48, revol_bal#49, revol_util#4927, total_acc#51, initial_list_status#52, out_prncp#53, out_prncp_inv#54, total_pymnt#55, total_pymnt_inv#56, total_rec_prncp#57, total_rec_int#58, total_rec_late_fee#59, recoveries#60, collection_recovery_fee#61, last_pymnt_amnt#63, last_fico_range_high#66, last_fico_range_low#67, collections_12_mths_ex_med#68, application_type#71, acc_now_delinq#75, tot_coll_amt#76, tot_cur_bal#77, open_acc_6m#5348, open_act_il#5349, open_il_12m#5350, open_il_24m#5351, total_bal_il#4928, il_util#4929, open_rv_12m#5352, open_rv_24m#5353, max_bal_bc#4930, all_util#4931, total_rev_hi_lim#89, inq_fi#5354, total_cu_tl#5355, inq_last_12m#5356, acc_open_past_24mths#93, avg_cur_bal#4932, bc_open_to_buy#4933, bc_util#4934, chargeoff_within_12_mths#97, delinq_amnt#98, mort_acc#103, num_accts_ever_120_pd#108, num_actv_bc_tl#109, num_actv_rev_tl#110, num_bc_sats#111, num_bc_tl#112, num_il_tl#113, num_op_rev_tl#114, num_rev_accts#5357, num_rev_tl_bal_gt_0#116, num_sats#117, num_tl_120dpd_2m#5358, num_tl_30dpd#119, num_tl_90g_dpd_24m#120, num_tl_op_past_12m#121, percent_tl_nvr_deliq#4935, percent_bc_gt_75#4936, pub_rec_bankruptcies#124, tax_liens#125, tot_hi_cred_lim#126, total_bal_ex_mort#127, total_bc_limit#128, total_il_high_credit_limit#129, hardship_flag#6469, deferral_term#4581, debt_settlement_flag#157, loan_amnt_to_annual_inc#7041, delinq_amnt_to_annual_inc#7185, revol_bal_to_annual_inc#7330, tot_cur_bal_to_annual_inc#7476, max_bal_bc_to_annual_inc#7623, avg_cur_bal_to_annual_inc#7771, total_bal_ex_mort_to_annual_inc#7920, open_cl_ratio#8070, num_mths_to_pay#10107, term_int#8525, calculated_installment#8678, installment_to_annual_inc#8832, calculated_installment_to_annual_inc#8987, avg_pymnt#9143, ROI#9936, IRR#10582, pythonUDF0#12485, pythonUDF1#12486]\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.explain(mode=\"formatted\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "id": "0429c271",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "ERROR:root:KeyboardInterrupt while sending command.\n",
      "Traceback (most recent call last):\n",
      "  File \"C:\\Users\\User\\anaconda3\\Lib\\site-packages\\py4j\\java_gateway.py\", line 1038, in send_command\n",
      "    response = connection.send_command(command)\n",
      "               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n",
      "  File \"C:\\Users\\User\\anaconda3\\Lib\\site-packages\\py4j\\clientserver.py\", line 511, in send_command\n",
      "    answer = smart_decode(self.stream.readline()[:-1])\n",
      "                          ^^^^^^^^^^^^^^^^^^^^^^\n",
      "  File \"C:\\Users\\User\\anaconda3\\Lib\\socket.py\", line 706, in readinto\n",
      "    return self._sock.recv_into(b)\n",
      "           ^^^^^^^^^^^^^^^^^^^^^^^\n",
      "KeyboardInterrupt\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[82], line 10\u001b[0m\n\u001b[0;32m      2\u001b[0m csv_path \u001b[38;5;241m=\u001b[39m \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mdata\u001b[39m\u001b[38;5;124m\"\u001b[39m\n\u001b[0;32m      4\u001b[0m \u001b[38;5;66;03m# Write DataFrame to CSV\u001b[39;00m\n\u001b[0;32m      5\u001b[0m df\u001b[38;5;241m.\u001b[39mlimit(\u001b[38;5;241m10\u001b[39m) \\\n\u001b[0;32m      6\u001b[0m     \u001b[38;5;241m.\u001b[39mwrite \\\n\u001b[0;32m      7\u001b[0m     \u001b[38;5;241m.\u001b[39mformat(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mcsv\u001b[39m\u001b[38;5;124m\"\u001b[39m) \\\n\u001b[0;32m      8\u001b[0m     \u001b[38;5;241m.\u001b[39moption(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mheader\u001b[39m\u001b[38;5;124m\"\u001b[39m, \u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mtrue\u001b[39m\u001b[38;5;124m\"\u001b[39m) \\\n\u001b[0;32m      9\u001b[0m     \u001b[38;5;241m.\u001b[39mmode(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124moverwrite\u001b[39m\u001b[38;5;124m\"\u001b[39m) \\\n\u001b[1;32m---> 10\u001b[0m     \u001b[38;5;241m.\u001b[39mcsv(csv_path)\n",
      "File \u001b[1;32m~\\anaconda3\\Lib\\site-packages\\pyspark\\sql\\readwriter.py:1864\u001b[0m, in \u001b[0;36mDataFrameWriter.csv\u001b[1;34m(self, path, mode, compression, sep, quote, escape, header, nullValue, escapeQuotes, quoteAll, dateFormat, timestampFormat, ignoreLeadingWhiteSpace, ignoreTrailingWhiteSpace, charToEscapeQuoteEscaping, encoding, emptyValue, lineSep)\u001b[0m\n\u001b[0;32m   1845\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mmode(mode)\n\u001b[0;32m   1846\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_set_opts(\n\u001b[0;32m   1847\u001b[0m     compression\u001b[38;5;241m=\u001b[39mcompression,\n\u001b[0;32m   1848\u001b[0m     sep\u001b[38;5;241m=\u001b[39msep,\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m   1862\u001b[0m     lineSep\u001b[38;5;241m=\u001b[39mlineSep,\n\u001b[0;32m   1863\u001b[0m )\n\u001b[1;32m-> 1864\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_jwrite\u001b[38;5;241m.\u001b[39mcsv(path)\n",
      "File \u001b[1;32m~\\anaconda3\\Lib\\site-packages\\py4j\\java_gateway.py:1321\u001b[0m, in \u001b[0;36mJavaMember.__call__\u001b[1;34m(self, *args)\u001b[0m\n\u001b[0;32m   1314\u001b[0m args_command, temp_args \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_build_args(\u001b[38;5;241m*\u001b[39margs)\n\u001b[0;32m   1316\u001b[0m command \u001b[38;5;241m=\u001b[39m proto\u001b[38;5;241m.\u001b[39mCALL_COMMAND_NAME \u001b[38;5;241m+\u001b[39m\\\n\u001b[0;32m   1317\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mcommand_header \u001b[38;5;241m+\u001b[39m\\\n\u001b[0;32m   1318\u001b[0m     args_command \u001b[38;5;241m+\u001b[39m\\\n\u001b[0;32m   1319\u001b[0m     proto\u001b[38;5;241m.\u001b[39mEND_COMMAND_PART\n\u001b[1;32m-> 1321\u001b[0m answer \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mgateway_client\u001b[38;5;241m.\u001b[39msend_command(command)\n\u001b[0;32m   1322\u001b[0m return_value \u001b[38;5;241m=\u001b[39m get_return_value(\n\u001b[0;32m   1323\u001b[0m     answer, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mgateway_client, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mtarget_id, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mname)\n\u001b[0;32m   1325\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m temp_arg \u001b[38;5;129;01min\u001b[39;00m temp_args:\n",
      "File \u001b[1;32m~\\anaconda3\\Lib\\site-packages\\py4j\\java_gateway.py:1038\u001b[0m, in \u001b[0;36mGatewayClient.send_command\u001b[1;34m(self, command, retry, binary)\u001b[0m\n\u001b[0;32m   1036\u001b[0m connection \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_get_connection()\n\u001b[0;32m   1037\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m-> 1038\u001b[0m     response \u001b[38;5;241m=\u001b[39m connection\u001b[38;5;241m.\u001b[39msend_command(command)\n\u001b[0;32m   1039\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m binary:\n\u001b[0;32m   1040\u001b[0m         \u001b[38;5;28;01mreturn\u001b[39;00m response, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_create_connection_guard(connection)\n",
      "File \u001b[1;32m~\\anaconda3\\Lib\\site-packages\\py4j\\clientserver.py:511\u001b[0m, in \u001b[0;36mClientServerConnection.send_command\u001b[1;34m(self, command)\u001b[0m\n\u001b[0;32m    509\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m    510\u001b[0m     \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28;01mTrue\u001b[39;00m:\n\u001b[1;32m--> 511\u001b[0m         answer \u001b[38;5;241m=\u001b[39m smart_decode(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mstream\u001b[38;5;241m.\u001b[39mreadline()[:\u001b[38;5;241m-\u001b[39m\u001b[38;5;241m1\u001b[39m])\n\u001b[0;32m    512\u001b[0m         logger\u001b[38;5;241m.\u001b[39mdebug(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mAnswer received: \u001b[39m\u001b[38;5;132;01m{0}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;241m.\u001b[39mformat(answer))\n\u001b[0;32m    513\u001b[0m         \u001b[38;5;66;03m# Happens when a the other end is dead. There might be an empty\u001b[39;00m\n\u001b[0;32m    514\u001b[0m         \u001b[38;5;66;03m# answer before the socket raises an error.\u001b[39;00m\n",
      "File \u001b[1;32m~\\anaconda3\\Lib\\socket.py:706\u001b[0m, in \u001b[0;36mSocketIO.readinto\u001b[1;34m(self, b)\u001b[0m\n\u001b[0;32m    704\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28;01mTrue\u001b[39;00m:\n\u001b[0;32m    705\u001b[0m     \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m--> 706\u001b[0m         \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_sock\u001b[38;5;241m.\u001b[39mrecv_into(b)\n\u001b[0;32m    707\u001b[0m     \u001b[38;5;28;01mexcept\u001b[39;00m timeout:\n\u001b[0;32m    708\u001b[0m         \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_timeout_occurred \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mTrue\u001b[39;00m\n",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "# Specify the path where you want to save the CSV file\n",
    "csv_path = \"data\"\n",
    "\n",
    "# Write DataFrame to CSV\n",
    "df.limit(10) \\\n",
    "    .write \\\n",
    "    .format(\"csv\") \\\n",
    "    .option(\"header\", \"true\") \\\n",
    "    .mode(\"overwrite\") \\\n",
    "    .csv(csv_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ef6bd120",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}