churn/4. Feature Engineering on Spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction: Feature Engineering with Spark\n",
    "\n",
    "Problem: In `Feature Engineering`, we developed a pipeline for automated feature engineering using a dataset of customer transactions and label times. Running this pipeline on a single partition of customers takes about 15 minutes which means computing all of the features would require several days if done one at a time. \n",
    "\n",
    "Solution: Break the dataset into independent partitions of customers and run multiple subsets in parallel. This can be done using multiple processors on a single machine or a cluster of machines.\n",
    "\n",
    "## Spark with PySpark\n",
    "\n",
    "[Apache Spark](http://spark.apache.org) is a popular framework for distributed computed and large-data processing. It allows us to run computations in parallel either on a single machine, or distributed across a cluster of machines. In this notebook, we will run automated feature engineering in [Featuretools](https://github.com/Featuretools/featuretools) using Spark with the [PySpark library](http://spark.apache.org/docs/2.2.0/api/python/pyspark.html). \n",
    "\n",
    "The first step is initializing Spark. We can use the `findspark` library to make sure that `pyspark` can find Spark in the Jupyter Notebook. This notebook assumes the Spark cluster is already running. To get started with a Spark cluster, refer to [this guide](https://data-flair.training/blogs/install-apache-spark-multi-node-cluster/). \n",
    "\n",
    "(We'll skip the Featuretools details in this notebook, but for an introduction see [this article](https://towardsdatascience.com/automated-feature-engineering-in-python-99baf11cc219). For a comparison of manual to automated feature engineering, see [this article](https://towardsdatascience.com/why-automated-feature-engineering-will-change-the-way-you-do-machine-learning-5c15bf188b96). )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "# Initialize with Spark file location\n",
    "findspark.init('/usr/local/spark')\n",
    "\n",
    "import pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Set up Spark \n",
    "\n",
    "A `SparkContext` is the interface to a running Spark cluster. We pass in a number of parameters to the `SparkContext` using a `SparkConf` object. Namely, we'll turn on logging, tell Spark to use all cores on our 3 machines, and direct Spark to the location of the master (parent) node. \n",
    "\n",
    "Adjust the parameters depending on your cluster set up. I found [this guide](https://spoddutur.github.io/spark-notes/distribution_of_executors_cores_and_memory_for_spark_application.html) to be helpful in choosing the parameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "dict_items([('spark.eventLog.enabled', 'True'), ('spark.eventLog.dir', 'tmp/'), ('spark.num.executors', '3'), ('spark.executor.memory', '12g'), ('spark.executor.cores', '4'), ('spark.master', 'spark://ip-172-31-8-174.ec2.internal:7077')])"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "conf = pyspark.SparkConf()\n",
    "\n",
    "# Enable logging\n",
    "conf.set('spark.eventLog.enabled', True);\n",
    "conf.set('spark.eventLog.dir', 'tmp/');\n",
    "\n",
    "# Use all cores on all machines\n",
    "conf.set('spark.num.executors', 3)\n",
    "conf.set('spark.executor.memory', '12g')\n",
    "conf.set('spark.executor.cores', 4)\n",
    "\n",
    "# Set the parent\n",
    "conf.set('spark.master', 'spark://ip-172-31-8-174.ec2.internal:7077')\n",
    "conf.getAll()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Testing Spark \n",
    "\n",
    "Before we get to the feature engineering, we want to test if our cluster is running correctly. We'll instantiate a `Spark` cluster and run a simple program that calculates the value of pi. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://ip-172-31-23-133.ec2.internal:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.3.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>spark://ip-172-31-23-133.ec2.internal:7077</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>pi_calc</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=spark://ip-172-31-23-133.ec2.internal:7077 appName=pi_calc>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc = pyspark.SparkContext(appName=\"pi_calc\", \n",
    "                          conf = conf)\n",
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3.14152968\n"
     ]
    }
   ],
   "source": [
    "num_samples = 100000000\n",
    "import random\n",
    "\n",
    "def inside(p):     \n",
    "  x, y = random.random(), random.random()\n",
    "  return x*x + y*y < 1\n",
    "\n",
    "# Parallelize counting samples inside circle using Spark\n",
    "count = sc.parallelize(range(0, num_samples)).filter(inside).count()\n",
    "pi = 4 * count / num_samples\n",
    "print(pi)\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark Dashboards\n",
    "\n",
    "After starting the Spark cluster  from the command line- before running any of the code in the notebook - you can view a dashboard of the cluster at localhost:8080. This shows basic information such as the number of workers and the currently running or completed jobs.\n",
    "\n",
    "\n",
    "![](../images/spark_cluster_main.png)\n",
    "\n",
    "Once a `SparkContext` has been initialized, the job can be viewed at localhost:4040. This shows particular details such as the number of tasks completed and the directed acyclic graph of the operation. \n",
    "\n",
    "![](../images/stages.png)\n",
    "\n",
    "Using the web dashboard can be a helpful method to help debug your cluster. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the cluster is running correctly, we can move on to feature engineering. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Storage\n",
    "\n",
    "All of the reading and writing for running with Spark will happen through S3. The partitioned files are all on s3 and we can use `pandas.read_csv` to read directly from s3. To write to s3, we use the `s3fs` library (shown a little later). \n",
    "\n",
    "### Read in Data from S3\n",
    "\n",
    "Before running this code, make sure to authenticate with Amazon Web Services from the command line to access your files in S3. Run `aws configure` and then input the appropriate information. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import featuretools as ft\n",
    "import featuretools.variable_types as vtypes\n",
    "\n",
    "partition = 20\n",
    "directory = 's3://customer-churn-spark/p' + str(partition)\n",
    "cutoff_times_file = 'MS-31_labels.csv'\n",
    "\n",
    "\n",
    "# Read in the data files\n",
    "members = pd.read_csv(f'{directory}/members.csv', \n",
    "                  parse_dates=['registration_init_time'], \n",
    "                  infer_datetime_format = True, \n",
    "                  dtype = {'gender': 'category'})\n",
    "\n",
    "trans = pd.read_csv(f'{directory}/transactions.csv',\n",
    "                   parse_dates=['transaction_date', 'membership_expire_date'], \n",
    "                    infer_datetime_format = True)\n",
    "\n",
    "logs = pd.read_csv(f'{directory}/logs.csv', parse_dates = ['date'])\n",
    "\n",
    "cutoff_times = pd.read_csv(f'{directory}/{cutoff_times_file}', parse_dates = ['cutoff_time'])\n",
    "cutoff_times = cutoff_times.drop_duplicates(subset = ['msno', 'cutoff_time'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Feature Engineering\n",
    "\n",
    "First we'll make the set of features using a single partiton so we don't have to recalculate them for each partition. This also ensures that the same exact features are made for each subset of customers. (It also is possible to load in calculated features from disk.) Again, I'm skipping the explanation for what is going on here so check out the [Featuretools documentation](https://docs.featuretools.com/) or some of the [online tutorials](https://www.featuretools.com/demos). \n",
    "\n",
    "### Features for One Partition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Entityset: customers\n",
       "  Entities:\n",
       "    members [Rows: 6817, Columns: 6]\n",
       "    transactions [Rows: 23423, Columns: 13]\n",
       "    logs [Rows: 418190, Columns: 13]\n",
       "  Relationships:\n",
       "    transactions.msno -> members.msno\n",
       "    logs.msno -> members.msno"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Create empty entityset\n",
    "es = ft.EntitySet(id = 'customers')\n",
    "\n",
    "# Add the members parent table\n",
    "es.entity_from_dataframe(entity_id='members', dataframe=members,\n",
    "                         index = 'msno', time_index = 'registration_init_time', \n",
    "                         variable_types = {'city': vtypes.Categorical, 'bd': vtypes.Categorical,\n",
    "                                           'registered_via': vtypes.Categorical})\n",
    "# Create new features in transactions\n",
    "trans['price_difference'] = trans['plan_list_price'] - trans['actual_amount_paid']\n",
    "trans['planned_daily_price'] = trans['plan_list_price'] / trans['payment_plan_days']\n",
    "trans['daily_price'] = trans['actual_amount_paid'] / trans['payment_plan_days']\n",
    "\n",
    "# Add the transactions child table\n",
    "es.entity_from_dataframe(entity_id='transactions', dataframe=trans,\n",
    "                         index = 'transactions_index', make_index = True,\n",
    "                         time_index = 'transaction_date', \n",
    "                         variable_types = {'payment_method_id': vtypes.Categorical, \n",
    "                                           'is_auto_renew': vtypes.Boolean, 'is_cancel': vtypes.Boolean})\n",
    "\n",
    "# Add transactions interesting values\n",
    "es['transactions']['is_cancel'].interesting_values = [0, 1]\n",
    "es['transactions']['is_auto_renew'].interesting_values = [0, 1]\n",
    "\n",
    "# Create new features in logs\n",
    "logs['total'] = logs[['num_25', 'num_50', 'num_75', 'num_985', 'num_100']].sum(axis = 1)\n",
    "logs['percent_100'] = logs['num_100'] / logs['total']\n",
    "logs['percent_unique'] = logs['num_unq'] / logs['total']\n",
    "\n",
    "# Add the logs child table\n",
    "es.entity_from_dataframe(entity_id='logs', dataframe=logs,\n",
    "                     index = 'logs_index', make_index = True,\n",
    "                     time_index = 'date')\n",
    "\n",
    "# Add the relationships\n",
    "r_member_transactions = ft.Relationship(es['members']['msno'], es['transactions']['msno'])\n",
    "r_member_logs = ft.Relationship(es['members']['msno'], es['logs']['msno'])\n",
    "es.add_relationships([r_member_transactions, r_member_logs])\n",
    "\n",
    "es"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Custom Primitives\n",
    "\n",
    "Below is a custom primitive we wrote (see the `Feature Engineering` notebook) for this dataset. It calculates the total amount of a quantity in the previous month."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def total_previous_month(numeric, datetime, time):\n",
    "    \"\"\"Return total of `numeric` column in the month prior to `time`.\"\"\"\n",
    "    df = pd.DataFrame({'value': numeric, 'date': datetime})\n",
    "    previous_month = time.month - 1\n",
    "    year = time.year\n",
    "   \n",
    "    # Handle January\n",
    "    if previous_month == 0:\n",
    "        previous_month = 12\n",
    "        year = time.year - 1\n",
    "        \n",
    "    # Filter data and sum up total\n",
    "    df = df[(df['date'].dt.month == previous_month) & (df['date'].dt.year == year)]\n",
    "    total = df['value'].sum()\n",
    "    \n",
    "    return total"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "from featuretools.primitives import make_agg_primitive\n",
    "\n",
    "# Takes in a number and outputs a number\n",
    "total_previous = make_agg_primitive(total_previous_month, input_types = [ft.variable_types.Numeric,\n",
    "                                                                         ft.variable_types.Datetime],\n",
    "                                    return_type = ft.variable_types.Numeric, \n",
    "                                    uses_calc_time = True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Run Deep Feature Synthesis"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The first time we create the features, we use `ft.dfs` passing in the selected primitives, the target entity, the critical `cutoff_time`, the depth of the feature to stack, and several other parameters. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Specify aggregation primitives\n",
    "agg_primitives = ['sum', 'time_since_last', 'avg_time_between', 'all', 'mode', 'num_unique', 'min', 'last', \n",
    "                  'mean', 'percent_true', 'max', 'std', 'count', total_previous]\n",
    "# Specify transformation primitives\n",
    "trans_primitives = ['weekend', 'cum_sum', 'day', 'month', 'diff', 'time_since_previous']\n",
    "\n",
    "# Specify where primitives\n",
    "where_primitives = ['sum', 'mean', 'percent_true', 'all', 'any']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Built 248 features\n",
      "Elapsed: 17:17 | Remaining: 00:00 | Progress: 100%|██████████| Calculated: 274/274 chunks\n"
     ]
    }
   ],
   "source": [
    "# Run deep feature synthesis\n",
    "feature_matrix, feature_defs = ft.dfs(entityset=es, target_entity='members', \n",
    "                                      cutoff_time = cutoff_times, \n",
    "                                      agg_primitives = agg_primitives,\n",
    "                                      trans_primitives = trans_primitives,\n",
    "                                      where_primitives = where_primitives,\n",
    "                                      max_depth = 2, features_only = False,\n",
    "                                      chunk_size = 100, n_jobs = 1, verbose = 1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These features definitions can be saved on disk. Every time we want to make the same exact features, we can just pass in these into the `ft.calculate_feature_matrix` function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "ft.save_features(feature_defs, '/data/churn/features.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 248 features.\n"
     ]
    }
   ],
   "source": [
    "feature_defs = ft.load_features('/data/churn/features.txt')\n",
    "print(f'There are {len(feature_defs)} features.')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Writing Feature Matrix to S3 \n",
    "\n",
    "In order to save each feature matrix from a partition to the cloud, we'll write it directly to s3. For this we can use the `s3fs` (s3 file system) Python library. We first have to authenticate with aws by loading in the credentials and then we can upload our csv much the same as we would write any csv. We use the [`s3fs` library](https://s3fs.readthedocs.io/). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "import s3fs\n",
    "\n",
    "# Credentials\n",
    "with open('/data/credentials.txt', 'r') as f:\n",
    "    info = f.read().strip().split(',')\n",
    "    key = info[0]\n",
    "    secret = info[1]\n",
    "\n",
    "fs = s3fs.S3FileSystem(key=key, secret=secret)\n",
    "\n",
    "# S3 directory\n",
    "directory = 's3://customer-churn-spark/p' + str(partition)\n",
    "\n",
    "# Encode in order to write to s3\n",
    "bytes_to_write = feature_matrix.to_csv(None).encode()\n",
    "\n",
    "# Write to s3\n",
    "with fs.open(f'{directory}/feature_matrix.csv', 'wb') as f:\n",
    "    f.write(bytes_to_write)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Partition to Feature Matrix Function\n",
    "\n",
    "The main function of this notebook is used to make features from a single partition. \n",
    "\n",
    "This function, `partition_to_feature_matrix`, does the following:\n",
    "\n",
    "1. Takes in the name of a partition \n",
    "2. Reads the data from s3\n",
    "3. Creates an entityset from the data\n",
    "4. Computes the feature matrix for the partition\n",
    "5. Saves the feature matrix to s3\n",
    "\n",
    "Because all reading and writing happens through S3, we don't have to worry about disc space or about putting a copy of the data on each machine. Instead, we can simply read from and write to the cloud."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "N_PARTITIONS = 1000\n",
    "BASE_DIR = 's3://customer-churn-spark/'\n",
    "    \n",
    "def partition_to_feature_matrix(partition, feature_defs = feature_defs, \n",
    "                                cutoff_time_name = 'MS-31_labels.csv', write = True):\n",
    "    \"\"\"Take in a partition number, create a feature matrix, and save to Amazon S3\n",
    "    \n",
    "    Params\n",
    "    --------\n",
    "        partition (int): number of partition\n",
    "        feature_defs (list of ft features): features to make for the partition\n",
    "        cutoff_time_name (str): name of cutoff time file\n",
    "        write: (boolean): whether to write the data to S3. Defaults to True\n",
    "        \n",
    "    Return\n",
    "    --------\n",
    "        None: saves the feature matrix to Amazon S3\n",
    "    \n",
    "    \"\"\"\n",
    "    \n",
    "    partition_dir = BASE_DIR + 'p' + str(partition)\n",
    "    \n",
    "    # Read in the data files\n",
    "    members = pd.read_csv(f'{partition_dir}/members.csv', \n",
    "                      parse_dates=['registration_init_time'], \n",
    "                      infer_datetime_format = True, \n",
    "                      dtype = {'gender': 'category'})\n",
    "\n",
    "    trans = pd.read_csv(f'{partition_dir}/transactions.csv',\n",
    "                       parse_dates=['transaction_date', 'membership_expire_date'], \n",
    "                        infer_datetime_format = True)\n",
    "    logs = pd.read_csv(f'{partition_dir}/logs.csv', parse_dates = ['date'])\n",
    "    \n",
    "    # Make sure to drop duplicates\n",
    "    cutoff_times = pd.read_csv(f'{partition_dir}/{cutoff_time_name}', parse_dates = ['cutoff_time'])\n",
    "    cutoff_times = cutoff_times.drop_duplicates(subset = ['msno', 'cutoff_time'])\n",
    "    \n",
    "    # Needed for saving\n",
    "    cutoff_spec = cutoff_time_name.split('_')[0]\n",
    "    \n",
    "    # Create empty entityset\n",
    "    es = ft.EntitySet(id = 'customers')\n",
    "\n",
    "    # Add the members parent table\n",
    "    es.entity_from_dataframe(entity_id='members', dataframe=members,\n",
    "                             index = 'msno', time_index = 'registration_init_time', \n",
    "                             variable_types = {'city': vtypes.Categorical,\n",
    "                                               'registered_via': vtypes.Categorical})\n",
    "    # Create new features in transactions\n",
    "    trans['price_difference'] = trans['plan_list_price'] - trans['actual_amount_paid']\n",
    "    trans['planned_daily_price'] = trans['plan_list_price'] / trans['payment_plan_days']\n",
    "    trans['daily_price'] = trans['actual_amount_paid'] / trans['payment_plan_days']\n",
    "\n",
    "    # Add the transactions child table\n",
    "    es.entity_from_dataframe(entity_id='transactions', dataframe=trans,\n",
    "                             index = 'transactions_index', make_index = True,\n",
    "                             time_index = 'transaction_date', \n",
    "                             variable_types = {'payment_method_id': vtypes.Categorical, \n",
    "                                               'is_auto_renew': vtypes.Boolean, 'is_cancel': vtypes.Boolean})\n",
    "\n",
    "    # Add transactions interesting values\n",
    "    es['transactions']['is_cancel'].interesting_values = [0, 1]\n",
    "    es['transactions']['is_auto_renew'].interesting_values = [0, 1]\n",
    "    \n",
    "    # Create new features in logs\n",
    "    logs['total'] = logs[['num_25', 'num_50', 'num_75', 'num_985', 'num_100']].sum(axis = 1)\n",
    "    logs['percent_100'] = logs['num_100'] / logs['total']\n",
    "    logs['percent_unique'] = logs['num_unq'] / logs['total']\n",
    "    logs['seconds_per_song'] = logs['total_secs'] / logs['total'] \n",
    "    \n",
    "    # Add the logs child table\n",
    "    es.entity_from_dataframe(entity_id='logs', dataframe=logs,\n",
    "                         index = 'logs_index', make_index = True,\n",
    "                         time_index = 'date')\n",
    "\n",
    "    # Add the relationships\n",
    "    r_member_transactions = ft.Relationship(es['members']['msno'], es['transactions']['msno'])\n",
    "    r_member_logs = ft.Relationship(es['members']['msno'], es['logs']['msno'])\n",
    "    es.add_relationships([r_member_transactions, r_member_logs])\n",
    "    \n",
    "    # Calculate the feature matrix using pre-calculated features\n",
    "    feature_matrix = ft.calculate_feature_matrix(entityset=es, features=feature_defs, \n",
    "                                                 cutoff_time=cutoff_times, cutoff_time_in_index = True,\n",
    "                                                 chunk_size = 1000)\n",
    "\n",
    "    if write:\n",
    "        # Save to Amazon S3\n",
    "        bytes_to_write = feature_matrix.to_csv(None).encode()\n",
    "\n",
    "        with fs.open(f'{partition_dir}/{cutoff_spec}_feature_matrix.csv', 'wb') as f:\n",
    "            f.write(bytes_to_write)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test Function\n",
    "\n",
    "Let's give the function a test with 2 different partitions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "856 seconds elapsed.\n"
     ]
    }
   ],
   "source": [
    "from timeit import default_timer as timer\n",
    "\n",
    "start = timer()\n",
    "partition_to_feature_matrix(950, feature_defs, 'MS-31_labels.csv')\n",
    "end = timer()\n",
    "print(f'{round(end - start)} seconds elapsed.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>time</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>gender</th>\n",
       "      <th>SUM(transactions.payment_plan_days)</th>\n",
       "      <th>SUM(transactions.plan_list_price)</th>\n",
       "      <th>SUM(transactions.actual_amount_paid)</th>\n",
       "      <th>SUM(transactions.price_difference)</th>\n",
       "      <th>...</th>\n",
       "      <th>WEEKEND(LAST(logs.date))</th>\n",
       "      <th>DAY(LAST(transactions.transaction_date))</th>\n",
       "      <th>DAY(LAST(transactions.membership_expire_date))</th>\n",
       "      <th>DAY(LAST(logs.date))</th>\n",
       "      <th>MONTH(LAST(transactions.transaction_date))</th>\n",
       "      <th>MONTH(LAST(transactions.membership_expire_date))</th>\n",
       "      <th>MONTH(LAST(logs.date))</th>\n",
       "      <th>label</th>\n",
       "      <th>days_to_churn</th>\n",
       "      <th>churn_date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>+8eyBkJyRyRK08Fu+mpDQ0/JljpCcOdPiWfOCuxqZWQ=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>15.0</td>\n",
       "      <td>32.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>female</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>+BDZoGIRQSQHuzwSu5hiOJ6sVaQkEBixPBI42HhtvX8=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>22.0</td>\n",
       "      <td>29.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>male</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>472.0</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>+IETRNY5pdehTZK7HxJS55bUVmZpLCbkXjYdxplIt+c=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>22.0</td>\n",
       "      <td>32.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>male</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>459.0</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>+Ico+LLCU6UPRQAoS9Q+BDxkU+CyQvr44bjrWY4RJjI=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>7.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>+LO1Iu1Tc0Cz5rjcIo1CgZEqr3poGDFMhPLGA3uvWVo=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>6.0</td>\n",
       "      <td>21.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>female</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>418.0</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 253 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno        time  city    bd  \\\n",
       "0  +8eyBkJyRyRK08Fu+mpDQ0/JljpCcOdPiWfOCuxqZWQ=  2015-01-01  15.0  32.0   \n",
       "1  +BDZoGIRQSQHuzwSu5hiOJ6sVaQkEBixPBI42HhtvX8=  2015-01-01  22.0  29.0   \n",
       "2  +IETRNY5pdehTZK7HxJS55bUVmZpLCbkXjYdxplIt+c=  2015-01-01  22.0  32.0   \n",
       "3  +Ico+LLCU6UPRQAoS9Q+BDxkU+CyQvr44bjrWY4RJjI=  2015-01-01   1.0   0.0   \n",
       "4  +LO1Iu1Tc0Cz5rjcIo1CgZEqr3poGDFMhPLGA3uvWVo=  2015-01-01   6.0  21.0   \n",
       "\n",
       "   registered_via  gender  SUM(transactions.payment_plan_days)  \\\n",
       "0             9.0  female                                  0.0   \n",
       "1             9.0    male                                  0.0   \n",
       "2             9.0    male                                  0.0   \n",
       "3             7.0     NaN                                  0.0   \n",
       "4             9.0  female                                  0.0   \n",
       "\n",
       "   SUM(transactions.plan_list_price)  SUM(transactions.actual_amount_paid)  \\\n",
       "0                                0.0                                   0.0   \n",
       "1                                0.0                                   0.0   \n",
       "2                                0.0                                   0.0   \n",
       "3                                0.0                                   0.0   \n",
       "4                                0.0                                   0.0   \n",
       "\n",
       "   SUM(transactions.price_difference)     ...      WEEKEND(LAST(logs.date))  \\\n",
       "0                                 0.0     ...                           0.0   \n",
       "1                                 0.0     ...                           0.0   \n",
       "2                                 0.0     ...                           0.0   \n",
       "3                                 0.0     ...                           0.0   \n",
       "4                                 0.0     ...                           0.0   \n",
       "\n",
       "   DAY(LAST(transactions.transaction_date))  \\\n",
       "0                                       NaN   \n",
       "1                                       NaN   \n",
       "2                                       NaN   \n",
       "3                                       NaN   \n",
       "4                                       NaN   \n",
       "\n",
       "   DAY(LAST(transactions.membership_expire_date))  DAY(LAST(logs.date))  \\\n",
       "0                                             NaN                   1.0   \n",
       "1                                             NaN                   1.0   \n",
       "2                                             NaN                   1.0   \n",
       "3                                             NaN                   1.0   \n",
       "4                                             NaN                   1.0   \n",
       "\n",
       "  MONTH(LAST(transactions.transaction_date))  \\\n",
       "0                                        NaN   \n",
       "1                                        NaN   \n",
       "2                                        NaN   \n",
       "3                                        NaN   \n",
       "4                                        NaN   \n",
       "\n",
       "  MONTH(LAST(transactions.membership_expire_date))  MONTH(LAST(logs.date))  \\\n",
       "0                                              NaN                     1.0   \n",
       "1                                              NaN                     1.0   \n",
       "2                                              NaN                     1.0   \n",
       "3                                              NaN                     1.0   \n",
       "4                                              NaN                     1.0   \n",
       "\n",
       "   label  days_to_churn  churn_date  \n",
       "0    0.0            NaN         NaN  \n",
       "1    NaN          472.0         NaN  \n",
       "2    NaN          459.0         NaN  \n",
       "3    0.0            NaN         NaN  \n",
       "4    NaN          418.0         NaN  \n",
       "\n",
       "[5 rows x 253 columns]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "feature_matrix = pd.read_csv('s3://customer-churn-spark/p950/MS-31_feature_matrix.csv', low_memory = False)\n",
    "feature_matrix.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "875 seconds elapsed.\n"
     ]
    }
   ],
   "source": [
    "start = timer()\n",
    "partition_to_feature_matrix(530, feature_defs, 'MS-31_labels.csv')\n",
    "end = timer()\n",
    "print(f'{round(end - start)} seconds elapsed.')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>time</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>gender</th>\n",
       "      <th>SUM(transactions.payment_plan_days)</th>\n",
       "      <th>SUM(transactions.plan_list_price)</th>\n",
       "      <th>SUM(transactions.actual_amount_paid)</th>\n",
       "      <th>SUM(transactions.price_difference)</th>\n",
       "      <th>...</th>\n",
       "      <th>WEEKEND(LAST(logs.date))</th>\n",
       "      <th>DAY(LAST(transactions.transaction_date))</th>\n",
       "      <th>DAY(LAST(transactions.membership_expire_date))</th>\n",
       "      <th>DAY(LAST(logs.date))</th>\n",
       "      <th>MONTH(LAST(transactions.transaction_date))</th>\n",
       "      <th>MONTH(LAST(transactions.membership_expire_date))</th>\n",
       "      <th>MONTH(LAST(logs.date))</th>\n",
       "      <th>label</th>\n",
       "      <th>days_to_churn</th>\n",
       "      <th>churn_date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>+2oyBGdHsUwF9UZQAF6JFSlOwohoHPFriNBUQDzj6xw=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>14.0</td>\n",
       "      <td>49.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>male</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>+C8j6Pj/MCr/nAANcuJzta8lCkoZ6oopypdhllkqXlM=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>4.0</td>\n",
       "      <td>27.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>male</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>+E+cBkZzqIXPd4L1vLHYO1xxoD6VF7J5mi1Z/GKA9r0=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>7.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>+O51KSmGMnp+ItBpBgZNBJ94K/e//4fhXGYmxNHvZcg=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>13.0</td>\n",
       "      <td>39.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>female</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>+WiZkfIp5sDsf0xZvBnR2j6Kxi1u2k0t0mJBJqhQIJo=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>13.0</td>\n",
       "      <td>25.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>male</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>472.0</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 253 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno        time  city    bd  \\\n",
       "0  +2oyBGdHsUwF9UZQAF6JFSlOwohoHPFriNBUQDzj6xw=  2015-01-01  14.0  49.0   \n",
       "1  +C8j6Pj/MCr/nAANcuJzta8lCkoZ6oopypdhllkqXlM=  2015-01-01   4.0  27.0   \n",
       "2  +E+cBkZzqIXPd4L1vLHYO1xxoD6VF7J5mi1Z/GKA9r0=  2015-01-01   1.0   0.0   \n",
       "3  +O51KSmGMnp+ItBpBgZNBJ94K/e//4fhXGYmxNHvZcg=  2015-01-01  13.0  39.0   \n",
       "4  +WiZkfIp5sDsf0xZvBnR2j6Kxi1u2k0t0mJBJqhQIJo=  2015-01-01  13.0  25.0   \n",
       "\n",
       "   registered_via  gender  SUM(transactions.payment_plan_days)  \\\n",
       "0             9.0    male                                  0.0   \n",
       "1             9.0    male                                  0.0   \n",
       "2             7.0     NaN                                  0.0   \n",
       "3             9.0  female                                  0.0   \n",
       "4             9.0    male                                  0.0   \n",
       "\n",
       "   SUM(transactions.plan_list_price)  SUM(transactions.actual_amount_paid)  \\\n",
       "0                                0.0                                   0.0   \n",
       "1                                0.0                                   0.0   \n",
       "2                                0.0                                   0.0   \n",
       "3                                0.0                                   0.0   \n",
       "4                                0.0                                   0.0   \n",
       "\n",
       "   SUM(transactions.price_difference)     ...      WEEKEND(LAST(logs.date))  \\\n",
       "0                                 0.0     ...                           0.0   \n",
       "1                                 0.0     ...                           0.0   \n",
       "2                                 0.0     ...                           0.0   \n",
       "3                                 0.0     ...                           0.0   \n",
       "4                                 0.0     ...                           0.0   \n",
       "\n",
       "   DAY(LAST(transactions.transaction_date))  \\\n",
       "0                                       NaN   \n",
       "1                                       NaN   \n",
       "2                                       NaN   \n",
       "3                                       NaN   \n",
       "4                                       NaN   \n",
       "\n",
       "   DAY(LAST(transactions.membership_expire_date))  DAY(LAST(logs.date))  \\\n",
       "0                                             NaN                   NaN   \n",
       "1                                             NaN                   1.0   \n",
       "2                                             NaN                   1.0   \n",
       "3                                             NaN                   NaN   \n",
       "4                                             NaN                   1.0   \n",
       "\n",
       "  MONTH(LAST(transactions.transaction_date))  \\\n",
       "0                                        NaN   \n",
       "1                                        NaN   \n",
       "2                                        NaN   \n",
       "3                                        NaN   \n",
       "4                                        NaN   \n",
       "\n",
       "  MONTH(LAST(transactions.membership_expire_date))  MONTH(LAST(logs.date))  \\\n",
       "0                                              NaN                     NaN   \n",
       "1                                              NaN                     1.0   \n",
       "2                                              NaN                     1.0   \n",
       "3                                              NaN                     NaN   \n",
       "4                                              NaN                     1.0   \n",
       "\n",
       "   label  days_to_churn  churn_date  \n",
       "0    0.0            NaN         NaN  \n",
       "1    0.0            NaN         NaN  \n",
       "2    0.0            NaN         NaN  \n",
       "3    0.0            NaN         NaN  \n",
       "4    NaN          472.0         NaN  \n",
       "\n",
       "[5 rows x 253 columns]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "feature_matrix = pd.read_csv('s3://customer-churn-spark/p530/MS-31_feature_matrix.csv', low_memory = False)\n",
    "feature_matrix.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Run with Spark\n",
    "\n",
    "The next cell parallelizes all the feature engineering calculations using Spark. We want to `map` the partitions to the function and we let Spark divide the work between the executors, each of which is one core on one machine. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create list of partitions\n",
    "partitions = list(range(N_PARTITIONS))\n",
    "\n",
    "# Create Spark context\n",
    "sc = pyspark.SparkContext(master = 'spark://ip-172-31-8-174.ec2.internal:7077',\n",
    "                          appName = 'featuretools', conf = conf)\n",
    "\n",
    "# Parallelize feature engineering\n",
    "r = sc.parallelize(partitions, numSlices=N_PARTITIONS).\\\n",
    "    map(lambda x: partition_to_feature_matrix(x, feature_defs,\n",
    "                                              'MS-31_labels.csv')).collect()\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "While the run is going on, we can look at the status of the cluster at localhost:8080 and the state of the particular job at localhost:4040. \n",
    "\n",
    "__Here is the overall state of the cluster.__\n",
    "\n",
    "![](../images/spark_cluster2.png)\n",
    "\n",
    "__Here is information about the submitted job.__\n",
    "\n",
    "![](../images/spark_job.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Joining the Data\n",
    "\n",
    "From here, we could read in all the partitioned feature matrices and build a single feature matrix, or if we have a model that supports [incremental (also known as on-line) learning](https://en.wikipedia.org/wiki/Incremental_learning), we can train it with one partition at a time. With all of the data in S3, we can access it from any machine which means we don't have to worry about losing data through stopping/starting machines."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>msno</th>\n",
       "      <th>time</th>\n",
       "      <th>city</th>\n",
       "      <th>bd</th>\n",
       "      <th>registered_via</th>\n",
       "      <th>gender</th>\n",
       "      <th>SUM(logs.num_25)</th>\n",
       "      <th>SUM(logs.num_50)</th>\n",
       "      <th>SUM(logs.num_75)</th>\n",
       "      <th>SUM(logs.num_985)</th>\n",
       "      <th>...</th>\n",
       "      <th>WEEKEND(LAST(transactions.membership_expire_date))</th>\n",
       "      <th>DAY(LAST(logs.date))</th>\n",
       "      <th>DAY(LAST(transactions.transaction_date))</th>\n",
       "      <th>DAY(LAST(transactions.membership_expire_date))</th>\n",
       "      <th>MONTH(LAST(logs.date))</th>\n",
       "      <th>MONTH(LAST(transactions.transaction_date))</th>\n",
       "      <th>MONTH(LAST(transactions.membership_expire_date))</th>\n",
       "      <th>label</th>\n",
       "      <th>days_to_churn</th>\n",
       "      <th>churn_date</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>++Q7h1vrVXrpWGnIDtttZ2O6bYGkF1fwOkAF5na5hF4=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>5.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>+KEJoiWQSU7N0XwM9XCa+wJNvyxhkO5g5uG1GAIWeRI=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>13.0</td>\n",
       "      <td>27.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>female</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>451.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>+WbnkpvQ8I4qLjXiHeASos5MzW0qJby5mSBNbFkqGN4=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>7.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>2.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>+oDiKDaW9HDo7BV23lD2O2zAz3UQ2oY+CcDRAMX36Wc=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>1.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>7.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>0.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>+vMYXB1GFE6vREfLE2RzbLNU9JZSlJdXGAD82xxSh4o=</td>\n",
       "      <td>2015-01-01</td>\n",
       "      <td>5.0</td>\n",
       "      <td>26.0</td>\n",
       "      <td>9.0</td>\n",
       "      <td>female</td>\n",
       "      <td>25.0</td>\n",
       "      <td>7.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>4.0</td>\n",
       "      <td>...</td>\n",
       "      <td>0.0</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>NaN</td>\n",
       "      <td>0.0</td>\n",
       "      <td>218.0</td>\n",
       "      <td>0</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 253 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "                                           msno        time  city    bd  \\\n",
       "0  ++Q7h1vrVXrpWGnIDtttZ2O6bYGkF1fwOkAF5na5hF4=  2015-01-01   1.0   0.0   \n",
       "1  +KEJoiWQSU7N0XwM9XCa+wJNvyxhkO5g5uG1GAIWeRI=  2015-01-01  13.0  27.0   \n",
       "2  +WbnkpvQ8I4qLjXiHeASos5MzW0qJby5mSBNbFkqGN4=  2015-01-01   1.0   0.0   \n",
       "3  +oDiKDaW9HDo7BV23lD2O2zAz3UQ2oY+CcDRAMX36Wc=  2015-01-01   1.0   0.0   \n",
       "4  +vMYXB1GFE6vREfLE2RzbLNU9JZSlJdXGAD82xxSh4o=  2015-01-01   5.0  26.0   \n",
       "\n",
       "   registered_via  gender  SUM(logs.num_25)  SUM(logs.num_50)  \\\n",
       "0             9.0     NaN               5.0               1.0   \n",
       "1             9.0  female               0.0               0.0   \n",
       "2             7.0     NaN               1.0               0.0   \n",
       "3             7.0     NaN               0.0               0.0   \n",
       "4             9.0  female              25.0               7.0   \n",
       "\n",
       "   SUM(logs.num_75)  SUM(logs.num_985)     ...      \\\n",
       "0               0.0                0.0     ...       \n",
       "1               0.0                0.0     ...       \n",
       "2               0.0                2.0     ...       \n",
       "3               0.0                0.0     ...       \n",
       "4               1.0                4.0     ...       \n",
       "\n",
       "   WEEKEND(LAST(transactions.membership_expire_date))  DAY(LAST(logs.date))  \\\n",
       "0                                                0.0                    1.0   \n",
       "1                                                0.0                    NaN   \n",
       "2                                                0.0                    1.0   \n",
       "3                                                0.0                    NaN   \n",
       "4                                                0.0                    1.0   \n",
       "\n",
       "   DAY(LAST(transactions.transaction_date))  \\\n",
       "0                                       NaN   \n",
       "1                                       NaN   \n",
       "2                                       NaN   \n",
       "3                                       NaN   \n",
       "4                                       NaN   \n",
       "\n",
       "   DAY(LAST(transactions.membership_expire_date))  MONTH(LAST(logs.date))  \\\n",
       "0                                             NaN                     1.0   \n",
       "1                                             NaN                     NaN   \n",
       "2                                             NaN                     1.0   \n",
       "3                                             NaN                     NaN   \n",
       "4                                             NaN                     1.0   \n",
       "\n",
       "   MONTH(LAST(transactions.transaction_date))  \\\n",
       "0                                         NaN   \n",
       "1                                         NaN   \n",
       "2                                         NaN   \n",
       "3                                         NaN   \n",
       "4                                         NaN   \n",
       "\n",
       "   MONTH(LAST(transactions.membership_expire_date))  label  days_to_churn  \\\n",
       "0                                               NaN    0.0            NaN   \n",
       "1                                               NaN    0.0          451.0   \n",
       "2                                               NaN    0.0            NaN   \n",
       "3                                               NaN    0.0            NaN   \n",
       "4                                               NaN    0.0          218.0   \n",
       "\n",
       "   churn_date  \n",
       "0         NaN  \n",
       "1           0  \n",
       "2         NaN  \n",
       "3         NaN  \n",
       "4           0  \n",
       "\n",
       "[5 rows x 253 columns]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "feature_matrix = pd.read_csv('s3://customer-churn-spark/p999/MS-31_feature_matrix.csv', low_memory = False)\n",
    "feature_matrix.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Conclusions\n",
    "\n",
    "In this notebook, we saw how to distribute feature engineering in Featuretools using the Spark framework. This big-data processing technology lets us use multiple computers to parallelize calculations, resulting in efficient data science workflows even on large datasets. \n",
    "\n",
    "The basic approach is:\n",
    "\n",
    "1. Divide data into independent partitions\n",
    "2. Run each subset in parallel with a different worker\n",
    "3. Join results together if necessary \n",
    "\n",
    "The nice part about using frameworks such as Dask and Spark with PySpark is we don't have to change the underlying Featuretools code. We write our code in native Python, change the backend running the calculations, and distribute the calculations across a cluster of machines. Using this approach, we'll be able to scale to any size datasets and take on even more exciting data science and machine learning problems. \n",
    "\n",
    "## Next Steps\n",
    "\n",
    "The final step of the machine learning pipeline is to build a model to make predictions for these features. This is implemented in the `Modeling` notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": false
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}