Notebook_2_FeatureEngineering_RollingCompute.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Notebook #2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook, we worked with the result dataset from Notebook #1 and computed rolling statistics (mean, difference, std, max, min) for a list of features over various time windows.  \n",
    "This was the most time consuming and computational expensive part of the entire tutorial. We encountered some roadblocks and found some workarounds. Please see below for more details."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as F\n",
    "import time\n",
    "import subprocess\n",
    "import sys\n",
    "import os\n",
    "import re\n",
    "\n",
    "from pyspark import SparkConf\n",
    "from pyspark import SparkContext\n",
    "from pyspark import SQLContext\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions import col,udf,lag,date_add,explode,lit,concat,unix_timestamp\n",
    "from pyspark.sql.dataframe import *\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.types import DateType\n",
    "from datetime import datetime, timedelta\n",
    "from pyspark.sql import Row\n",
    "\n",
    "start_time = time.time()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define list of features for rolling compute, window sizes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "46\n"
     ]
    }
   ],
   "source": [
    "rolling_features = [\n",
    "    'warn_type1_total', 'warn_type2_total', \n",
    "    'pca_1_warn','pca_2_warn', 'pca_3_warn', 'pca_4_warn', 'pca_5_warn',\n",
    "    'pca_6_warn','pca_7_warn', 'pca_8_warn', 'pca_9_warn', 'pca_10_warn',\n",
    "    'pca_11_warn','pca_12_warn', 'pca_13_warn', 'pca_14_warn', 'pca_15_warn',\n",
    "    'pca_16_warn','pca_17_warn', 'pca_18_warn', 'pca_19_warn', 'pca_20_warn',\n",
    "    'problem_type_1', 'problem_type_2', 'problem_type_3','problem_type_4',\n",
    "    'problem_type_1_per_usage1','problem_type_2_per_usage1',\n",
    "    'problem_type_3_per_usage1','problem_type_4_per_usage1',\n",
    "    'problem_type_1_per_usage2','problem_type_2_per_usage2',\n",
    "    'problem_type_3_per_usage2','problem_type_4_per_usage2',                \n",
    "    'fault_code_type_1_count', 'fault_code_type_2_count', 'fault_code_type_3_count', 'fault_code_type_4_count',                          \n",
    "    'fault_code_type_1_count_per_usage1','fault_code_type_2_count_per_usage1',\n",
    "    'fault_code_type_3_count_per_usage1', 'fault_code_type_4_count_per_usage1',\n",
    "    'fault_code_type_1_count_per_usage2','fault_code_type_2_count_per_usage2',\n",
    "    'fault_code_type_3_count_per_usage2', 'fault_code_type_4_count_per_usage2']\n",
    "               \n",
    "# lag window 3, 7, 14, 30, 90 days\n",
    "lags = [3, 7, 14, 30, 90]\n",
    "\n",
    "print(len(rolling_features))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## What issues we encountered using Pyspark and how we solved them?\n",
    "\n",
    "-  If the entire list of **46 features** and **5 time windows** were computed for **5 different types of rolling** (mean, difference, std, max, min) all in one go, we always ran into \"StackOverFlow\" error. \n",
    "-  It was because the lineage was too long and Spark could not handle it.\n",
    "-  We could either create checkPoint and materialize it throughout the process.\n",
    "-  OR break the workload into chunks and save the result from each chunk as parquet file.\n",
    "\n",
    "## A few things we found helpful:\n",
    "-  Before the rolling compute, save the upstream work as a parquet file in Notebook_1 (\"Notebook_1_DataCleansing_FeatureEngineering\"). It will speed up the whole process because we no need to repeat all the previous steps. It will also help reduce the lineage.\n",
    "-  Print out the lag and feature name to track progress.\n",
    "-  Use \"htop\" command from the terminal to keep track how many CPUs are running for a particular task. For rolling compute, we were considering two potential approaches: 1) Use Spark clusters on HDInsight to perform rolling compute in parallel; 2) Use single node Spark on a powerful VM. By looking at htop dashboard, we saw all the 32 cores were running at the same time for a single task (for example compute rolling mean). So if say we divide the workload onto multiple nodes and each node runs a type of rolling compute, the amount of time taken will be comparable with running everything in a sequential manner on a single node Spark on a powerful machine.\n",
    "-  Use \"%%time\" for each cell to get an estimate of the total run time, we will then have a better idea where and what to optimze the process.\n",
    "-  Materialize the intermediate results by either caching in memory or writing as parquet files. We chose to save as parquet files because we did not want to repeat the compute again in case cache() did not work or any part of the rolling compute did not work.\n",
    "-  Why parquet? There are many reasons, just to name a few: parquet not only saves the data but also the schema, it is a preferred file format by Spark, you are allowed to read only the data you need, etc..\n",
    "\n",
    "<br>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rolling Mean"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Lag = 3, Column = warn_type1_total\n",
      "Lag = 3, Column = warn_type2_total\n",
      "Lag = 3, Column = pca_1_warn\n",
      "Lag = 3, Column = pca_2_warn\n",
      "Lag = 3, Column = pca_3_warn\n",
      "Lag = 3, Column = pca_4_warn\n",
      "Lag = 3, Column = pca_5_warn\n",
      "Lag = 3, Column = pca_6_warn\n",
      "Lag = 3, Column = pca_7_warn\n",
      "Lag = 3, Column = pca_8_warn\n",
      "Lag = 3, Column = pca_9_warn\n",
      "Lag = 3, Column = pca_10_warn\n",
      "Lag = 3, Column = pca_11_warn\n",
      "Lag = 3, Column = pca_12_warn\n",
      "Lag = 3, Column = pca_13_warn\n",
      "Lag = 3, Column = pca_14_warn\n",
      "Lag = 3, Column = pca_15_warn\n",
      "Lag = 3, Column = pca_16_warn\n",
      "Lag = 3, Column = pca_17_warn\n",
      "Lag = 3, Column = pca_18_warn\n",
      "Lag = 3, Column = pca_19_warn\n",
      "Lag = 3, Column = pca_20_warn\n",
      "Lag = 3, Column = problem_type_1\n",
      "Lag = 3, Column = problem_type_2\n",
      "Lag = 3, Column = problem_type_3\n",
      "Lag = 3, Column = problem_type_4\n",
      "Lag = 3, Column = problem_type_1_per_usage1\n",
      "Lag = 3, Column = problem_type_2_per_usage1\n",
      "Lag = 3, Column = problem_type_3_per_usage1\n",
      "Lag = 3, Column = problem_type_4_per_usage1\n",
      "Lag = 3, Column = problem_type_1_per_usage2\n",
      "Lag = 3, Column = problem_type_2_per_usage2\n",
      "Lag = 3, Column = problem_type_3_per_usage2\n",
      "Lag = 3, Column = problem_type_4_per_usage2\n",
      "Lag = 3, Column = fault_code_type_1_count\n",
      "Lag = 3, Column = fault_code_type_2_count\n",
      "Lag = 3, Column = fault_code_type_3_count\n",
      "Lag = 3, Column = fault_code_type_4_count\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 7, Column = warn_type1_total\n",
      "Lag = 7, Column = warn_type2_total\n",
      "Lag = 7, Column = pca_1_warn\n",
      "Lag = 7, Column = pca_2_warn\n",
      "Lag = 7, Column = pca_3_warn\n",
      "Lag = 7, Column = pca_4_warn\n",
      "Lag = 7, Column = pca_5_warn\n",
      "Lag = 7, Column = pca_6_warn\n",
      "Lag = 7, Column = pca_7_warn\n",
      "Lag = 7, Column = pca_8_warn\n",
      "Lag = 7, Column = pca_9_warn\n",
      "Lag = 7, Column = pca_10_warn\n",
      "Lag = 7, Column = pca_11_warn\n",
      "Lag = 7, Column = pca_12_warn\n",
      "Lag = 7, Column = pca_13_warn\n",
      "Lag = 7, Column = pca_14_warn\n",
      "Lag = 7, Column = pca_15_warn\n",
      "Lag = 7, Column = pca_16_warn\n",
      "Lag = 7, Column = pca_17_warn\n",
      "Lag = 7, Column = pca_18_warn\n",
      "Lag = 7, Column = pca_19_warn\n",
      "Lag = 7, Column = pca_20_warn\n",
      "Lag = 7, Column = problem_type_1\n",
      "Lag = 7, Column = problem_type_2\n",
      "Lag = 7, Column = problem_type_3\n",
      "Lag = 7, Column = problem_type_4\n",
      "Lag = 7, Column = problem_type_1_per_usage1\n",
      "Lag = 7, Column = problem_type_2_per_usage1\n",
      "Lag = 7, Column = problem_type_3_per_usage1\n",
      "Lag = 7, Column = problem_type_4_per_usage1\n",
      "Lag = 7, Column = problem_type_1_per_usage2\n",
      "Lag = 7, Column = problem_type_2_per_usage2\n",
      "Lag = 7, Column = problem_type_3_per_usage2\n",
      "Lag = 7, Column = problem_type_4_per_usage2\n",
      "Lag = 7, Column = fault_code_type_1_count\n",
      "Lag = 7, Column = fault_code_type_2_count\n",
      "Lag = 7, Column = fault_code_type_3_count\n",
      "Lag = 7, Column = fault_code_type_4_count\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 14, Column = warn_type1_total\n",
      "Lag = 14, Column = warn_type2_total\n",
      "Lag = 14, Column = pca_1_warn\n",
      "Lag = 14, Column = pca_2_warn\n",
      "Lag = 14, Column = pca_3_warn\n",
      "Lag = 14, Column = pca_4_warn\n",
      "Lag = 14, Column = pca_5_warn\n",
      "Lag = 14, Column = pca_6_warn\n",
      "Lag = 14, Column = pca_7_warn\n",
      "Lag = 14, Column = pca_8_warn\n",
      "Lag = 14, Column = pca_9_warn\n",
      "Lag = 14, Column = pca_10_warn\n",
      "Lag = 14, Column = pca_11_warn\n",
      "Lag = 14, Column = pca_12_warn\n",
      "Lag = 14, Column = pca_13_warn\n",
      "Lag = 14, Column = pca_14_warn\n",
      "Lag = 14, Column = pca_15_warn\n",
      "Lag = 14, Column = pca_16_warn\n",
      "Lag = 14, Column = pca_17_warn\n",
      "Lag = 14, Column = pca_18_warn\n",
      "Lag = 14, Column = pca_19_warn\n",
      "Lag = 14, Column = pca_20_warn\n",
      "Lag = 14, Column = problem_type_1\n",
      "Lag = 14, Column = problem_type_2\n",
      "Lag = 14, Column = problem_type_3\n",
      "Lag = 14, Column = problem_type_4\n",
      "Lag = 14, Column = problem_type_1_per_usage1\n",
      "Lag = 14, Column = problem_type_2_per_usage1\n",
      "Lag = 14, Column = problem_type_3_per_usage1\n",
      "Lag = 14, Column = problem_type_4_per_usage1\n",
      "Lag = 14, Column = problem_type_1_per_usage2\n",
      "Lag = 14, Column = problem_type_2_per_usage2\n",
      "Lag = 14, Column = problem_type_3_per_usage2\n",
      "Lag = 14, Column = problem_type_4_per_usage2\n",
      "Lag = 14, Column = fault_code_type_1_count\n",
      "Lag = 14, Column = fault_code_type_2_count\n",
      "Lag = 14, Column = fault_code_type_3_count\n",
      "Lag = 14, Column = fault_code_type_4_count\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 30, Column = warn_type1_total\n",
      "Lag = 30, Column = warn_type2_total\n",
      "Lag = 30, Column = pca_1_warn\n",
      "Lag = 30, Column = pca_2_warn\n",
      "Lag = 30, Column = pca_3_warn\n",
      "Lag = 30, Column = pca_4_warn\n",
      "Lag = 30, Column = pca_5_warn\n",
      "Lag = 30, Column = pca_6_warn\n",
      "Lag = 30, Column = pca_7_warn\n",
      "Lag = 30, Column = pca_8_warn\n",
      "Lag = 30, Column = pca_9_warn\n",
      "Lag = 30, Column = pca_10_warn\n",
      "Lag = 30, Column = pca_11_warn\n",
      "Lag = 30, Column = pca_12_warn\n",
      "Lag = 30, Column = pca_13_warn\n",
      "Lag = 30, Column = pca_14_warn\n",
      "Lag = 30, Column = pca_15_warn\n",
      "Lag = 30, Column = pca_16_warn\n",
      "Lag = 30, Column = pca_17_warn\n",
      "Lag = 30, Column = pca_18_warn\n",
      "Lag = 30, Column = pca_19_warn\n",
      "Lag = 30, Column = pca_20_warn\n",
      "Lag = 30, Column = problem_type_1\n",
      "Lag = 30, Column = problem_type_2\n",
      "Lag = 30, Column = problem_type_3\n",
      "Lag = 30, Column = problem_type_4\n",
      "Lag = 30, Column = problem_type_1_per_usage1\n",
      "Lag = 30, Column = problem_type_2_per_usage1\n",
      "Lag = 30, Column = problem_type_3_per_usage1\n",
      "Lag = 30, Column = problem_type_4_per_usage1\n",
      "Lag = 30, Column = problem_type_1_per_usage2\n",
      "Lag = 30, Column = problem_type_2_per_usage2\n",
      "Lag = 30, Column = problem_type_3_per_usage2\n",
      "Lag = 30, Column = problem_type_4_per_usage2\n",
      "Lag = 30, Column = fault_code_type_1_count\n",
      "Lag = 30, Column = fault_code_type_2_count\n",
      "Lag = 30, Column = fault_code_type_3_count\n",
      "Lag = 30, Column = fault_code_type_4_count\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 90, Column = warn_type1_total\n",
      "Lag = 90, Column = warn_type2_total\n",
      "Lag = 90, Column = pca_1_warn\n",
      "Lag = 90, Column = pca_2_warn\n",
      "Lag = 90, Column = pca_3_warn\n",
      "Lag = 90, Column = pca_4_warn\n",
      "Lag = 90, Column = pca_5_warn\n",
      "Lag = 90, Column = pca_6_warn\n",
      "Lag = 90, Column = pca_7_warn\n",
      "Lag = 90, Column = pca_8_warn\n",
      "Lag = 90, Column = pca_9_warn\n",
      "Lag = 90, Column = pca_10_warn\n",
      "Lag = 90, Column = pca_11_warn\n",
      "Lag = 90, Column = pca_12_warn\n",
      "Lag = 90, Column = pca_13_warn\n",
      "Lag = 90, Column = pca_14_warn\n",
      "Lag = 90, Column = pca_15_warn\n",
      "Lag = 90, Column = pca_16_warn\n",
      "Lag = 90, Column = pca_17_warn\n",
      "Lag = 90, Column = pca_18_warn\n",
      "Lag = 90, Column = pca_19_warn\n",
      "Lag = 90, Column = pca_20_warn\n",
      "Lag = 90, Column = problem_type_1\n",
      "Lag = 90, Column = problem_type_2\n",
      "Lag = 90, Column = problem_type_3\n",
      "Lag = 90, Column = problem_type_4\n",
      "Lag = 90, Column = problem_type_1_per_usage1\n",
      "Lag = 90, Column = problem_type_2_per_usage1\n",
      "Lag = 90, Column = problem_type_3_per_usage1\n",
      "Lag = 90, Column = problem_type_4_per_usage1\n",
      "Lag = 90, Column = problem_type_1_per_usage2\n",
      "Lag = 90, Column = problem_type_2_per_usage2\n",
      "Lag = 90, Column = problem_type_3_per_usage2\n",
      "Lag = 90, Column = problem_type_4_per_usage2\n",
      "Lag = 90, Column = fault_code_type_1_count\n",
      "Lag = 90, Column = fault_code_type_2_count\n",
      "Lag = 90, Column = fault_code_type_3_count\n",
      "Lag = 90, Column = fault_code_type_4_count\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage2\n",
      "CPU times: user 848 ms, sys: 279 ms, total: 1.13 s\n",
      "Wall time: 28min 6s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Load result dataset from Notebook #1\n",
    "df = sqlContext.read.parquet('/mnt/resource/PysparkExample/notebook1_result.parquet')\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('deviceid').orderBy('date').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        df = df.withColumn(col_name+'_rollingmean_'+str(lag_n), F.avg(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))\n",
    "\n",
    "# Save the intermediate result for downstream work\n",
    "df.write.mode('overwrite').parquet('/mnt/resource/PysparkExample/data_rollingmean.parquet')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rolling Difference"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Lag = 3, Column = warn_type1_total\n",
      "Lag = 3, Column = warn_type2_total\n",
      "Lag = 3, Column = pca_1_warn\n",
      "Lag = 3, Column = pca_2_warn\n",
      "Lag = 3, Column = pca_3_warn\n",
      "Lag = 3, Column = pca_4_warn\n",
      "Lag = 3, Column = pca_5_warn\n",
      "Lag = 3, Column = pca_6_warn\n",
      "Lag = 3, Column = pca_7_warn\n",
      "Lag = 3, Column = pca_8_warn\n",
      "Lag = 3, Column = pca_9_warn\n",
      "Lag = 3, Column = pca_10_warn\n",
      "Lag = 3, Column = pca_11_warn\n",
      "Lag = 3, Column = pca_12_warn\n",
      "Lag = 3, Column = pca_13_warn\n",
      "Lag = 3, Column = pca_14_warn\n",
      "Lag = 3, Column = pca_15_warn\n",
      "Lag = 3, Column = pca_16_warn\n",
      "Lag = 3, Column = pca_17_warn\n",
      "Lag = 3, Column = pca_18_warn\n",
      "Lag = 3, Column = pca_19_warn\n",
      "Lag = 3, Column = pca_20_warn\n",
      "Lag = 3, Column = problem_type_1\n",
      "Lag = 3, Column = problem_type_2\n",
      "Lag = 3, Column = problem_type_3\n",
      "Lag = 3, Column = problem_type_4\n",
      "Lag = 3, Column = problem_type_1_per_usage1\n",
      "Lag = 3, Column = problem_type_2_per_usage1\n",
      "Lag = 3, Column = problem_type_3_per_usage1\n",
      "Lag = 3, Column = problem_type_4_per_usage1\n",
      "Lag = 3, Column = problem_type_1_per_usage2\n",
      "Lag = 3, Column = problem_type_2_per_usage2\n",
      "Lag = 3, Column = problem_type_3_per_usage2\n",
      "Lag = 3, Column = problem_type_4_per_usage2\n",
      "Lag = 3, Column = fault_code_type_1_count\n",
      "Lag = 3, Column = fault_code_type_2_count\n",
      "Lag = 3, Column = fault_code_type_3_count\n",
      "Lag = 3, Column = fault_code_type_4_count\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 7, Column = warn_type1_total\n",
      "Lag = 7, Column = warn_type2_total\n",
      "Lag = 7, Column = pca_1_warn\n",
      "Lag = 7, Column = pca_2_warn\n",
      "Lag = 7, Column = pca_3_warn\n",
      "Lag = 7, Column = pca_4_warn\n",
      "Lag = 7, Column = pca_5_warn\n",
      "Lag = 7, Column = pca_6_warn\n",
      "Lag = 7, Column = pca_7_warn\n",
      "Lag = 7, Column = pca_8_warn\n",
      "Lag = 7, Column = pca_9_warn\n",
      "Lag = 7, Column = pca_10_warn\n",
      "Lag = 7, Column = pca_11_warn\n",
      "Lag = 7, Column = pca_12_warn\n",
      "Lag = 7, Column = pca_13_warn\n",
      "Lag = 7, Column = pca_14_warn\n",
      "Lag = 7, Column = pca_15_warn\n",
      "Lag = 7, Column = pca_16_warn\n",
      "Lag = 7, Column = pca_17_warn\n",
      "Lag = 7, Column = pca_18_warn\n",
      "Lag = 7, Column = pca_19_warn\n",
      "Lag = 7, Column = pca_20_warn\n",
      "Lag = 7, Column = problem_type_1\n",
      "Lag = 7, Column = problem_type_2\n",
      "Lag = 7, Column = problem_type_3\n",
      "Lag = 7, Column = problem_type_4\n",
      "Lag = 7, Column = problem_type_1_per_usage1\n",
      "Lag = 7, Column = problem_type_2_per_usage1\n",
      "Lag = 7, Column = problem_type_3_per_usage1\n",
      "Lag = 7, Column = problem_type_4_per_usage1\n",
      "Lag = 7, Column = problem_type_1_per_usage2\n",
      "Lag = 7, Column = problem_type_2_per_usage2\n",
      "Lag = 7, Column = problem_type_3_per_usage2\n",
      "Lag = 7, Column = problem_type_4_per_usage2\n",
      "Lag = 7, Column = fault_code_type_1_count\n",
      "Lag = 7, Column = fault_code_type_2_count\n",
      "Lag = 7, Column = fault_code_type_3_count\n",
      "Lag = 7, Column = fault_code_type_4_count\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 14, Column = warn_type1_total\n",
      "Lag = 14, Column = warn_type2_total\n",
      "Lag = 14, Column = pca_1_warn\n",
      "Lag = 14, Column = pca_2_warn\n",
      "Lag = 14, Column = pca_3_warn\n",
      "Lag = 14, Column = pca_4_warn\n",
      "Lag = 14, Column = pca_5_warn\n",
      "Lag = 14, Column = pca_6_warn\n",
      "Lag = 14, Column = pca_7_warn\n",
      "Lag = 14, Column = pca_8_warn\n",
      "Lag = 14, Column = pca_9_warn\n",
      "Lag = 14, Column = pca_10_warn\n",
      "Lag = 14, Column = pca_11_warn\n",
      "Lag = 14, Column = pca_12_warn\n",
      "Lag = 14, Column = pca_13_warn\n",
      "Lag = 14, Column = pca_14_warn\n",
      "Lag = 14, Column = pca_15_warn\n",
      "Lag = 14, Column = pca_16_warn\n",
      "Lag = 14, Column = pca_17_warn\n",
      "Lag = 14, Column = pca_18_warn\n",
      "Lag = 14, Column = pca_19_warn\n",
      "Lag = 14, Column = pca_20_warn\n",
      "Lag = 14, Column = problem_type_1\n",
      "Lag = 14, Column = problem_type_2\n",
      "Lag = 14, Column = problem_type_3\n",
      "Lag = 14, Column = problem_type_4\n",
      "Lag = 14, Column = problem_type_1_per_usage1\n",
      "Lag = 14, Column = problem_type_2_per_usage1\n",
      "Lag = 14, Column = problem_type_3_per_usage1\n",
      "Lag = 14, Column = problem_type_4_per_usage1\n",
      "Lag = 14, Column = problem_type_1_per_usage2\n",
      "Lag = 14, Column = problem_type_2_per_usage2\n",
      "Lag = 14, Column = problem_type_3_per_usage2\n",
      "Lag = 14, Column = problem_type_4_per_usage2\n",
      "Lag = 14, Column = fault_code_type_1_count\n",
      "Lag = 14, Column = fault_code_type_2_count\n",
      "Lag = 14, Column = fault_code_type_3_count\n",
      "Lag = 14, Column = fault_code_type_4_count\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 30, Column = warn_type1_total\n",
      "Lag = 30, Column = warn_type2_total\n",
      "Lag = 30, Column = pca_1_warn\n",
      "Lag = 30, Column = pca_2_warn\n",
      "Lag = 30, Column = pca_3_warn\n",
      "Lag = 30, Column = pca_4_warn\n",
      "Lag = 30, Column = pca_5_warn\n",
      "Lag = 30, Column = pca_6_warn\n",
      "Lag = 30, Column = pca_7_warn\n",
      "Lag = 30, Column = pca_8_warn\n",
      "Lag = 30, Column = pca_9_warn\n",
      "Lag = 30, Column = pca_10_warn\n",
      "Lag = 30, Column = pca_11_warn\n",
      "Lag = 30, Column = pca_12_warn\n",
      "Lag = 30, Column = pca_13_warn\n",
      "Lag = 30, Column = pca_14_warn\n",
      "Lag = 30, Column = pca_15_warn\n",
      "Lag = 30, Column = pca_16_warn\n",
      "Lag = 30, Column = pca_17_warn\n",
      "Lag = 30, Column = pca_18_warn\n",
      "Lag = 30, Column = pca_19_warn\n",
      "Lag = 30, Column = pca_20_warn\n",
      "Lag = 30, Column = problem_type_1\n",
      "Lag = 30, Column = problem_type_2\n",
      "Lag = 30, Column = problem_type_3\n",
      "Lag = 30, Column = problem_type_4\n",
      "Lag = 30, Column = problem_type_1_per_usage1\n",
      "Lag = 30, Column = problem_type_2_per_usage1\n",
      "Lag = 30, Column = problem_type_3_per_usage1\n",
      "Lag = 30, Column = problem_type_4_per_usage1\n",
      "Lag = 30, Column = problem_type_1_per_usage2\n",
      "Lag = 30, Column = problem_type_2_per_usage2\n",
      "Lag = 30, Column = problem_type_3_per_usage2\n",
      "Lag = 30, Column = problem_type_4_per_usage2\n",
      "Lag = 30, Column = fault_code_type_1_count\n",
      "Lag = 30, Column = fault_code_type_2_count\n",
      "Lag = 30, Column = fault_code_type_3_count\n",
      "Lag = 30, Column = fault_code_type_4_count\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 90, Column = warn_type1_total\n",
      "Lag = 90, Column = warn_type2_total\n",
      "Lag = 90, Column = pca_1_warn\n",
      "Lag = 90, Column = pca_2_warn\n",
      "Lag = 90, Column = pca_3_warn\n",
      "Lag = 90, Column = pca_4_warn\n",
      "Lag = 90, Column = pca_5_warn\n",
      "Lag = 90, Column = pca_6_warn\n",
      "Lag = 90, Column = pca_7_warn\n",
      "Lag = 90, Column = pca_8_warn\n",
      "Lag = 90, Column = pca_9_warn\n",
      "Lag = 90, Column = pca_10_warn\n",
      "Lag = 90, Column = pca_11_warn\n",
      "Lag = 90, Column = pca_12_warn\n",
      "Lag = 90, Column = pca_13_warn\n",
      "Lag = 90, Column = pca_14_warn\n",
      "Lag = 90, Column = pca_15_warn\n",
      "Lag = 90, Column = pca_16_warn\n",
      "Lag = 90, Column = pca_17_warn\n",
      "Lag = 90, Column = pca_18_warn\n",
      "Lag = 90, Column = pca_19_warn\n",
      "Lag = 90, Column = pca_20_warn\n",
      "Lag = 90, Column = problem_type_1\n",
      "Lag = 90, Column = problem_type_2\n",
      "Lag = 90, Column = problem_type_3\n",
      "Lag = 90, Column = problem_type_4\n",
      "Lag = 90, Column = problem_type_1_per_usage1\n",
      "Lag = 90, Column = problem_type_2_per_usage1\n",
      "Lag = 90, Column = problem_type_3_per_usage1\n",
      "Lag = 90, Column = problem_type_4_per_usage1\n",
      "Lag = 90, Column = problem_type_1_per_usage2\n",
      "Lag = 90, Column = problem_type_2_per_usage2\n",
      "Lag = 90, Column = problem_type_3_per_usage2\n",
      "Lag = 90, Column = problem_type_4_per_usage2\n",
      "Lag = 90, Column = fault_code_type_1_count\n",
      "Lag = 90, Column = fault_code_type_2_count\n",
      "Lag = 90, Column = fault_code_type_3_count\n",
      "Lag = 90, Column = fault_code_type_4_count\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage2\n",
      "CPU times: user 1.18 s, sys: 383 ms, total: 1.56 s\n",
      "Wall time: 45min 12s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Load result dataset from Notebook #1\n",
    "df = sqlContext.read.parquet('/mnt/resource/PysparkExample/notebook1_result.parquet')\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('deviceid').orderBy('date').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        df = df.withColumn(col_name+'_rollingdiff_'+str(lag_n), col(col_name)-F.avg(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))\n",
    "\n",
    "rollingdiff = df.select(['key'] + list(s for s in df.columns if \"rollingdiff\" in s))\n",
    "\n",
    "# Save the intermediate result for downstream work\n",
    "rollingdiff.write.mode('overwrite').parquet('/mnt/resource/PysparkExample/rollingdiff.parquet')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rolling Std"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Lag = 3, Column = warn_type1_total\n",
      "Lag = 3, Column = warn_type2_total\n",
      "Lag = 3, Column = pca_1_warn\n",
      "Lag = 3, Column = pca_2_warn\n",
      "Lag = 3, Column = pca_3_warn\n",
      "Lag = 3, Column = pca_4_warn\n",
      "Lag = 3, Column = pca_5_warn\n",
      "Lag = 3, Column = pca_6_warn\n",
      "Lag = 3, Column = pca_7_warn\n",
      "Lag = 3, Column = pca_8_warn\n",
      "Lag = 3, Column = pca_9_warn\n",
      "Lag = 3, Column = pca_10_warn\n",
      "Lag = 3, Column = pca_11_warn\n",
      "Lag = 3, Column = pca_12_warn\n",
      "Lag = 3, Column = pca_13_warn\n",
      "Lag = 3, Column = pca_14_warn\n",
      "Lag = 3, Column = pca_15_warn\n",
      "Lag = 3, Column = pca_16_warn\n",
      "Lag = 3, Column = pca_17_warn\n",
      "Lag = 3, Column = pca_18_warn\n",
      "Lag = 3, Column = pca_19_warn\n",
      "Lag = 3, Column = pca_20_warn\n",
      "Lag = 3, Column = problem_type_1\n",
      "Lag = 3, Column = problem_type_2\n",
      "Lag = 3, Column = problem_type_3\n",
      "Lag = 3, Column = problem_type_4\n",
      "Lag = 3, Column = problem_type_1_per_usage1\n",
      "Lag = 3, Column = problem_type_2_per_usage1\n",
      "Lag = 3, Column = problem_type_3_per_usage1\n",
      "Lag = 3, Column = problem_type_4_per_usage1\n",
      "Lag = 3, Column = problem_type_1_per_usage2\n",
      "Lag = 3, Column = problem_type_2_per_usage2\n",
      "Lag = 3, Column = problem_type_3_per_usage2\n",
      "Lag = 3, Column = problem_type_4_per_usage2\n",
      "Lag = 3, Column = fault_code_type_1_count\n",
      "Lag = 3, Column = fault_code_type_2_count\n",
      "Lag = 3, Column = fault_code_type_3_count\n",
      "Lag = 3, Column = fault_code_type_4_count\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 7, Column = warn_type1_total\n",
      "Lag = 7, Column = warn_type2_total\n",
      "Lag = 7, Column = pca_1_warn\n",
      "Lag = 7, Column = pca_2_warn\n",
      "Lag = 7, Column = pca_3_warn\n",
      "Lag = 7, Column = pca_4_warn\n",
      "Lag = 7, Column = pca_5_warn\n",
      "Lag = 7, Column = pca_6_warn\n",
      "Lag = 7, Column = pca_7_warn\n",
      "Lag = 7, Column = pca_8_warn\n",
      "Lag = 7, Column = pca_9_warn\n",
      "Lag = 7, Column = pca_10_warn\n",
      "Lag = 7, Column = pca_11_warn\n",
      "Lag = 7, Column = pca_12_warn\n",
      "Lag = 7, Column = pca_13_warn\n",
      "Lag = 7, Column = pca_14_warn\n",
      "Lag = 7, Column = pca_15_warn\n",
      "Lag = 7, Column = pca_16_warn\n",
      "Lag = 7, Column = pca_17_warn\n",
      "Lag = 7, Column = pca_18_warn\n",
      "Lag = 7, Column = pca_19_warn\n",
      "Lag = 7, Column = pca_20_warn\n",
      "Lag = 7, Column = problem_type_1\n",
      "Lag = 7, Column = problem_type_2\n",
      "Lag = 7, Column = problem_type_3\n",
      "Lag = 7, Column = problem_type_4\n",
      "Lag = 7, Column = problem_type_1_per_usage1\n",
      "Lag = 7, Column = problem_type_2_per_usage1\n",
      "Lag = 7, Column = problem_type_3_per_usage1\n",
      "Lag = 7, Column = problem_type_4_per_usage1\n",
      "Lag = 7, Column = problem_type_1_per_usage2\n",
      "Lag = 7, Column = problem_type_2_per_usage2\n",
      "Lag = 7, Column = problem_type_3_per_usage2\n",
      "Lag = 7, Column = problem_type_4_per_usage2\n",
      "Lag = 7, Column = fault_code_type_1_count\n",
      "Lag = 7, Column = fault_code_type_2_count\n",
      "Lag = 7, Column = fault_code_type_3_count\n",
      "Lag = 7, Column = fault_code_type_4_count\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 14, Column = warn_type1_total\n",
      "Lag = 14, Column = warn_type2_total\n",
      "Lag = 14, Column = pca_1_warn\n",
      "Lag = 14, Column = pca_2_warn\n",
      "Lag = 14, Column = pca_3_warn\n",
      "Lag = 14, Column = pca_4_warn\n",
      "Lag = 14, Column = pca_5_warn\n",
      "Lag = 14, Column = pca_6_warn\n",
      "Lag = 14, Column = pca_7_warn\n",
      "Lag = 14, Column = pca_8_warn\n",
      "Lag = 14, Column = pca_9_warn\n",
      "Lag = 14, Column = pca_10_warn\n",
      "Lag = 14, Column = pca_11_warn\n",
      "Lag = 14, Column = pca_12_warn\n",
      "Lag = 14, Column = pca_13_warn\n",
      "Lag = 14, Column = pca_14_warn\n",
      "Lag = 14, Column = pca_15_warn\n",
      "Lag = 14, Column = pca_16_warn\n",
      "Lag = 14, Column = pca_17_warn\n",
      "Lag = 14, Column = pca_18_warn\n",
      "Lag = 14, Column = pca_19_warn\n",
      "Lag = 14, Column = pca_20_warn\n",
      "Lag = 14, Column = problem_type_1\n",
      "Lag = 14, Column = problem_type_2\n",
      "Lag = 14, Column = problem_type_3\n",
      "Lag = 14, Column = problem_type_4\n",
      "Lag = 14, Column = problem_type_1_per_usage1\n",
      "Lag = 14, Column = problem_type_2_per_usage1\n",
      "Lag = 14, Column = problem_type_3_per_usage1\n",
      "Lag = 14, Column = problem_type_4_per_usage1\n",
      "Lag = 14, Column = problem_type_1_per_usage2\n",
      "Lag = 14, Column = problem_type_2_per_usage2\n",
      "Lag = 14, Column = problem_type_3_per_usage2\n",
      "Lag = 14, Column = problem_type_4_per_usage2\n",
      "Lag = 14, Column = fault_code_type_1_count\n",
      "Lag = 14, Column = fault_code_type_2_count\n",
      "Lag = 14, Column = fault_code_type_3_count\n",
      "Lag = 14, Column = fault_code_type_4_count\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 30, Column = warn_type1_total\n",
      "Lag = 30, Column = warn_type2_total\n",
      "Lag = 30, Column = pca_1_warn\n",
      "Lag = 30, Column = pca_2_warn\n",
      "Lag = 30, Column = pca_3_warn\n",
      "Lag = 30, Column = pca_4_warn\n",
      "Lag = 30, Column = pca_5_warn\n",
      "Lag = 30, Column = pca_6_warn\n",
      "Lag = 30, Column = pca_7_warn\n",
      "Lag = 30, Column = pca_8_warn\n",
      "Lag = 30, Column = pca_9_warn\n",
      "Lag = 30, Column = pca_10_warn\n",
      "Lag = 30, Column = pca_11_warn\n",
      "Lag = 30, Column = pca_12_warn\n",
      "Lag = 30, Column = pca_13_warn\n",
      "Lag = 30, Column = pca_14_warn\n",
      "Lag = 30, Column = pca_15_warn\n",
      "Lag = 30, Column = pca_16_warn\n",
      "Lag = 30, Column = pca_17_warn\n",
      "Lag = 30, Column = pca_18_warn\n",
      "Lag = 30, Column = pca_19_warn\n",
      "Lag = 30, Column = pca_20_warn\n",
      "Lag = 30, Column = problem_type_1\n",
      "Lag = 30, Column = problem_type_2\n",
      "Lag = 30, Column = problem_type_3\n",
      "Lag = 30, Column = problem_type_4\n",
      "Lag = 30, Column = problem_type_1_per_usage1\n",
      "Lag = 30, Column = problem_type_2_per_usage1\n",
      "Lag = 30, Column = problem_type_3_per_usage1\n",
      "Lag = 30, Column = problem_type_4_per_usage1\n",
      "Lag = 30, Column = problem_type_1_per_usage2\n",
      "Lag = 30, Column = problem_type_2_per_usage2\n",
      "Lag = 30, Column = problem_type_3_per_usage2\n",
      "Lag = 30, Column = problem_type_4_per_usage2\n",
      "Lag = 30, Column = fault_code_type_1_count\n",
      "Lag = 30, Column = fault_code_type_2_count\n",
      "Lag = 30, Column = fault_code_type_3_count\n",
      "Lag = 30, Column = fault_code_type_4_count\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 90, Column = warn_type1_total\n",
      "Lag = 90, Column = warn_type2_total\n",
      "Lag = 90, Column = pca_1_warn\n",
      "Lag = 90, Column = pca_2_warn\n",
      "Lag = 90, Column = pca_3_warn\n",
      "Lag = 90, Column = pca_4_warn\n",
      "Lag = 90, Column = pca_5_warn\n",
      "Lag = 90, Column = pca_6_warn\n",
      "Lag = 90, Column = pca_7_warn\n",
      "Lag = 90, Column = pca_8_warn\n",
      "Lag = 90, Column = pca_9_warn\n",
      "Lag = 90, Column = pca_10_warn\n",
      "Lag = 90, Column = pca_11_warn\n",
      "Lag = 90, Column = pca_12_warn\n",
      "Lag = 90, Column = pca_13_warn\n",
      "Lag = 90, Column = pca_14_warn\n",
      "Lag = 90, Column = pca_15_warn\n",
      "Lag = 90, Column = pca_16_warn\n",
      "Lag = 90, Column = pca_17_warn\n",
      "Lag = 90, Column = pca_18_warn\n",
      "Lag = 90, Column = pca_19_warn\n",
      "Lag = 90, Column = pca_20_warn\n",
      "Lag = 90, Column = problem_type_1\n",
      "Lag = 90, Column = problem_type_2\n",
      "Lag = 90, Column = problem_type_3\n",
      "Lag = 90, Column = problem_type_4\n",
      "Lag = 90, Column = problem_type_1_per_usage1\n",
      "Lag = 90, Column = problem_type_2_per_usage1\n",
      "Lag = 90, Column = problem_type_3_per_usage1\n",
      "Lag = 90, Column = problem_type_4_per_usage1\n",
      "Lag = 90, Column = problem_type_1_per_usage2\n",
      "Lag = 90, Column = problem_type_2_per_usage2\n",
      "Lag = 90, Column = problem_type_3_per_usage2\n",
      "Lag = 90, Column = problem_type_4_per_usage2\n",
      "Lag = 90, Column = fault_code_type_1_count\n",
      "Lag = 90, Column = fault_code_type_2_count\n",
      "Lag = 90, Column = fault_code_type_3_count\n",
      "Lag = 90, Column = fault_code_type_4_count\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage2\n",
      "CPU times: user 1.03 s, sys: 411 ms, total: 1.44 s\n",
      "Wall time: 30min 16s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Load result dataset from Notebook #1\n",
    "df = sqlContext.read.parquet('/mnt/resource/PysparkExample/notebook1_result.parquet')\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('deviceid').orderBy('date').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        df = df.withColumn(col_name+'_rollingstd_'+str(lag_n), F.stddev(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))\n",
    "\n",
    "# There are some missing values for rollingstd features\n",
    "rollingstd_features = list(s for s in df.columns if \"rollingstd\" in s)\n",
    "df = df.fillna(0, subset=rollingstd_features)\n",
    "rollingstd = df.select(['key'] + list(s for s in df.columns if \"rollingstd\" in s))\n",
    "\n",
    "# Save the intermediate result for downstream work\n",
    "rollingstd.write.mode('overwrite').parquet('/mnt/resource/PysparkExample/rollingstd.parquet')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rolling Max"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Lag = 3, Column = warn_type1_total\n",
      "Lag = 3, Column = warn_type2_total\n",
      "Lag = 3, Column = pca_1_warn\n",
      "Lag = 3, Column = pca_2_warn\n",
      "Lag = 3, Column = pca_3_warn\n",
      "Lag = 3, Column = pca_4_warn\n",
      "Lag = 3, Column = pca_5_warn\n",
      "Lag = 3, Column = pca_6_warn\n",
      "Lag = 3, Column = pca_7_warn\n",
      "Lag = 3, Column = pca_8_warn\n",
      "Lag = 3, Column = pca_9_warn\n",
      "Lag = 3, Column = pca_10_warn\n",
      "Lag = 3, Column = pca_11_warn\n",
      "Lag = 3, Column = pca_12_warn\n",
      "Lag = 3, Column = pca_13_warn\n",
      "Lag = 3, Column = pca_14_warn\n",
      "Lag = 3, Column = pca_15_warn\n",
      "Lag = 3, Column = pca_16_warn\n",
      "Lag = 3, Column = pca_17_warn\n",
      "Lag = 3, Column = pca_18_warn\n",
      "Lag = 3, Column = pca_19_warn\n",
      "Lag = 3, Column = pca_20_warn\n",
      "Lag = 3, Column = problem_type_1\n",
      "Lag = 3, Column = problem_type_2\n",
      "Lag = 3, Column = problem_type_3\n",
      "Lag = 3, Column = problem_type_4\n",
      "Lag = 3, Column = problem_type_1_per_usage1\n",
      "Lag = 3, Column = problem_type_2_per_usage1\n",
      "Lag = 3, Column = problem_type_3_per_usage1\n",
      "Lag = 3, Column = problem_type_4_per_usage1\n",
      "Lag = 3, Column = problem_type_1_per_usage2\n",
      "Lag = 3, Column = problem_type_2_per_usage2\n",
      "Lag = 3, Column = problem_type_3_per_usage2\n",
      "Lag = 3, Column = problem_type_4_per_usage2\n",
      "Lag = 3, Column = fault_code_type_1_count\n",
      "Lag = 3, Column = fault_code_type_2_count\n",
      "Lag = 3, Column = fault_code_type_3_count\n",
      "Lag = 3, Column = fault_code_type_4_count\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 7, Column = warn_type1_total\n",
      "Lag = 7, Column = warn_type2_total\n",
      "Lag = 7, Column = pca_1_warn\n",
      "Lag = 7, Column = pca_2_warn\n",
      "Lag = 7, Column = pca_3_warn\n",
      "Lag = 7, Column = pca_4_warn\n",
      "Lag = 7, Column = pca_5_warn\n",
      "Lag = 7, Column = pca_6_warn\n",
      "Lag = 7, Column = pca_7_warn\n",
      "Lag = 7, Column = pca_8_warn\n",
      "Lag = 7, Column = pca_9_warn\n",
      "Lag = 7, Column = pca_10_warn\n",
      "Lag = 7, Column = pca_11_warn\n",
      "Lag = 7, Column = pca_12_warn\n",
      "Lag = 7, Column = pca_13_warn\n",
      "Lag = 7, Column = pca_14_warn\n",
      "Lag = 7, Column = pca_15_warn\n",
      "Lag = 7, Column = pca_16_warn\n",
      "Lag = 7, Column = pca_17_warn\n",
      "Lag = 7, Column = pca_18_warn\n",
      "Lag = 7, Column = pca_19_warn\n",
      "Lag = 7, Column = pca_20_warn\n",
      "Lag = 7, Column = problem_type_1\n",
      "Lag = 7, Column = problem_type_2\n",
      "Lag = 7, Column = problem_type_3\n",
      "Lag = 7, Column = problem_type_4\n",
      "Lag = 7, Column = problem_type_1_per_usage1\n",
      "Lag = 7, Column = problem_type_2_per_usage1\n",
      "Lag = 7, Column = problem_type_3_per_usage1\n",
      "Lag = 7, Column = problem_type_4_per_usage1\n",
      "Lag = 7, Column = problem_type_1_per_usage2\n",
      "Lag = 7, Column = problem_type_2_per_usage2\n",
      "Lag = 7, Column = problem_type_3_per_usage2\n",
      "Lag = 7, Column = problem_type_4_per_usage2\n",
      "Lag = 7, Column = fault_code_type_1_count\n",
      "Lag = 7, Column = fault_code_type_2_count\n",
      "Lag = 7, Column = fault_code_type_3_count\n",
      "Lag = 7, Column = fault_code_type_4_count\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 14, Column = warn_type1_total\n",
      "Lag = 14, Column = warn_type2_total\n",
      "Lag = 14, Column = pca_1_warn\n",
      "Lag = 14, Column = pca_2_warn\n",
      "Lag = 14, Column = pca_3_warn\n",
      "Lag = 14, Column = pca_4_warn\n",
      "Lag = 14, Column = pca_5_warn\n",
      "Lag = 14, Column = pca_6_warn\n",
      "Lag = 14, Column = pca_7_warn\n",
      "Lag = 14, Column = pca_8_warn\n",
      "Lag = 14, Column = pca_9_warn\n",
      "Lag = 14, Column = pca_10_warn\n",
      "Lag = 14, Column = pca_11_warn\n",
      "Lag = 14, Column = pca_12_warn\n",
      "Lag = 14, Column = pca_13_warn\n",
      "Lag = 14, Column = pca_14_warn\n",
      "Lag = 14, Column = pca_15_warn\n",
      "Lag = 14, Column = pca_16_warn\n",
      "Lag = 14, Column = pca_17_warn\n",
      "Lag = 14, Column = pca_18_warn\n",
      "Lag = 14, Column = pca_19_warn\n",
      "Lag = 14, Column = pca_20_warn\n",
      "Lag = 14, Column = problem_type_1\n",
      "Lag = 14, Column = problem_type_2\n",
      "Lag = 14, Column = problem_type_3\n",
      "Lag = 14, Column = problem_type_4\n",
      "Lag = 14, Column = problem_type_1_per_usage1\n",
      "Lag = 14, Column = problem_type_2_per_usage1\n",
      "Lag = 14, Column = problem_type_3_per_usage1\n",
      "Lag = 14, Column = problem_type_4_per_usage1\n",
      "Lag = 14, Column = problem_type_1_per_usage2\n",
      "Lag = 14, Column = problem_type_2_per_usage2\n",
      "Lag = 14, Column = problem_type_3_per_usage2\n",
      "Lag = 14, Column = problem_type_4_per_usage2\n",
      "Lag = 14, Column = fault_code_type_1_count\n",
      "Lag = 14, Column = fault_code_type_2_count\n",
      "Lag = 14, Column = fault_code_type_3_count\n",
      "Lag = 14, Column = fault_code_type_4_count\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 30, Column = warn_type1_total\n",
      "Lag = 30, Column = warn_type2_total\n",
      "Lag = 30, Column = pca_1_warn\n",
      "Lag = 30, Column = pca_2_warn\n",
      "Lag = 30, Column = pca_3_warn\n",
      "Lag = 30, Column = pca_4_warn\n",
      "Lag = 30, Column = pca_5_warn\n",
      "Lag = 30, Column = pca_6_warn\n",
      "Lag = 30, Column = pca_7_warn\n",
      "Lag = 30, Column = pca_8_warn\n",
      "Lag = 30, Column = pca_9_warn\n",
      "Lag = 30, Column = pca_10_warn\n",
      "Lag = 30, Column = pca_11_warn\n",
      "Lag = 30, Column = pca_12_warn\n",
      "Lag = 30, Column = pca_13_warn\n",
      "Lag = 30, Column = pca_14_warn\n",
      "Lag = 30, Column = pca_15_warn\n",
      "Lag = 30, Column = pca_16_warn\n",
      "Lag = 30, Column = pca_17_warn\n",
      "Lag = 30, Column = pca_18_warn\n",
      "Lag = 30, Column = pca_19_warn\n",
      "Lag = 30, Column = pca_20_warn\n",
      "Lag = 30, Column = problem_type_1\n",
      "Lag = 30, Column = problem_type_2\n",
      "Lag = 30, Column = problem_type_3\n",
      "Lag = 30, Column = problem_type_4\n",
      "Lag = 30, Column = problem_type_1_per_usage1\n",
      "Lag = 30, Column = problem_type_2_per_usage1\n",
      "Lag = 30, Column = problem_type_3_per_usage1\n",
      "Lag = 30, Column = problem_type_4_per_usage1\n",
      "Lag = 30, Column = problem_type_1_per_usage2\n",
      "Lag = 30, Column = problem_type_2_per_usage2\n",
      "Lag = 30, Column = problem_type_3_per_usage2\n",
      "Lag = 30, Column = problem_type_4_per_usage2\n",
      "Lag = 30, Column = fault_code_type_1_count\n",
      "Lag = 30, Column = fault_code_type_2_count\n",
      "Lag = 30, Column = fault_code_type_3_count\n",
      "Lag = 30, Column = fault_code_type_4_count\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 90, Column = warn_type1_total\n",
      "Lag = 90, Column = warn_type2_total\n",
      "Lag = 90, Column = pca_1_warn\n",
      "Lag = 90, Column = pca_2_warn\n",
      "Lag = 90, Column = pca_3_warn\n",
      "Lag = 90, Column = pca_4_warn\n",
      "Lag = 90, Column = pca_5_warn\n",
      "Lag = 90, Column = pca_6_warn\n",
      "Lag = 90, Column = pca_7_warn\n",
      "Lag = 90, Column = pca_8_warn\n",
      "Lag = 90, Column = pca_9_warn\n",
      "Lag = 90, Column = pca_10_warn\n",
      "Lag = 90, Column = pca_11_warn\n",
      "Lag = 90, Column = pca_12_warn\n",
      "Lag = 90, Column = pca_13_warn\n",
      "Lag = 90, Column = pca_14_warn\n",
      "Lag = 90, Column = pca_15_warn\n",
      "Lag = 90, Column = pca_16_warn\n",
      "Lag = 90, Column = pca_17_warn\n",
      "Lag = 90, Column = pca_18_warn\n",
      "Lag = 90, Column = pca_19_warn\n",
      "Lag = 90, Column = pca_20_warn\n",
      "Lag = 90, Column = problem_type_1\n",
      "Lag = 90, Column = problem_type_2\n",
      "Lag = 90, Column = problem_type_3\n",
      "Lag = 90, Column = problem_type_4\n",
      "Lag = 90, Column = problem_type_1_per_usage1\n",
      "Lag = 90, Column = problem_type_2_per_usage1\n",
      "Lag = 90, Column = problem_type_3_per_usage1\n",
      "Lag = 90, Column = problem_type_4_per_usage1\n",
      "Lag = 90, Column = problem_type_1_per_usage2\n",
      "Lag = 90, Column = problem_type_2_per_usage2\n",
      "Lag = 90, Column = problem_type_3_per_usage2\n",
      "Lag = 90, Column = problem_type_4_per_usage2\n",
      "Lag = 90, Column = fault_code_type_1_count\n",
      "Lag = 90, Column = fault_code_type_2_count\n",
      "Lag = 90, Column = fault_code_type_3_count\n",
      "Lag = 90, Column = fault_code_type_4_count\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage2\n",
      "CPU times: user 860 ms, sys: 316 ms, total: 1.18 s\n",
      "Wall time: 24min 41s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Load result dataset from Notebook #1\n",
    "df = sqlContext.read.parquet('/mnt/resource/PysparkExample/notebook1_result.parquet')\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('deviceid').orderBy('date').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        df = df.withColumn(col_name+'_rollingmax_'+str(lag_n), F.max(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))\n",
    "\n",
    "rollingmax = df.select(['key'] + list(s for s in df.columns if \"rollingmax\" in s))\n",
    "\n",
    "# Save the intermediate result for downstream work\n",
    "rollingmax.write.mode('overwrite').parquet('/mnt/resource/PysparkExample/rollingmax.parquet')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rolling Min"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Lag = 3, Column = warn_type1_total\n",
      "Lag = 3, Column = warn_type2_total\n",
      "Lag = 3, Column = pca_1_warn\n",
      "Lag = 3, Column = pca_2_warn\n",
      "Lag = 3, Column = pca_3_warn\n",
      "Lag = 3, Column = pca_4_warn\n",
      "Lag = 3, Column = pca_5_warn\n",
      "Lag = 3, Column = pca_6_warn\n",
      "Lag = 3, Column = pca_7_warn\n",
      "Lag = 3, Column = pca_8_warn\n",
      "Lag = 3, Column = pca_9_warn\n",
      "Lag = 3, Column = pca_10_warn\n",
      "Lag = 3, Column = pca_11_warn\n",
      "Lag = 3, Column = pca_12_warn\n",
      "Lag = 3, Column = pca_13_warn\n",
      "Lag = 3, Column = pca_14_warn\n",
      "Lag = 3, Column = pca_15_warn\n",
      "Lag = 3, Column = pca_16_warn\n",
      "Lag = 3, Column = pca_17_warn\n",
      "Lag = 3, Column = pca_18_warn\n",
      "Lag = 3, Column = pca_19_warn\n",
      "Lag = 3, Column = pca_20_warn\n",
      "Lag = 3, Column = problem_type_1\n",
      "Lag = 3, Column = problem_type_2\n",
      "Lag = 3, Column = problem_type_3\n",
      "Lag = 3, Column = problem_type_4\n",
      "Lag = 3, Column = problem_type_1_per_usage1\n",
      "Lag = 3, Column = problem_type_2_per_usage1\n",
      "Lag = 3, Column = problem_type_3_per_usage1\n",
      "Lag = 3, Column = problem_type_4_per_usage1\n",
      "Lag = 3, Column = problem_type_1_per_usage2\n",
      "Lag = 3, Column = problem_type_2_per_usage2\n",
      "Lag = 3, Column = problem_type_3_per_usage2\n",
      "Lag = 3, Column = problem_type_4_per_usage2\n",
      "Lag = 3, Column = fault_code_type_1_count\n",
      "Lag = 3, Column = fault_code_type_2_count\n",
      "Lag = 3, Column = fault_code_type_3_count\n",
      "Lag = 3, Column = fault_code_type_4_count\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 3, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 3, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 7, Column = warn_type1_total\n",
      "Lag = 7, Column = warn_type2_total\n",
      "Lag = 7, Column = pca_1_warn\n",
      "Lag = 7, Column = pca_2_warn\n",
      "Lag = 7, Column = pca_3_warn\n",
      "Lag = 7, Column = pca_4_warn\n",
      "Lag = 7, Column = pca_5_warn\n",
      "Lag = 7, Column = pca_6_warn\n",
      "Lag = 7, Column = pca_7_warn\n",
      "Lag = 7, Column = pca_8_warn\n",
      "Lag = 7, Column = pca_9_warn\n",
      "Lag = 7, Column = pca_10_warn\n",
      "Lag = 7, Column = pca_11_warn\n",
      "Lag = 7, Column = pca_12_warn\n",
      "Lag = 7, Column = pca_13_warn\n",
      "Lag = 7, Column = pca_14_warn\n",
      "Lag = 7, Column = pca_15_warn\n",
      "Lag = 7, Column = pca_16_warn\n",
      "Lag = 7, Column = pca_17_warn\n",
      "Lag = 7, Column = pca_18_warn\n",
      "Lag = 7, Column = pca_19_warn\n",
      "Lag = 7, Column = pca_20_warn\n",
      "Lag = 7, Column = problem_type_1\n",
      "Lag = 7, Column = problem_type_2\n",
      "Lag = 7, Column = problem_type_3\n",
      "Lag = 7, Column = problem_type_4\n",
      "Lag = 7, Column = problem_type_1_per_usage1\n",
      "Lag = 7, Column = problem_type_2_per_usage1\n",
      "Lag = 7, Column = problem_type_3_per_usage1\n",
      "Lag = 7, Column = problem_type_4_per_usage1\n",
      "Lag = 7, Column = problem_type_1_per_usage2\n",
      "Lag = 7, Column = problem_type_2_per_usage2\n",
      "Lag = 7, Column = problem_type_3_per_usage2\n",
      "Lag = 7, Column = problem_type_4_per_usage2\n",
      "Lag = 7, Column = fault_code_type_1_count\n",
      "Lag = 7, Column = fault_code_type_2_count\n",
      "Lag = 7, Column = fault_code_type_3_count\n",
      "Lag = 7, Column = fault_code_type_4_count\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 7, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 7, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 14, Column = warn_type1_total\n",
      "Lag = 14, Column = warn_type2_total\n",
      "Lag = 14, Column = pca_1_warn\n",
      "Lag = 14, Column = pca_2_warn\n",
      "Lag = 14, Column = pca_3_warn\n",
      "Lag = 14, Column = pca_4_warn\n",
      "Lag = 14, Column = pca_5_warn\n",
      "Lag = 14, Column = pca_6_warn\n",
      "Lag = 14, Column = pca_7_warn\n",
      "Lag = 14, Column = pca_8_warn\n",
      "Lag = 14, Column = pca_9_warn\n",
      "Lag = 14, Column = pca_10_warn\n",
      "Lag = 14, Column = pca_11_warn\n",
      "Lag = 14, Column = pca_12_warn\n",
      "Lag = 14, Column = pca_13_warn\n",
      "Lag = 14, Column = pca_14_warn\n",
      "Lag = 14, Column = pca_15_warn\n",
      "Lag = 14, Column = pca_16_warn\n",
      "Lag = 14, Column = pca_17_warn\n",
      "Lag = 14, Column = pca_18_warn\n",
      "Lag = 14, Column = pca_19_warn\n",
      "Lag = 14, Column = pca_20_warn\n",
      "Lag = 14, Column = problem_type_1\n",
      "Lag = 14, Column = problem_type_2\n",
      "Lag = 14, Column = problem_type_3\n",
      "Lag = 14, Column = problem_type_4\n",
      "Lag = 14, Column = problem_type_1_per_usage1\n",
      "Lag = 14, Column = problem_type_2_per_usage1\n",
      "Lag = 14, Column = problem_type_3_per_usage1\n",
      "Lag = 14, Column = problem_type_4_per_usage1\n",
      "Lag = 14, Column = problem_type_1_per_usage2\n",
      "Lag = 14, Column = problem_type_2_per_usage2\n",
      "Lag = 14, Column = problem_type_3_per_usage2\n",
      "Lag = 14, Column = problem_type_4_per_usage2\n",
      "Lag = 14, Column = fault_code_type_1_count\n",
      "Lag = 14, Column = fault_code_type_2_count\n",
      "Lag = 14, Column = fault_code_type_3_count\n",
      "Lag = 14, Column = fault_code_type_4_count\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 14, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 14, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 30, Column = warn_type1_total\n",
      "Lag = 30, Column = warn_type2_total\n",
      "Lag = 30, Column = pca_1_warn\n",
      "Lag = 30, Column = pca_2_warn\n",
      "Lag = 30, Column = pca_3_warn\n",
      "Lag = 30, Column = pca_4_warn\n",
      "Lag = 30, Column = pca_5_warn\n",
      "Lag = 30, Column = pca_6_warn\n",
      "Lag = 30, Column = pca_7_warn\n",
      "Lag = 30, Column = pca_8_warn\n",
      "Lag = 30, Column = pca_9_warn\n",
      "Lag = 30, Column = pca_10_warn\n",
      "Lag = 30, Column = pca_11_warn\n",
      "Lag = 30, Column = pca_12_warn\n",
      "Lag = 30, Column = pca_13_warn\n",
      "Lag = 30, Column = pca_14_warn\n",
      "Lag = 30, Column = pca_15_warn\n",
      "Lag = 30, Column = pca_16_warn\n",
      "Lag = 30, Column = pca_17_warn\n",
      "Lag = 30, Column = pca_18_warn\n",
      "Lag = 30, Column = pca_19_warn\n",
      "Lag = 30, Column = pca_20_warn\n",
      "Lag = 30, Column = problem_type_1\n",
      "Lag = 30, Column = problem_type_2\n",
      "Lag = 30, Column = problem_type_3\n",
      "Lag = 30, Column = problem_type_4\n",
      "Lag = 30, Column = problem_type_1_per_usage1\n",
      "Lag = 30, Column = problem_type_2_per_usage1\n",
      "Lag = 30, Column = problem_type_3_per_usage1\n",
      "Lag = 30, Column = problem_type_4_per_usage1\n",
      "Lag = 30, Column = problem_type_1_per_usage2\n",
      "Lag = 30, Column = problem_type_2_per_usage2\n",
      "Lag = 30, Column = problem_type_3_per_usage2\n",
      "Lag = 30, Column = problem_type_4_per_usage2\n",
      "Lag = 30, Column = fault_code_type_1_count\n",
      "Lag = 30, Column = fault_code_type_2_count\n",
      "Lag = 30, Column = fault_code_type_3_count\n",
      "Lag = 30, Column = fault_code_type_4_count\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 30, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 30, Column = fault_code_type_4_count_per_usage2\n",
      "Lag = 90, Column = warn_type1_total\n",
      "Lag = 90, Column = warn_type2_total\n",
      "Lag = 90, Column = pca_1_warn\n",
      "Lag = 90, Column = pca_2_warn\n",
      "Lag = 90, Column = pca_3_warn\n",
      "Lag = 90, Column = pca_4_warn\n",
      "Lag = 90, Column = pca_5_warn\n",
      "Lag = 90, Column = pca_6_warn\n",
      "Lag = 90, Column = pca_7_warn\n",
      "Lag = 90, Column = pca_8_warn\n",
      "Lag = 90, Column = pca_9_warn\n",
      "Lag = 90, Column = pca_10_warn\n",
      "Lag = 90, Column = pca_11_warn\n",
      "Lag = 90, Column = pca_12_warn\n",
      "Lag = 90, Column = pca_13_warn\n",
      "Lag = 90, Column = pca_14_warn\n",
      "Lag = 90, Column = pca_15_warn\n",
      "Lag = 90, Column = pca_16_warn\n",
      "Lag = 90, Column = pca_17_warn\n",
      "Lag = 90, Column = pca_18_warn\n",
      "Lag = 90, Column = pca_19_warn\n",
      "Lag = 90, Column = pca_20_warn\n",
      "Lag = 90, Column = problem_type_1\n",
      "Lag = 90, Column = problem_type_2\n",
      "Lag = 90, Column = problem_type_3\n",
      "Lag = 90, Column = problem_type_4\n",
      "Lag = 90, Column = problem_type_1_per_usage1\n",
      "Lag = 90, Column = problem_type_2_per_usage1\n",
      "Lag = 90, Column = problem_type_3_per_usage1\n",
      "Lag = 90, Column = problem_type_4_per_usage1\n",
      "Lag = 90, Column = problem_type_1_per_usage2\n",
      "Lag = 90, Column = problem_type_2_per_usage2\n",
      "Lag = 90, Column = problem_type_3_per_usage2\n",
      "Lag = 90, Column = problem_type_4_per_usage2\n",
      "Lag = 90, Column = fault_code_type_1_count\n",
      "Lag = 90, Column = fault_code_type_2_count\n",
      "Lag = 90, Column = fault_code_type_3_count\n",
      "Lag = 90, Column = fault_code_type_4_count\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage1\n",
      "Lag = 90, Column = fault_code_type_1_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_2_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_3_count_per_usage2\n",
      "Lag = 90, Column = fault_code_type_4_count_per_usage2\n",
      "CPU times: user 870 ms, sys: 306 ms, total: 1.18 s\n",
      "Wall time: 23min 27s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Load result dataset from Notebook #1\n",
    "df = sqlContext.read.parquet('/mnt/resource/PysparkExample/notebook1_result.parquet')\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('deviceid').orderBy('date').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        df = df.withColumn(col_name+'_rollingmin_'+str(lag_n), F.min(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))\n",
    "\n",
    "rollingmin = df.select(['key'] + list(s for s in df.columns if \"rollingmin\" in s))\n",
    "\n",
    "# Save the intermediate result for downstream work\n",
    "rollingmin.write.mode('overwrite').parquet('/mnt/resource/PysparkExample/rollingmin.parquet')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Join result dataset from the five rolling compute cells:\n",
    "-  Join in Spark is usually very slow, it is better to reduce the number of partitions before the join.\n",
    "-  Check the number of partitions of the pyspark dataframe.\n",
    "-  **repartition vs coalesce**. If we only want to reduce the number of partitions, it is better to use coalesce because repartition involves reshuffling which is computational more expensive and takes more time.\n",
    "<br>\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "33\n",
      "33\n",
      "33\n",
      "31\n",
      "31\n"
     ]
    }
   ],
   "source": [
    "# Import result dataset \n",
    "rollingmean = sqlContext.read.parquet('/mnt/resource/PysparkExample/data_rollingmean.parquet')\n",
    "rollingdiff = sqlContext.read.parquet('/mnt/resource/PysparkExample/rollingdiff.parquet')\n",
    "rollingstd = sqlContext.read.parquet('/mnt/resource/PysparkExample/rollingstd.parquet')\n",
    "rollingmax = sqlContext.read.parquet('/mnt/resource/PysparkExample/rollingmax.parquet')\n",
    "rollingmin = sqlContext.read.parquet('/mnt/resource/PysparkExample/rollingmin.parquet')\n",
    "\n",
    "# Check the number of partitions for each dataset\n",
    "print(rollingmean.rdd.getNumPartitions())\n",
    "print(rollingdiff.rdd.getNumPartitions())\n",
    "print(rollingstd.rdd.getNumPartitions())\n",
    "print(rollingmax.rdd.getNumPartitions())\n",
    "print(rollingmin.rdd.getNumPartitions())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 901 ms, sys: 303 ms, total: 1.2 s\n",
      "Wall time: 1h 50min 38s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# To make join faster, reduce the number of partitions (not necessarily to \"1\")\n",
    "rollingmean = rollingmean.coalesce(1)\n",
    "rollingdiff = rollingdiff.coalesce(1)\n",
    "rollingstd = rollingstd.coalesce(1)\n",
    "rollingmax = rollingmax.coalesce(1)\n",
    "rollingmin = rollingmin.coalesce(1)\n",
    "\n",
    "rolling_result = rollingmean.join(rollingdiff, 'key', 'inner')\\\n",
    "                 .join(rollingstd, 'key', 'inner')\\\n",
    "                 .join(rollingmax, 'key', 'inner')\\\n",
    "                 .join(rollingmin, 'key', 'inner')\n",
    "            \n",
    "\n",
    "## Write the final result as parquet file for downstream work in Notebook_3\n",
    "rolling_result.write.mode('overwrite').parquet('/mnt/resource/PysparkExample/notebook2_result.parquet')\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Spark - python",
   "language": "python",
   "name": "spark-python"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}