<commit_before>{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Step 2: Feature Engineering\n",
    "\n",
    "Feature engineering combines the different data sources together to create a single data set of features (variables) that can be used to infer a machines's health condition over time. \n",
    "\n",
    "In this notebook, we will load the data stored in Azure Blob containers in the previous **Data Ingestion** notebook (`Code/data_ingestion.ipynb`). The note book uses several feature engineering methods to create a data set for use in our predictive maintenance machine learning solution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# import the libraries\n",
    "import os\n",
    "\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.functions import col, unix_timestamp, round\n",
    "from pyspark.sql.functions import datediff\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.types import DoubleType\n",
    "\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import OneHotEncoder\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "from azure.storage.blob import BlockBlobService\n",
    "from azure.storage.blob import PublicAccess\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load data from Azure Blob storage container\n",
    "\n",
    "We have previously downloaded and stored the following data in an Azure blob storage container:\n",
    "\n",
    "  * Machines: Features differentiating each machine. For example age and model.\n",
    "  * Error: The log of non-critical errors. These errors may still indicate an impending component failure.\n",
    "  * Maint: Machine maintenance history detailing component replacement or regular maintenance activities withe the date of replacement.\n",
    "  * Telemetry: The operating conditions of a machine e.g. data collected from sensors.\n",
    "  * Failure history: The failure history of a machine or component within the machine.\n",
    "\n",
    "We'll load these files from blob, and create our analysis data set here. We'll write this data set back into a new blob container to use in our model building and evaluation notebook later. \n",
    "\n",
    "Since the Azure Blob storage account name and account key are not passed between notebooks, you'll need to provide those here again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Enter your Azure blob storage details here \n",
    "ACCOUNT_NAME = \"pdmamlworkbench\"   ## \"<your blob storage account name>\"\n",
    "\n",
    "# You can find the account key under the _Access Keys_ link in the \n",
    "# [Azure Portal](portal.azure.com) page for your Azure storage container.\n",
    "ACCOUNT_KEY = \"<KEY> ## \"<account key>\"\n",
    "\n",
    "#-------------------------------------------------------------------------------------------\n",
    "# The data from the Data Aquisition note book is stored in the dataingestion container.\n",
    "CONTAINER_NAME = \"dataingestion\"\n",
    "\n",
    "# Connect to your blob service     \n",
    "my_service = BlockBlobService(account_name=ACCOUNT_NAME, account_key=ACCOUNT_KEY)\n",
    "\n",
    "\n",
    "# We will store each of these data sets in blob storage in an \n",
    "# Azure Storage Container on your Azure subscription.\n",
    "# See https://github.com/Azure/ViennaDocs/blob/master/Documentation/UsingBlobForStorage.md\n",
    "# for details.\n",
    "\n",
    "# These file names detail which blob each files is stored under. \n",
    "MACH_DATA = 'machines_files.parquet'\n",
    "MAINT_DATA = 'maint_files.parquet'\n",
    "ERROR_DATA = 'errors_files.parquet'\n",
    "TELEMETRY_DATA = 'telemetry_files.parquet'\n",
    "FAILURE_DATA = 'failure_files.parquet'\n",
    "\n",
    "# create a local path where to store the results later.\n",
    "MACH_LOCAL_DIRECT = 'dataingestion_mach_result.parquet'\n",
    "ERROR_LOCAL_DIRECT = 'dataingestion_err_result.parquet'\n",
    "MAINT_LOCAL_DIRECT = 'dataingestion_maint_result.parquet'\n",
    "TELEMETRY_LOCAL_DIRECT = 'dataingestion_tel_result.parquet'\n",
    "FAILURES_LOCAL_DIRECT = 'dataingestion_fail_result.parquet'\n",
    "\n",
    "# define your blob service     \n",
    "my_service = BlockBlobService(account_name=ACCOUNT_NAME, account_key=ACCOUNT_KEY)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Machines data set\n",
    "\n",
    "Load the machines data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-3-05b91dc7eb0b>, line 6)",
     "output_type": "error",
     "traceback": [
      "\u001b[0;36m  File \u001b[0;32m\"<ipython-input-3-05b91dc7eb0b>\"\u001b[0;36m, line \u001b[0;32m6\u001b[0m\n\u001b[0;31m    print('DONE creating a local directory!')\u001b[0m\n\u001b[0m        ^\u001b[0m\n\u001b[0;31mSyntaxError\u001b[0m\u001b[0;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "# download the entire parquet result folder to local path for a new run \n",
    "\n",
    "# create a local path  to store the data.\n",
    "if not os.path.exists(MACH_LOCAL_DIRECT):\n",
    "    os.makedirs(MACH_LOCAL_DIRECT\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "# Connect to blob storage container\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if MACH_DATA in blob.name:\n",
    "        local_file = os.path.join(MACH_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "# Read in the data\n",
    "machines = spark.read.parquet(MACH_LOCAL_DIRECT)\n",
    "\n",
    "print(machines.count())\n",
    "machines.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Errors data set\n",
    "\n",
    "Load the errors data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load the previous created final dataset into the workspace\n",
    "\n",
    "# create a local path  to store the data.\n",
    "if not os.path.exists(ERROR_LOCAL_DIRECT):\n",
    "    os.makedirs(ERROR_LOCAL_DIRECT)\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "# Connect to blob storage container\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if ERROR_DATA in blob.name:\n",
    "        local_file = os.path.join(ERROR_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "# Read in the data\n",
    "errors = spark.read.parquet(ERROR_LOCAL_DIRECT)\n",
    "\n",
    "print(errors.count())\n",
    "errors.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Maintenance data set\n",
    "\n",
    "Load the maintenance data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load the previous created final dataset into the workspace\n",
    "\n",
    "# create a local path  to store the data.\n",
    " if not os.path.exists(MAINT_LOCAL_DIRECT):\n",
    "    os.makedirs(MAINT_LOCAL_DIRECT)\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "# Connect to blob storage container\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if MAINT_DATA in blob.name:\n",
    "        local_file = os.path.join(MAINT_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "# Read in the data\n",
    "maint = spark.read.parquet(MAINT_LOCAL_DIRECT)\n",
    "\n",
    "print(maint.count())\n",
    "maint.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Telemetry\n",
    "\n",
    "Load the telemetry data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load the previous created final dataset into the workspace\n",
    "\n",
    "# create a local path  to store the data.\n",
    "if not os.path.exists(TELEMETRY_LOCAL_DIRECT):\n",
    "    os.makedirs(TELEMETRY_LOCAL_DIRECT)\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "# Connect to blob storage container\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if TELEMETRY_DATA in blob.name:\n",
    "        local_file = os.path.join(TELEMETRY_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "# Read in the data\n",
    "telemetry = spark.read.parquet(TELEMETRY_LOCAL_DIRECT)\n",
    "\n",
    "print(telemetry.count())\n",
    "telemetry.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Failures data set\n",
    "\n",
    "Load the failures data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load the previous created final dataset into the workspace\n",
    "\n",
    "# create a local path  to store the data.\n",
    "\n",
    "if not os.path.exists(FAILURES_LOCAL_DIRECT):\n",
    "    os.makedirs(FAILURES_LOCAL_DIRECT)\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "\n",
    "# download the entire parquet result folder to local path for a new run \n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if FAILURE_DATA in blob.name:\n",
    "        local_file = os.path.join(FAILURES_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "failures = spark.read.parquet(FAILURES_LOCAL_DIRECT)\n",
    "\n",
    "print(failures.count())\n",
    "failures.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature engineering \n",
    "\n",
    "Feature engineering combines the different data sources together to create a single data set of features (variables) that can be used to infer a machines's health condition over time. The ultimate goal is to generate a single record for each time unit for each asset combining its features and labels to be fed into the machine learning algorithm. In order to prepare that clean final data set, some pre-processing steps should be taken. First step is to divide the duration of data collection into time units where each record belongs to a time unit for an asset.\n",
    "\n",
    "The measurement unit for time can be in seconds, minutes, hours, days, months, cycles, miles or transactions depending on the efficiency of data preparation and the changes observed in the conditions of the asset from a time unit to the other or other factors specific to the domain. In other words, the time unit does not have to be the same as the frequency of data collection as in many cases data may not show any difference from one unit to the other. For example, if temperature values were being collected every 10 seconds, picking a time unit of 10 seconds for the whole analysis inflates the number of examples without providing any additional information. Better strategy would be to use average over an hour as an example.\n",
    "\n",
    "### Rolling aggregates\n",
    "\n",
    "For each record of an asset, we pick a rolling window of size \"W\" which is the number of units of time that we would like to compute historical aggregates for. We then compute rolling aggregate features using the W periods before the date of that record. Some example rolling aggregates can be rolling counts, means, standard deviations, outliers based on standard deviations, CUSUM measures, minimum and maximum values for the window. Another interesting technique is to capture trend changes, spikes and level changes using algorithms that detect anomalies in data using anomaly detection algorithms.\n",
    "\n",
    "### Lag features\n",
    "As mentioned earlier, in predictive maintenance, historical data usually comes with timestamps indicating the time of collection for each piece of data. There are many ways of creating features from the data that comes with timestamped data. In this section, we discuss some of these methods used for predictive maintenance. However, we are not limited by these methods alone. Since feature engineering is considered to be one of the most creative areas of predictive modeling, there could be many other ways to create features. Here, we provide some general techniques."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Telemetry features\n",
    "\n",
    "Because the telemetry data set is the largest time series data we have, we start feature engineering here. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# rolling mean\n",
    "# Temporary storage for rolling means\n",
    "tel_mean = telemetry\n",
    "\n",
    "# Which features are we interested in telemetry data set\n",
    "rolling_features = ['volt','rotate', 'pressure', 'vibration']\n",
    "               \n",
    "# We choose two windows for our rolling windows 3hrs, 24 hrs\n",
    "lags = [3,24]\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('machineID').orderBy('datetime').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        tel_mean = tel_mean.withColumn(col_name+'_rollingmean_'+str(lag_n), F.avg(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We repeat this rolling window process for the standard deviation. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# rolling standard deviation\n",
    "# Temporary storage for rolling means\n",
    "tel_sd = telemetry\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('machineID').orderBy('datetime').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        tel_sd = tel_sd.withColumn(col_name+'_rollingstd_'+str(lag_n), F.stddev(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resample every 3 hours"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# tel_mean rolling mean\n",
    "# 3 hours = 10800 seconds  \n",
    "time_val = 3 * 60 * 60\n",
    "\n",
    "# I think this is grabbing datetime from the tel_sd data set, which is equivalent to telemetry\n",
    "dt_truncated = ((round(unix_timestamp(col(\"datetime\")) / time_val) * time_val).cast(\"timestamp\"))\n",
    "\n",
    "tel_mean_resampled = tel_mean.withColumn(\"dt_truncated\", dt_truncated).drop('volt', 'rotate', 'pressure', 'vibration')\n",
    "\n",
    "tel_mean_resampled1 = (tel_mean_resampled.groupBy(\"machineID\",\"dt_truncated\")\n",
    "                               .agg(F.mean('volt_rollingmean_3').alias('volt_rollingmean_3'),\n",
    "                                    F.mean('rotate_rollingmean_3').alias('rotate_rollingmean_3'), \n",
    "                                    F.mean('pressure_rollingmean_3').alias('pressure_rollingmean_3'), \n",
    "                                    F.mean('vibration_rollingmean_3').alias('vibration_rollingmean_3'), \n",
    "                                    F.mean('volt_rollingmean_24').alias('volt_rollingmean_24'),\n",
    "                                    F.mean('rotate_rollingmean_24').alias('rotate_rollingmean_24'), \n",
    "                                    F.mean('pressure_rollingmean_24').alias('pressure_rollingmean_24'), \n",
    "                                    F.mean('vibration_rollingmean_24').alias('vibration_rollingmean_24')))\n",
    "\n",
    "tel_mean_resampled1.where((col(\"machineID\") == 1)).toPandas().head(5)\n",
    "tel_mean_resampled1.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# tel_sd rolling sd\n",
    "tel_sd_resampled = (tel_sd.withColumn(\"dt_truncated\", dt_truncated).drop('volt', 'rotate', 'pressure', 'vibration')\n",
    "                        .fillna(0))\n",
    "\n",
    "tel_sd_resampled1 = (tel_sd_resampled.groupBy(\"machineID\",\"dt_truncated\")\n",
    "                               .agg(F.sd('volt_rollingstd_3').alias('volt_rollingstd_3'),\n",
    "                                    F.sd('rotate_rollingstd_3').alias('rotate_rollingstd_3'), \n",
    "                                    F.sd('pressure_rollingstd_3').alias('pressure_rollingstd_3'), \n",
    "                                    F.sd('vibration_rollingstd_3').alias('vibration_rollingstd_3'), \n",
    "                                    F.sd('volt_rollingstd_24').alias('volt_rollingstd_24'),\n",
    "                                    F.sd('rotate_rollingstd_24').alias('rotate_rollingstd_24'), \n",
    "                                    F.sd('pressure_rollingstd_24').alias('pressure_rollingstd_24'), \n",
    "                                    F.sd('vibration_rollingstd_24').alias('vibration_rollingstd_24')))\n",
    "tel_sd_resampled1.show(5)\n",
    "tel_sd_resampled1.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lag features from Errors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a column for each errorID \n",
    "error1 = errors.groupBy(\"machineID\",\"datetime\",\"errorID\").pivot('errorID').agg(F.count('machineID').alias('dummy'))\n",
    "\n",
    "# remove the column called errorID and fill in missing values\n",
    "error2 = error1.drop('errorID').fillna(0)\n",
    "\n",
    "# combine errors for a given machine in a given hour\n",
    "error3 = (error2.groupBy(\"machineID\",\"datetime\")\n",
    "                .agg(F.sum('error1').alias('error1sum'), \n",
    "                     F.sum('error2').alias('error2sum'), \n",
    "                     F.sum('error3').alias('error3sum'), \n",
    "                     F.sum('error4').alias('error4sum'), \n",
    "                     F.sum('error5').alias('error5sum')))\n",
    "\n",
    "error3.toPandas().head(5)\n",
    "error3.count(), len(error3.columns)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We want the same number of rows in error as in telemetry.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# join the telemetry data with errors\n",
    "error_count = (telemetry.join(error3, ((telemetry['machineID'] == error3['machineID']) \n",
    "                                  & (telemetry['datetime'] == error3['datetime'])), \"left\")\n",
    "               .drop('volt', 'rotate', 'pressure', 'vibration').drop(error3.machineID).drop(error3.datetime))\n",
    "\n",
    "# fill in missing value\n",
    "error_count1 = error_count.fillna(0)\n",
    "\n",
    "error_count1.toPandas().show(5)\n",
    "error_count1.count(), len(error_count1.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rolling_features1 = ['error1sum','error2sum', 'error3sum', 'error4sum', 'error5sum']\n",
    "               \n",
    "# lag window 24 hrs\n",
    "lags = [24]\n",
    "\n",
    "# rolling mean\n",
    "err_mean = error_count1\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('machineID').orderBy('datetime').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features1:\n",
    "        err_mean = err_mean.withColumn(col_name+'_rollingmean_'+str(lag_n), F.avg(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resample to every 3 hours"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dt_truncated = ((round(unix_timestamp(col(\"datetime\")) / time_val) * time_val)\n",
    "    .cast(\"timestamp\"))\n",
    "\n",
    "err_mean_resampled = (err_mean.withColumn(\"dt_truncated\", dt_truncated)\n",
    "                    .drop('error1sum', 'error2sum', 'error3sum', 'error4sum', 'error5sum').fillna(0))\n",
    "\n",
    "err_mean_resampled1 = (err_mean_resampled.groupBy(\"machineID\",\"dt_truncated\")\n",
    "                               .agg(F.mean('error1sum_rollingmean_24').alias('error1sum_rollingmean_24'), \n",
    "                                    F.mean('error2sum_rollingmean_24').alias('error2sum_rollingmean_24'), \n",
    "                                    F.mean('error3sum_rollingmean_24').alias('error3sum_rollingmean_24'), \n",
    "                                    F.mean('error4sum_rollingmean_24').alias('error4sum_rollingmean_24'), \n",
    "                                    F.mean('error5sum_rollingmean_24').alias('error5sum_rollingmean_24')))\n",
    "err_mean_resampled1.toPandas().head(5)\n",
    "err_mean_resampled1.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement from maintenance "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a column for each comp \n",
    "maint1 = maint.groupBy(\"machineID\",\"datetime\",\"comp\").pivot('comp').agg(F.count('machineID').alias('dummy'))\n",
    "\n",
    "# remove the column called comp and fill in missing values\n",
    "maint2 = maint1.drop('comp').fillna(0)\n",
    "\n",
    "# combine maintenance for a given machine in a given hour\n",
    "maint3 = (maint2.groupBy(\"machineID\",\"datetime\").agg(F.sum('comp1').alias('comp1sum'), \n",
    "                                                    F.sum('comp2').alias('comp2sum'), \n",
    "                                                    F.sum('comp3').alias('comp3sum'),\n",
    "                                                    F.sum('comp4').alias('comp4sum')))\n",
    "maint3.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement for component-1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_maint_comp1 = (maint3.where((col(\"comp1sum\") == '1')).withColumnRenamed('datetime','datetime_maint')\n",
    "                           .drop('comp2sum', 'comp3sum', 'comp4sum'))\n",
    "\n",
    "test_tel_comp1 = (telemetry.withColumnRenamed('datetime','datetime_tel')\n",
    "                  .drop(telemetry.volt).drop(telemetry.rotate).drop(telemetry.pressure).drop(telemetry.vibration))\n",
    "\n",
    "test_maint_tel_comp1 = test_tel_comp1.join(test_maint_comp1, ((test_tel_comp1['machineID']==\n",
    "                                                               test_maint_comp1['machineID']) \n",
    "                                            & (test_tel_comp1['datetime_tel'] > test_maint_comp1['datetime_maint']) \n",
    "                                            & (test_maint_comp1['comp1sum'] == '1'))).drop(test_maint_comp1.machineID)\n",
    "\n",
    "comp1 = (test_maint_tel_comp1.withColumn(\"sincelastcomp1\", \n",
    "              datediff(test_maint_tel_comp1.datetime_tel, test_maint_tel_comp1.datetime_maint))\n",
    "              .drop(test_maint_tel_comp1.datetime_maint).drop(test_maint_tel_comp1.comp1sum))\n",
    "\n",
    "comp1.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement for component-2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_maint_comp2 = (maint3.where(col(\"comp2sum\") == '1').withColumnRenamed('datetime','datetime_maint')\n",
    "                         .drop('comp1sum', 'comp3sum', 'comp4sum'))\n",
    "\n",
    "test_tel_comp2 = (telemetry.withColumnRenamed('datetime','datetime_tel')\n",
    "                          .drop(telemetry.volt).drop(telemetry.rotate).drop(telemetry.pressure)\n",
    "                          .drop(telemetry.vibration))\n",
    "\n",
    "test_maint_tel_comp2 = (test_tel_comp2.join(test_maint_comp2, ((test_tel_comp2['machineID']==\n",
    "                                                                test_maint_comp2['machineID']) \n",
    "                                        & (test_tel_comp2['datetime_tel'] > test_maint_comp2['datetime_maint']) \n",
    "                                        & (test_maint_comp2['comp2sum'] == '1') \n",
    "                                           )).drop(test_maint_comp2.machineID))\n",
    "\n",
    "comp2 = (test_maint_tel_comp2.withColumn(\"sincelastcomp2\", \n",
    "              datediff(test_maint_tel_comp2.datetime_tel, test_maint_tel_comp2.datetime_maint))\n",
    "              .drop(test_maint_tel_comp2.datetime_maint).drop(test_maint_tel_comp2.comp2sum))\n",
    "comp2.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement for component-3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_maint_comp3 = (maint3.where(col(\"comp3sum\") == '1').withColumnRenamed('datetime','datetime_maint')\n",
    "                          .drop('comp1sum', 'comp2sum', 'comp4sum'))\n",
    "\n",
    "test_tel_comp3 = (telemetry.withColumnRenamed('datetime','datetime_tel')\n",
    "                    .drop(telemetry.volt).drop(telemetry.rotate).drop(telemetry.pressure).drop(telemetry.vibration))\n",
    "\n",
    "test_maint_tel_comp3 = test_tel_comp3.join(test_maint_comp3, ((test_tel_comp3['machineID']==\n",
    "                                                               test_maint_comp3['machineID']) \n",
    "                                        & (test_tel_comp3['datetime_tel'] > test_maint_comp3['datetime_maint']) \n",
    "                                        & (test_maint_comp3['comp3sum'] == '1') \n",
    "                                           )).drop(test_maint_comp3.machineID)\n",
    "\n",
    "comp3 = (test_maint_tel_comp3.withColumn(\"sincelastcomp3\", \n",
    "              datediff(test_maint_tel_comp3.datetime_tel, test_maint_tel_comp3.datetime_maint))\n",
    "              .drop(test_maint_tel_comp3.datetime_maint).drop(test_maint_tel_comp3.comp3sum))\n",
    "comp3.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement for component-4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_maint_comp4 = (maint3.where(col(\"comp4sum\") == '1').withColumnRenamed('datetime','datetime_maint')\n",
    "                         .drop('comp1sum', 'comp2sum', 'comp3sum'))\n",
    "\n",
    "test_tel_comp4 = (telemetry.withColumnRenamed('datetime','datetime_tel')\n",
    "                  .drop(telemetry.volt).drop(telemetry.rotate).drop(telemetry.pressure).drop(telemetry.vibration))\n",
    "\n",
    "test_maint_tel_comp4 = test_tel_comp4.join(test_maint_comp4, ((test_tel_comp4['machineID']==\n",
    "                                                               test_maint_comp4['machineID']) \n",
    "                                        & (test_tel_comp4['datetime_tel'] > test_maint_comp4['datetime_maint']) \n",
    "                                        & (test_maint_comp4['comp4sum'] == '1'))).drop(test_maint_comp4.machineID)\n",
    "\n",
    "comp4 = (test_maint_tel_comp4.withColumn(\"sincelastcomp4\", \n",
    "              datediff(test_maint_tel_comp4.datetime_tel, test_maint_tel_comp4.datetime_maint))\n",
    "              .drop(test_maint_tel_comp4.datetime_maint).drop(test_maint_tel_comp4.comp4sum))\n",
    "comp4.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##  Combine comp1, comp2, comp3, comp4 to generate the maintenance feature set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# left join comp1 with (comp2, comp3, comp4) \n",
    "# left join comp2 with (comp3, comp4) \n",
    "# left join comp3, comp4 \n",
    "comp3_4 = (comp3.join(comp4, ((comp3['machineID'] == comp4['machineID']) \n",
    "                                  & (comp3['datetime_tel'] == comp4['datetime_tel'])), \"left\")\n",
    "                                  .drop(comp4.machineID).drop(comp4.datetime_tel))comp2_3_4 = (comp2.join(comp3_4, ((comp2['machineID'] == comp3_4['machineID']) \n",
    "                                  & (comp2['datetime_tel'] == comp3_4['datetime_tel'])), \"left\")\n",
    "                                  .drop(comp3_4.machineID).drop(comp3_4.datetime_tel))\n",
    "comp1_2_3_4 = (comp1.join(comp2_3_4, ((comp1['machineID'] == comp2_3_4['machineID']) \n",
    "                                  & (comp1['datetime_tel'] == comp2_3_4['datetime_tel'])), \"left\")\n",
    "                                 .drop(comp2_3_4.machineID).drop(comp2_3_4.datetime_tel))\n",
    "comp1_2_3_4_final = (comp1_2_3_4.groupBy(\"machineID\", \"datetime_tel\")\n",
    "                                .agg(F.max('sincelastcomp1').alias('sincelastcomp1'), \n",
    "                                     F.max('sincelastcomp2').alias('sincelastcomp2'), \n",
    "                                     F.max('sincelastcomp3').alias('sincelastcomp3'), \n",
    "                                     F.max('sincelastcomp4').alias('sincelastcomp4')))\n",
    "\n",
    "# fill in missing value\n",
    "maint_count1 = comp1_2_3_4_final.fillna(0)\n",
    "\n",
    "maint_count1.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resample to every 3 hours"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# maint_count1 maintenance \n",
    "dt_truncated = ((round(unix_timestamp(col(\"datetime_tel\")) / time_val) * time_val)\n",
    "    .cast(\"timestamp\"))\n",
    "\n",
    "maint_resampled = maint_count1.withColumn(\"dt_truncated\", dt_truncated)\n",
    "maint_resampled1 = (maint_resampled.groupBy(\"machineID\",\"dt_truncated\")\n",
    "                                  .agg(F.mean('sincelastcomp1').alias('comp1sum'), \n",
    "                                       F.mean('sincelastcomp2').alias('comp2sum'), \n",
    "                                       F.mean('sincelastcomp3').alias('comp3sum'), \n",
    "                                       F.mean('sincelastcomp4').alias('comp4sum')))\n",
    "maint_resampled1.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Machine features - need to do one hot encoding for variable model "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# one hot encoding of the variable model\n",
    "catVarNames = ['model']  \n",
    "    \n",
    "sIndexers = [StringIndexer(inputCol=x, outputCol=x + '_indexed') for x in catVarNames]\n",
    "\n",
    "machines_cat = Pipeline(stages=sIndexers).fit(machines).transform(machines)\n",
    "\n",
    "# one-hot encode\n",
    "ohEncoders = [OneHotEncoder(inputCol=x + '_indexed', outputCol=x + '_encoded')\n",
    "              for x in catVarNames]\n",
    "ohPipelineModel = Pipeline(stages=ohEncoders).fit(machines_cat)\n",
    "machines_cat = ohPipelineModel.transform(machines_cat)\n",
    "\n",
    "drop_list = [col_n for col_n in machines_cat.columns if 'indexed' in col_n]\n",
    "\n",
    "machines_edit = machines_cat.select([column for column in machines_cat.columns if column not in drop_list])\n",
    "\n",
    "machines_edit.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating final feature matrix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# join error with components\n",
    "#err_mean_resampled1.show(3)\n",
    "#maint_resampled1.show(3)\n",
    "\n",
    "error_maint = (err_mean_resampled1.join(maint_resampled1, \n",
    "                                ((err_mean_resampled1['machineID'] == maint_resampled1['machineID']) \n",
    "                                  & (err_mean_resampled1['dt_truncated'] == maint_resampled1['dt_truncated'])), \"left\")\n",
    "                                  .drop(maint_resampled1.machineID).drop(maint_resampled1.dt_truncated))\n",
    "#error_maint.show(10, False)\n",
    "#error_maint.count(), len(error_maint.columns)\n",
    "\n",
    "# now join with machines\n",
    "#machines_edit.show(1)\n",
    "\n",
    "err_maint_mach = (error_maint.join(machines_edit, ((error_maint['machineID'] == machines_edit['machineID'])), \"left\")\n",
    "                             .drop(machines_edit.machineID))\n",
    "err_maint_mach_select = (err_maint_mach.select([c for c in err_maint_mach.columns if c not in \n",
    "                                               {'error1sum', 'error2sum', 'error3sum', 'error4sum', 'error5sum'}]))\n",
    "#err_maint_mach_select.show(10, False)\n",
    "#err_maint_mach_select.count(), len(err_maint_mach_select.columns)\n",
    "\n",
    "telemetry_all = (tel_mean_resampled1.join(tel_sd_resampled1, \n",
    "                             ((tel_mean_resampled1['machineID'] == tel_sd_resampled1['machineID']) \n",
    "                              & (tel_mean_resampled1['dt_truncated'] == tel_sd_resampled1['dt_truncated'])), \"left\")\n",
    "                              .drop(tel_sd_resampled1.machineID).drop(tel_sd_resampled1.dt_truncated))\n",
    "#telemetry_all.show(10, False)\n",
    "#telemetry_all.count(), len(telemetry_all.columns)\n",
    "\n",
    "# join telemetry_all with err_maint_mach_select to create final feature matrix\n",
    "final_feat = (telemetry_all.join(err_maint_mach_select, \n",
    "                                ((telemetry_all['machineID'] == err_maint_mach_select['machineID']) \n",
    "                                  & (telemetry_all['dt_truncated'] == err_maint_mach_select['dt_truncated'])), \"left\")\n",
    "                                 .drop(err_maint_mach_select.machineID).drop(err_maint_mach_select.dt_truncated))\n",
    "final_feat.show(5, False)\n",
    "#final_feat.count(), len(final_feat.columns)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Label construction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check failure sample data\n",
    "failures.show(5)\n",
    "\n",
    "# check the dimensions of the data\n",
    "failures.count(), len(failures.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check to see if there are duplicate rows based on machine, datetime\n",
    "failures1 = failures.dropDuplicates(['machineID', 'datetime'])\n",
    "\n",
    "# check the dimensions of the data\n",
    "failures1.count(), len(failures1.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# map the failure data to final feature matrix\n",
    "\n",
    "labeled_features = (final_feat.join(failures1, ((final_feat['machineID'] == failures1['machineID']) \n",
    "                                  & (final_feat['dt_truncated'] == failures1['datetime'])), \"left\")\n",
    "                                  .drop(failures1.machineID).drop(failures1.datetime))\n",
    "labeled_features.show(5, False)\n",
    "#labeled_features.count(), len(labeled_features.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# recoding the column 'failure' to be numeric double for the pyspark classification models\n",
    "labeled_features1 = (labeled_features.withColumn('failure', F.when(col('failure') == \"comp1\", 1.0)\n",
    "                                     .otherwise(col('failure')))\n",
    "                                     .withColumn('failure', F.when(col('failure') == \"comp2\", 2.0)\n",
    "                                     .otherwise(col('failure')))\n",
    "                                     .withColumn('failure', F.when(col('failure') == \"comp3\", 3.0)\n",
    "                                     .otherwise(col('failure')))\n",
    "                                     .withColumn('failure', F.when(col('failure') == \"comp4\", 4.0)\n",
    "                                     .otherwise(col('failure'))))\n",
    "\n",
    "labeled_features2 = labeled_features1.withColumn(\"failure1\", labeled_features1[\"failure\"].cast(DoubleType()))\n",
    "\n",
    "#labeled_features2.groupBy('failure').count().show()\n",
    "#labeled_features2.groupBy('failure1').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check data schema\n",
    "labeled_features2.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "labeled_features3 = labeled_features2.drop('failure').fillna(0)\n",
    "labeled_features3.dtypes\n",
    "#labeled_features3.groupBy('failure1').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# build the code for backfill with all machine data\n",
    "label_bfill1 = labeled_features3\n",
    "label_bfill1.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# lag values to manually backfill label (bfill =7)\n",
    "my_window = Window.partitionBy('machineID').orderBy(label_bfill1.dt_truncated.desc())\n",
    "\n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value1\", F.lag(label_bfill1.failure1).over(my_window)).fillna(0)\n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value2\", F.lag(label_bfill1.prev_value1).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value3\", F.lag(label_bfill1.prev_value2).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value4\", F.lag(label_bfill1.prev_value3).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value5\", F.lag(label_bfill1.prev_value4).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value6\", F.lag(label_bfill1.prev_value5).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value7\", F.lag(label_bfill1.prev_value6).over(my_window)).fillna(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# create the label column \n",
    "label_bfill2 = (label_bfill1.withColumn('label', label_bfill1.failure1 + label_bfill1.prev_value1 \n",
    "                         + label_bfill1.prev_value2 + label_bfill1.prev_value3 + label_bfill1.prev_value4 \n",
    "                         + label_bfill1.prev_value5 + label_bfill1.prev_value6 + label_bfill1.prev_value7))\n",
    "label_bfill2 = label_bfill2.withColumn('label_e', F.when(col('label') > 4, 4.0).otherwise(col('label')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "label_bfill3 = (label_bfill2.drop(label_bfill2.prev_value1).drop(label_bfill2.prev_value2)\n",
    "              .drop(label_bfill2.prev_value3).drop(label_bfill2.prev_value4)\n",
    "              .drop(label_bfill2.prev_value5).drop(label_bfill2.prev_value6)\n",
    "              .drop(label_bfill2.prev_value7).drop(label_bfill2.label))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "label_bfill3.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# write the final result as parquet file in blob location \n",
    "# https://github.com/Azure/ViennaDocs/blob/master/Documentation/UsingBlobForStorage.md\n",
    "CONTAINER_NAME = \"featureengineering\"\n",
    "\n",
    "# Create a new container if necessary, otherwise you can use an existing container.\n",
    "# This command creates the container if it does not already exist. Else it does nothing.\n",
    "my_service.create_container(CONTAINER_NAME, \n",
    "                            fail_on_exist=False, \n",
    "                            public_access=PublicAccess.Container)\n",
    "\n",
    "# you decide to partition the dataframe into three files and save them in the current folder.\n",
    "# if you wish to visualize them in the run history Output Files, specify the path \n",
    "# as './outputs/multiple_files.parquet'.\n",
    "#label_bfill3.coalesce(3).write.mode('overwrite').parquet('multiple_files.parquet')\n",
    "label_bfill3.write.mode('overwrite').parquet('featureengineering_files.parquet')\n",
    "\n",
    "# unlike the single file case, for multiple files we need to first delete results from the \n",
    "# previous run before uploading.\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if 'featureengineering_files.parquet' in blob.name:\n",
    "        my_service.delete_blob(CONTAINER_NAME, blob.name)\n",
    "\n",
    "# upload the entire folder into blob storage\n",
    "for name in glob.iglob('featureengineering_files.parquet/*'):\n",
    "    print(os.path.abspath(name))\n",
    "    my_service.create_blob_from_path(CONTAINER_NAME, name, name)\n",
    "\n",
    "print(\"Feature engineering final dataset files saved!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Hack2 docker",
   "language": "python",
   "name": "hack2_docker"
  },
  "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
<commit_msg>Remove my Azure subscription info<commit_after>{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Step 2: Feature Engineering\n",
    "\n",
    "Feature engineering combines the different data sources together to create a single data set of features (variables) that can be used to infer a machines's health condition over time. \n",
    "\n",
    "In this notebook, we will load the data stored in Azure Blob containers in the previous **Data Ingestion** notebook (`Code/data_ingestion.ipynb`). The note book uses several feature engineering methods to create a data set for use in our predictive maintenance machine learning solution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# import the libraries\n",
    "import os\n",
    "\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.functions import col, unix_timestamp, round\n",
    "from pyspark.sql.functions import datediff\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.types import DoubleType\n",
    "\n",
    "from pyspark.ml import Pipeline\n",
    "from pyspark.ml.feature import OneHotEncoder\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "from azure.storage.blob import BlockBlobService\n",
    "from azure.storage.blob import PublicAccess\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load data from Azure Blob storage container\n",
    "\n",
    "We have previously downloaded and stored the following data in an Azure blob storage container:\n",
    "\n",
    "  * Machines: Features differentiating each machine. For example age and model.\n",
    "  * Error: The log of non-critical errors. These errors may still indicate an impending component failure.\n",
    "  * Maint: Machine maintenance history detailing component replacement or regular maintenance activities withe the date of replacement.\n",
    "  * Telemetry: The operating conditions of a machine e.g. data collected from sensors.\n",
    "  * Failure history: The failure history of a machine or component within the machine.\n",
    "\n",
    "We'll load these files from blob, and create our analysis data set here. We'll write this data set back into a new blob container to use in our model building and evaluation notebook later. \n",
    "\n",
    "Since the Azure Blob storage account name and account key are not passed between notebooks, you'll need to provide those here again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Enter your Azure blob storage details here \n",
    "ACCOUNT_NAME = \"<your blob storage account name>\"\n",
    "\n",
    "# You can find the account key under the _Access Keys_ link in the \n",
    "# [Azure Portal](portal.azure.com) page for your Azure storage container.\n",
    "ACCOUNT_KEY =  \"<account key>\"\n",
    "\n",
    "#-------------------------------------------------------------------------------------------\n",
    "# The data from the Data Aquisition note book is stored in the dataingestion container.\n",
    "CONTAINER_NAME = \"dataingestion\"\n",
    "\n",
    "# Connect to your blob service     \n",
    "my_service = BlockBlobService(account_name=ACCOUNT_NAME, account_key=ACCOUNT_KEY)\n",
    "\n",
    "\n",
    "# We will store each of these data sets in blob storage in an \n",
    "# Azure Storage Container on your Azure subscription.\n",
    "# See https://github.com/Azure/ViennaDocs/blob/master/Documentation/UsingBlobForStorage.md\n",
    "# for details.\n",
    "\n",
    "# These file names detail which blob each files is stored under. \n",
    "MACH_DATA = 'machines_files.parquet'\n",
    "MAINT_DATA = 'maint_files.parquet'\n",
    "ERROR_DATA = 'errors_files.parquet'\n",
    "TELEMETRY_DATA = 'telemetry_files.parquet'\n",
    "FAILURE_DATA = 'failure_files.parquet'\n",
    "\n",
    "# create a local path where to store the results later.\n",
    "MACH_LOCAL_DIRECT = 'dataingestion_mach_result.parquet'\n",
    "ERROR_LOCAL_DIRECT = 'dataingestion_err_result.parquet'\n",
    "MAINT_LOCAL_DIRECT = 'dataingestion_maint_result.parquet'\n",
    "TELEMETRY_LOCAL_DIRECT = 'dataingestion_tel_result.parquet'\n",
    "FAILURES_LOCAL_DIRECT = 'dataingestion_fail_result.parquet'\n",
    "\n",
    "# define your blob service     \n",
    "my_service = BlockBlobService(account_name=ACCOUNT_NAME, account_key=ACCOUNT_KEY)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Machines data set\n",
    "\n",
    "Load the machines data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-3-05b91dc7eb0b>, line 6)",
     "output_type": "error",
     "traceback": [
      "\u001b[0;36m  File \u001b[0;32m\"<ipython-input-3-05b91dc7eb0b>\"\u001b[0;36m, line \u001b[0;32m6\u001b[0m\n\u001b[0;31m    print('DONE creating a local directory!')\u001b[0m\n\u001b[0m        ^\u001b[0m\n\u001b[0;31mSyntaxError\u001b[0m\u001b[0;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "# download the entire parquet result folder to local path for a new run \n",
    "\n",
    "# create a local path  to store the data.\n",
    "if not os.path.exists(MACH_LOCAL_DIRECT):\n",
    "    os.makedirs(MACH_LOCAL_DIRECT\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "# Connect to blob storage container\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if MACH_DATA in blob.name:\n",
    "        local_file = os.path.join(MACH_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "# Read in the data\n",
    "machines = spark.read.parquet(MACH_LOCAL_DIRECT)\n",
    "\n",
    "print(machines.count())\n",
    "machines.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Errors data set\n",
    "\n",
    "Load the errors data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load the previous created final dataset into the workspace\n",
    "\n",
    "# create a local path  to store the data.\n",
    "if not os.path.exists(ERROR_LOCAL_DIRECT):\n",
    "    os.makedirs(ERROR_LOCAL_DIRECT)\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "# Connect to blob storage container\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if ERROR_DATA in blob.name:\n",
    "        local_file = os.path.join(ERROR_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "# Read in the data\n",
    "errors = spark.read.parquet(ERROR_LOCAL_DIRECT)\n",
    "\n",
    "print(errors.count())\n",
    "errors.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Maintenance data set\n",
    "\n",
    "Load the maintenance data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load the previous created final dataset into the workspace\n",
    "\n",
    "# create a local path  to store the data.\n",
    " if not os.path.exists(MAINT_LOCAL_DIRECT):\n",
    "    os.makedirs(MAINT_LOCAL_DIRECT)\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "# Connect to blob storage container\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if MAINT_DATA in blob.name:\n",
    "        local_file = os.path.join(MAINT_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "# Read in the data\n",
    "maint = spark.read.parquet(MAINT_LOCAL_DIRECT)\n",
    "\n",
    "print(maint.count())\n",
    "maint.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Telemetry\n",
    "\n",
    "Load the telemetry data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load the previous created final dataset into the workspace\n",
    "\n",
    "# create a local path  to store the data.\n",
    "if not os.path.exists(TELEMETRY_LOCAL_DIRECT):\n",
    "    os.makedirs(TELEMETRY_LOCAL_DIRECT)\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "# Connect to blob storage container\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if TELEMETRY_DATA in blob.name:\n",
    "        local_file = os.path.join(TELEMETRY_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "# Read in the data\n",
    "telemetry = spark.read.parquet(TELEMETRY_LOCAL_DIRECT)\n",
    "\n",
    "print(telemetry.count())\n",
    "telemetry.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Failures data set\n",
    "\n",
    "Load the failures data set from your Azure blob."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# load the previous created final dataset into the workspace\n",
    "\n",
    "# create a local path  to store the data.\n",
    "\n",
    "if not os.path.exists(FAILURES_LOCAL_DIRECT):\n",
    "    os.makedirs(FAILURES_LOCAL_DIRECT)\n",
    "    print('DONE creating a local directory!')\n",
    "\n",
    "\n",
    "# download the entire parquet result folder to local path for a new run \n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if FAILURE_DATA in blob.name:\n",
    "        local_file = os.path.join(FAILURES_LOCAL_DIRECT, os.path.basename(blob.name))\n",
    "        my_service.get_blob_to_path(CONTAINER_NAME, blob.name, local_file)\n",
    "\n",
    "failures = spark.read.parquet(FAILURES_LOCAL_DIRECT)\n",
    "\n",
    "print(failures.count())\n",
    "failures.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature engineering \n",
    "\n",
    "Feature engineering combines the different data sources together to create a single data set of features (variables) that can be used to infer a machines's health condition over time. The ultimate goal is to generate a single record for each time unit for each asset combining its features and labels to be fed into the machine learning algorithm. In order to prepare that clean final data set, some pre-processing steps should be taken. First step is to divide the duration of data collection into time units where each record belongs to a time unit for an asset.\n",
    "\n",
    "The measurement unit for time can be in seconds, minutes, hours, days, months, cycles, miles or transactions depending on the efficiency of data preparation and the changes observed in the conditions of the asset from a time unit to the other or other factors specific to the domain. In other words, the time unit does not have to be the same as the frequency of data collection as in many cases data may not show any difference from one unit to the other. For example, if temperature values were being collected every 10 seconds, picking a time unit of 10 seconds for the whole analysis inflates the number of examples without providing any additional information. Better strategy would be to use average over an hour as an example.\n",
    "\n",
    "### Rolling aggregates\n",
    "\n",
    "For each record of an asset, we pick a rolling window of size \"W\" which is the number of units of time that we would like to compute historical aggregates for. We then compute rolling aggregate features using the W periods before the date of that record. Some example rolling aggregates can be rolling counts, means, standard deviations, outliers based on standard deviations, CUSUM measures, minimum and maximum values for the window. Another interesting technique is to capture trend changes, spikes and level changes using algorithms that detect anomalies in data using anomaly detection algorithms.\n",
    "\n",
    "### Lag features\n",
    "As mentioned earlier, in predictive maintenance, historical data usually comes with timestamps indicating the time of collection for each piece of data. There are many ways of creating features from the data that comes with timestamped data. In this section, we discuss some of these methods used for predictive maintenance. However, we are not limited by these methods alone. Since feature engineering is considered to be one of the most creative areas of predictive modeling, there could be many other ways to create features. Here, we provide some general techniques."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Telemetry features\n",
    "\n",
    "Because the telemetry data set is the largest time series data we have, we start feature engineering here. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# rolling mean\n",
    "# Temporary storage for rolling means\n",
    "tel_mean = telemetry\n",
    "\n",
    "# Which features are we interested in telemetry data set\n",
    "rolling_features = ['volt','rotate', 'pressure', 'vibration']\n",
    "               \n",
    "# We choose two windows for our rolling windows 3hrs, 24 hrs\n",
    "lags = [3,24]\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('machineID').orderBy('datetime').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        tel_mean = tel_mean.withColumn(col_name+'_rollingmean_'+str(lag_n), F.avg(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We repeat this rolling window process for the standard deviation. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# rolling standard deviation\n",
    "# Temporary storage for rolling means\n",
    "tel_sd = telemetry\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('machineID').orderBy('datetime').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features:\n",
    "        tel_sd = tel_sd.withColumn(col_name+'_rollingstd_'+str(lag_n), F.stddev(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resample every 3 hours"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# tel_mean rolling mean\n",
    "# 3 hours = 10800 seconds  \n",
    "time_val = 3 * 60 * 60\n",
    "\n",
    "# I think this is grabbing datetime from the tel_sd data set, which is equivalent to telemetry\n",
    "dt_truncated = ((round(unix_timestamp(col(\"datetime\")) / time_val) * time_val).cast(\"timestamp\"))\n",
    "\n",
    "tel_mean_resampled = tel_mean.withColumn(\"dt_truncated\", dt_truncated).drop('volt', 'rotate', 'pressure', 'vibration')\n",
    "\n",
    "tel_mean_resampled1 = (tel_mean_resampled.groupBy(\"machineID\",\"dt_truncated\")\n",
    "                               .agg(F.mean('volt_rollingmean_3').alias('volt_rollingmean_3'),\n",
    "                                    F.mean('rotate_rollingmean_3').alias('rotate_rollingmean_3'), \n",
    "                                    F.mean('pressure_rollingmean_3').alias('pressure_rollingmean_3'), \n",
    "                                    F.mean('vibration_rollingmean_3').alias('vibration_rollingmean_3'), \n",
    "                                    F.mean('volt_rollingmean_24').alias('volt_rollingmean_24'),\n",
    "                                    F.mean('rotate_rollingmean_24').alias('rotate_rollingmean_24'), \n",
    "                                    F.mean('pressure_rollingmean_24').alias('pressure_rollingmean_24'), \n",
    "                                    F.mean('vibration_rollingmean_24').alias('vibration_rollingmean_24')))\n",
    "\n",
    "tel_mean_resampled1.where((col(\"machineID\") == 1)).toPandas().head(5)\n",
    "tel_mean_resampled1.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# tel_sd rolling sd\n",
    "tel_sd_resampled = (tel_sd.withColumn(\"dt_truncated\", dt_truncated).drop('volt', 'rotate', 'pressure', 'vibration')\n",
    "                        .fillna(0))\n",
    "\n",
    "tel_sd_resampled1 = (tel_sd_resampled.groupBy(\"machineID\",\"dt_truncated\")\n",
    "                               .agg(F.sd('volt_rollingstd_3').alias('volt_rollingstd_3'),\n",
    "                                    F.sd('rotate_rollingstd_3').alias('rotate_rollingstd_3'), \n",
    "                                    F.sd('pressure_rollingstd_3').alias('pressure_rollingstd_3'), \n",
    "                                    F.sd('vibration_rollingstd_3').alias('vibration_rollingstd_3'), \n",
    "                                    F.sd('volt_rollingstd_24').alias('volt_rollingstd_24'),\n",
    "                                    F.sd('rotate_rollingstd_24').alias('rotate_rollingstd_24'), \n",
    "                                    F.sd('pressure_rollingstd_24').alias('pressure_rollingstd_24'), \n",
    "                                    F.sd('vibration_rollingstd_24').alias('vibration_rollingstd_24')))\n",
    "tel_sd_resampled1.show(5)\n",
    "tel_sd_resampled1.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Lag features from Errors"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a column for each errorID \n",
    "error1 = errors.groupBy(\"machineID\",\"datetime\",\"errorID\").pivot('errorID').agg(F.count('machineID').alias('dummy'))\n",
    "\n",
    "# remove the column called errorID and fill in missing values\n",
    "error2 = error1.drop('errorID').fillna(0)\n",
    "\n",
    "# combine errors for a given machine in a given hour\n",
    "error3 = (error2.groupBy(\"machineID\",\"datetime\")\n",
    "                .agg(F.sum('error1').alias('error1sum'), \n",
    "                     F.sum('error2').alias('error2sum'), \n",
    "                     F.sum('error3').alias('error3sum'), \n",
    "                     F.sum('error4').alias('error4sum'), \n",
    "                     F.sum('error5').alias('error5sum')))\n",
    "\n",
    "error3.toPandas().head(5)\n",
    "error3.count(), len(error3.columns)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We want the same number of rows in error as in telemetry.\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# join the telemetry data with errors\n",
    "error_count = (telemetry.join(error3, ((telemetry['machineID'] == error3['machineID']) \n",
    "                                  & (telemetry['datetime'] == error3['datetime'])), \"left\")\n",
    "               .drop('volt', 'rotate', 'pressure', 'vibration').drop(error3.machineID).drop(error3.datetime))\n",
    "\n",
    "# fill in missing value\n",
    "error_count1 = error_count.fillna(0)\n",
    "\n",
    "error_count1.toPandas().show(5)\n",
    "error_count1.count(), len(error_count1.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rolling_features1 = ['error1sum','error2sum', 'error3sum', 'error4sum', 'error5sum']\n",
    "               \n",
    "# lag window 24 hrs\n",
    "lags = [24]\n",
    "\n",
    "# rolling mean\n",
    "err_mean = error_count1\n",
    "\n",
    "for lag_n in lags:\n",
    "    wSpec = Window.partitionBy('machineID').orderBy('datetime').rowsBetween(1-lag_n, 0)\n",
    "    for col_name in rolling_features1:\n",
    "        err_mean = err_mean.withColumn(col_name+'_rollingmean_'+str(lag_n), F.avg(col(col_name)).over(wSpec))\n",
    "        print(\"Lag = %d, Column = %s\" % (lag_n, col_name))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resample to every 3 hours"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dt_truncated = ((round(unix_timestamp(col(\"datetime\")) / time_val) * time_val)\n",
    "    .cast(\"timestamp\"))\n",
    "\n",
    "err_mean_resampled = (err_mean.withColumn(\"dt_truncated\", dt_truncated)\n",
    "                    .drop('error1sum', 'error2sum', 'error3sum', 'error4sum', 'error5sum').fillna(0))\n",
    "\n",
    "err_mean_resampled1 = (err_mean_resampled.groupBy(\"machineID\",\"dt_truncated\")\n",
    "                               .agg(F.mean('error1sum_rollingmean_24').alias('error1sum_rollingmean_24'), \n",
    "                                    F.mean('error2sum_rollingmean_24').alias('error2sum_rollingmean_24'), \n",
    "                                    F.mean('error3sum_rollingmean_24').alias('error3sum_rollingmean_24'), \n",
    "                                    F.mean('error4sum_rollingmean_24').alias('error4sum_rollingmean_24'), \n",
    "                                    F.mean('error5sum_rollingmean_24').alias('error5sum_rollingmean_24')))\n",
    "err_mean_resampled1.toPandas().head(5)\n",
    "err_mean_resampled1.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement from maintenance "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a column for each comp \n",
    "maint1 = maint.groupBy(\"machineID\",\"datetime\",\"comp\").pivot('comp').agg(F.count('machineID').alias('dummy'))\n",
    "\n",
    "# remove the column called comp and fill in missing values\n",
    "maint2 = maint1.drop('comp').fillna(0)\n",
    "\n",
    "# combine maintenance for a given machine in a given hour\n",
    "maint3 = (maint2.groupBy(\"machineID\",\"datetime\").agg(F.sum('comp1').alias('comp1sum'), \n",
    "                                                    F.sum('comp2').alias('comp2sum'), \n",
    "                                                    F.sum('comp3').alias('comp3sum'),\n",
    "                                                    F.sum('comp4').alias('comp4sum')))\n",
    "maint3.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement for component-1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_maint_comp1 = (maint3.where((col(\"comp1sum\") == '1')).withColumnRenamed('datetime','datetime_maint')\n",
    "                           .drop('comp2sum', 'comp3sum', 'comp4sum'))\n",
    "\n",
    "test_tel_comp1 = (telemetry.withColumnRenamed('datetime','datetime_tel')\n",
    "                  .drop(telemetry.volt).drop(telemetry.rotate).drop(telemetry.pressure).drop(telemetry.vibration))\n",
    "\n",
    "test_maint_tel_comp1 = test_tel_comp1.join(test_maint_comp1, ((test_tel_comp1['machineID']==\n",
    "                                                               test_maint_comp1['machineID']) \n",
    "                                            & (test_tel_comp1['datetime_tel'] > test_maint_comp1['datetime_maint']) \n",
    "                                            & (test_maint_comp1['comp1sum'] == '1'))).drop(test_maint_comp1.machineID)\n",
    "\n",
    "comp1 = (test_maint_tel_comp1.withColumn(\"sincelastcomp1\", \n",
    "              datediff(test_maint_tel_comp1.datetime_tel, test_maint_tel_comp1.datetime_maint))\n",
    "              .drop(test_maint_tel_comp1.datetime_maint).drop(test_maint_tel_comp1.comp1sum))\n",
    "\n",
    "comp1.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement for component-2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_maint_comp2 = (maint3.where(col(\"comp2sum\") == '1').withColumnRenamed('datetime','datetime_maint')\n",
    "                         .drop('comp1sum', 'comp3sum', 'comp4sum'))\n",
    "\n",
    "test_tel_comp2 = (telemetry.withColumnRenamed('datetime','datetime_tel')\n",
    "                          .drop(telemetry.volt).drop(telemetry.rotate).drop(telemetry.pressure)\n",
    "                          .drop(telemetry.vibration))\n",
    "\n",
    "test_maint_tel_comp2 = (test_tel_comp2.join(test_maint_comp2, ((test_tel_comp2['machineID']==\n",
    "                                                                test_maint_comp2['machineID']) \n",
    "                                        & (test_tel_comp2['datetime_tel'] > test_maint_comp2['datetime_maint']) \n",
    "                                        & (test_maint_comp2['comp2sum'] == '1') \n",
    "                                           )).drop(test_maint_comp2.machineID))\n",
    "\n",
    "comp2 = (test_maint_tel_comp2.withColumn(\"sincelastcomp2\", \n",
    "              datediff(test_maint_tel_comp2.datetime_tel, test_maint_tel_comp2.datetime_maint))\n",
    "              .drop(test_maint_tel_comp2.datetime_maint).drop(test_maint_tel_comp2.comp2sum))\n",
    "comp2.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement for component-3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_maint_comp3 = (maint3.where(col(\"comp3sum\") == '1').withColumnRenamed('datetime','datetime_maint')\n",
    "                          .drop('comp1sum', 'comp2sum', 'comp4sum'))\n",
    "\n",
    "test_tel_comp3 = (telemetry.withColumnRenamed('datetime','datetime_tel')\n",
    "                    .drop(telemetry.volt).drop(telemetry.rotate).drop(telemetry.pressure).drop(telemetry.vibration))\n",
    "\n",
    "test_maint_tel_comp3 = test_tel_comp3.join(test_maint_comp3, ((test_tel_comp3['machineID']==\n",
    "                                                               test_maint_comp3['machineID']) \n",
    "                                        & (test_tel_comp3['datetime_tel'] > test_maint_comp3['datetime_maint']) \n",
    "                                        & (test_maint_comp3['comp3sum'] == '1') \n",
    "                                           )).drop(test_maint_comp3.machineID)\n",
    "\n",
    "comp3 = (test_maint_tel_comp3.withColumn(\"sincelastcomp3\", \n",
    "              datediff(test_maint_tel_comp3.datetime_tel, test_maint_tel_comp3.datetime_maint))\n",
    "              .drop(test_maint_tel_comp3.datetime_maint).drop(test_maint_tel_comp3.comp3sum))\n",
    "comp3.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Days since last replacement for component-4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_maint_comp4 = (maint3.where(col(\"comp4sum\") == '1').withColumnRenamed('datetime','datetime_maint')\n",
    "                         .drop('comp1sum', 'comp2sum', 'comp3sum'))\n",
    "\n",
    "test_tel_comp4 = (telemetry.withColumnRenamed('datetime','datetime_tel')\n",
    "                  .drop(telemetry.volt).drop(telemetry.rotate).drop(telemetry.pressure).drop(telemetry.vibration))\n",
    "\n",
    "test_maint_tel_comp4 = test_tel_comp4.join(test_maint_comp4, ((test_tel_comp4['machineID']==\n",
    "                                                               test_maint_comp4['machineID']) \n",
    "                                        & (test_tel_comp4['datetime_tel'] > test_maint_comp4['datetime_maint']) \n",
    "                                        & (test_maint_comp4['comp4sum'] == '1'))).drop(test_maint_comp4.machineID)\n",
    "\n",
    "comp4 = (test_maint_tel_comp4.withColumn(\"sincelastcomp4\", \n",
    "              datediff(test_maint_tel_comp4.datetime_tel, test_maint_tel_comp4.datetime_maint))\n",
    "              .drop(test_maint_tel_comp4.datetime_maint).drop(test_maint_tel_comp4.comp4sum))\n",
    "comp4.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##  Combine comp1, comp2, comp3, comp4 to generate the maintenance feature set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# left join comp1 with (comp2, comp3, comp4) \n",
    "# left join comp2 with (comp3, comp4) \n",
    "# left join comp3, comp4 \n",
    "comp3_4 = (comp3.join(comp4, ((comp3['machineID'] == comp4['machineID']) \n",
    "                                  & (comp3['datetime_tel'] == comp4['datetime_tel'])), \"left\")\n",
    "                                  .drop(comp4.machineID).drop(comp4.datetime_tel))comp2_3_4 = (comp2.join(comp3_4, ((comp2['machineID'] == comp3_4['machineID']) \n",
    "                                  & (comp2['datetime_tel'] == comp3_4['datetime_tel'])), \"left\")\n",
    "                                  .drop(comp3_4.machineID).drop(comp3_4.datetime_tel))\n",
    "comp1_2_3_4 = (comp1.join(comp2_3_4, ((comp1['machineID'] == comp2_3_4['machineID']) \n",
    "                                  & (comp1['datetime_tel'] == comp2_3_4['datetime_tel'])), \"left\")\n",
    "                                 .drop(comp2_3_4.machineID).drop(comp2_3_4.datetime_tel))\n",
    "comp1_2_3_4_final = (comp1_2_3_4.groupBy(\"machineID\", \"datetime_tel\")\n",
    "                                .agg(F.max('sincelastcomp1').alias('sincelastcomp1'), \n",
    "                                     F.max('sincelastcomp2').alias('sincelastcomp2'), \n",
    "                                     F.max('sincelastcomp3').alias('sincelastcomp3'), \n",
    "                                     F.max('sincelastcomp4').alias('sincelastcomp4')))\n",
    "\n",
    "# fill in missing value\n",
    "maint_count1 = comp1_2_3_4_final.fillna(0)\n",
    "\n",
    "maint_count1.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Resample to every 3 hours"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# maint_count1 maintenance \n",
    "dt_truncated = ((round(unix_timestamp(col(\"datetime_tel\")) / time_val) * time_val)\n",
    "    .cast(\"timestamp\"))\n",
    "\n",
    "maint_resampled = maint_count1.withColumn(\"dt_truncated\", dt_truncated)\n",
    "maint_resampled1 = (maint_resampled.groupBy(\"machineID\",\"dt_truncated\")\n",
    "                                  .agg(F.mean('sincelastcomp1').alias('comp1sum'), \n",
    "                                       F.mean('sincelastcomp2').alias('comp2sum'), \n",
    "                                       F.mean('sincelastcomp3').alias('comp3sum'), \n",
    "                                       F.mean('sincelastcomp4').alias('comp4sum')))\n",
    "maint_resampled1.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Machine features - need to do one hot encoding for variable model "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# one hot encoding of the variable model\n",
    "catVarNames = ['model']  \n",
    "    \n",
    "sIndexers = [StringIndexer(inputCol=x, outputCol=x + '_indexed') for x in catVarNames]\n",
    "\n",
    "machines_cat = Pipeline(stages=sIndexers).fit(machines).transform(machines)\n",
    "\n",
    "# one-hot encode\n",
    "ohEncoders = [OneHotEncoder(inputCol=x + '_indexed', outputCol=x + '_encoded')\n",
    "              for x in catVarNames]\n",
    "ohPipelineModel = Pipeline(stages=ohEncoders).fit(machines_cat)\n",
    "machines_cat = ohPipelineModel.transform(machines_cat)\n",
    "\n",
    "drop_list = [col_n for col_n in machines_cat.columns if 'indexed' in col_n]\n",
    "\n",
    "machines_edit = machines_cat.select([column for column in machines_cat.columns if column not in drop_list])\n",
    "\n",
    "machines_edit.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creating final feature matrix"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# join error with components\n",
    "#err_mean_resampled1.show(3)\n",
    "#maint_resampled1.show(3)\n",
    "\n",
    "error_maint = (err_mean_resampled1.join(maint_resampled1, \n",
    "                                ((err_mean_resampled1['machineID'] == maint_resampled1['machineID']) \n",
    "                                  & (err_mean_resampled1['dt_truncated'] == maint_resampled1['dt_truncated'])), \"left\")\n",
    "                                  .drop(maint_resampled1.machineID).drop(maint_resampled1.dt_truncated))\n",
    "#error_maint.show(10, False)\n",
    "#error_maint.count(), len(error_maint.columns)\n",
    "\n",
    "# now join with machines\n",
    "#machines_edit.show(1)\n",
    "\n",
    "err_maint_mach = (error_maint.join(machines_edit, ((error_maint['machineID'] == machines_edit['machineID'])), \"left\")\n",
    "                             .drop(machines_edit.machineID))\n",
    "err_maint_mach_select = (err_maint_mach.select([c for c in err_maint_mach.columns if c not in \n",
    "                                               {'error1sum', 'error2sum', 'error3sum', 'error4sum', 'error5sum'}]))\n",
    "#err_maint_mach_select.show(10, False)\n",
    "#err_maint_mach_select.count(), len(err_maint_mach_select.columns)\n",
    "\n",
    "telemetry_all = (tel_mean_resampled1.join(tel_sd_resampled1, \n",
    "                             ((tel_mean_resampled1['machineID'] == tel_sd_resampled1['machineID']) \n",
    "                              & (tel_mean_resampled1['dt_truncated'] == tel_sd_resampled1['dt_truncated'])), \"left\")\n",
    "                              .drop(tel_sd_resampled1.machineID).drop(tel_sd_resampled1.dt_truncated))\n",
    "#telemetry_all.show(10, False)\n",
    "#telemetry_all.count(), len(telemetry_all.columns)\n",
    "\n",
    "# join telemetry_all with err_maint_mach_select to create final feature matrix\n",
    "final_feat = (telemetry_all.join(err_maint_mach_select, \n",
    "                                ((telemetry_all['machineID'] == err_maint_mach_select['machineID']) \n",
    "                                  & (telemetry_all['dt_truncated'] == err_maint_mach_select['dt_truncated'])), \"left\")\n",
    "                                 .drop(err_maint_mach_select.machineID).drop(err_maint_mach_select.dt_truncated))\n",
    "final_feat.show(5, False)\n",
    "#final_feat.count(), len(final_feat.columns)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Label construction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check failure sample data\n",
    "failures.show(5)\n",
    "\n",
    "# check the dimensions of the data\n",
    "failures.count(), len(failures.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check to see if there are duplicate rows based on machine, datetime\n",
    "failures1 = failures.dropDuplicates(['machineID', 'datetime'])\n",
    "\n",
    "# check the dimensions of the data\n",
    "failures1.count(), len(failures1.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# map the failure data to final feature matrix\n",
    "\n",
    "labeled_features = (final_feat.join(failures1, ((final_feat['machineID'] == failures1['machineID']) \n",
    "                                  & (final_feat['dt_truncated'] == failures1['datetime'])), \"left\")\n",
    "                                  .drop(failures1.machineID).drop(failures1.datetime))\n",
    "labeled_features.show(5, False)\n",
    "#labeled_features.count(), len(labeled_features.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# recoding the column 'failure' to be numeric double for the pyspark classification models\n",
    "labeled_features1 = (labeled_features.withColumn('failure', F.when(col('failure') == \"comp1\", 1.0)\n",
    "                                     .otherwise(col('failure')))\n",
    "                                     .withColumn('failure', F.when(col('failure') == \"comp2\", 2.0)\n",
    "                                     .otherwise(col('failure')))\n",
    "                                     .withColumn('failure', F.when(col('failure') == \"comp3\", 3.0)\n",
    "                                     .otherwise(col('failure')))\n",
    "                                     .withColumn('failure', F.when(col('failure') == \"comp4\", 4.0)\n",
    "                                     .otherwise(col('failure'))))\n",
    "\n",
    "labeled_features2 = labeled_features1.withColumn(\"failure1\", labeled_features1[\"failure\"].cast(DoubleType()))\n",
    "\n",
    "#labeled_features2.groupBy('failure').count().show()\n",
    "#labeled_features2.groupBy('failure1').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check data schema\n",
    "labeled_features2.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "labeled_features3 = labeled_features2.drop('failure').fillna(0)\n",
    "labeled_features3.dtypes\n",
    "#labeled_features3.groupBy('failure1').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# build the code for backfill with all machine data\n",
    "label_bfill1 = labeled_features3\n",
    "label_bfill1.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# lag values to manually backfill label (bfill =7)\n",
    "my_window = Window.partitionBy('machineID').orderBy(label_bfill1.dt_truncated.desc())\n",
    "\n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value1\", F.lag(label_bfill1.failure1).over(my_window)).fillna(0)\n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value2\", F.lag(label_bfill1.prev_value1).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value3\", F.lag(label_bfill1.prev_value2).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value4\", F.lag(label_bfill1.prev_value3).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value5\", F.lag(label_bfill1.prev_value4).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value6\", F.lag(label_bfill1.prev_value5).over(my_window)).fillna(0) \n",
    "label_bfill1 = label_bfill1.withColumn(\"prev_value7\", F.lag(label_bfill1.prev_value6).over(my_window)).fillna(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# create the label column \n",
    "label_bfill2 = (label_bfill1.withColumn('label', label_bfill1.failure1 + label_bfill1.prev_value1 \n",
    "                         + label_bfill1.prev_value2 + label_bfill1.prev_value3 + label_bfill1.prev_value4 \n",
    "                         + label_bfill1.prev_value5 + label_bfill1.prev_value6 + label_bfill1.prev_value7))\n",
    "label_bfill2 = label_bfill2.withColumn('label_e', F.when(col('label') > 4, 4.0).otherwise(col('label')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "label_bfill3 = (label_bfill2.drop(label_bfill2.prev_value1).drop(label_bfill2.prev_value2)\n",
    "              .drop(label_bfill2.prev_value3).drop(label_bfill2.prev_value4)\n",
    "              .drop(label_bfill2.prev_value5).drop(label_bfill2.prev_value6)\n",
    "              .drop(label_bfill2.prev_value7).drop(label_bfill2.label))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "label_bfill3.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# write the final result as parquet file in blob location \n",
    "# https://github.com/Azure/ViennaDocs/blob/master/Documentation/UsingBlobForStorage.md\n",
    "CONTAINER_NAME = \"featureengineering\"\n",
    "\n",
    "# Create a new container if necessary, otherwise you can use an existing container.\n",
    "# This command creates the container if it does not already exist. Else it does nothing.\n",
    "my_service.create_container(CONTAINER_NAME, \n",
    "                            fail_on_exist=False, \n",
    "                            public_access=PublicAccess.Container)\n",
    "\n",
    "# you decide to partition the dataframe into three files and save them in the current folder.\n",
    "# if you wish to visualize them in the run history Output Files, specify the path \n",
    "# as './outputs/multiple_files.parquet'.\n",
    "#label_bfill3.coalesce(3).write.mode('overwrite').parquet('multiple_files.parquet')\n",
    "label_bfill3.write.mode('overwrite').parquet('featureengineering_files.parquet')\n",
    "\n",
    "# unlike the single file case, for multiple files we need to first delete results from the \n",
    "# previous run before uploading.\n",
    "for blob in my_service.list_blobs(CONTAINER_NAME):\n",
    "    if 'featureengineering_files.parquet' in blob.name:\n",
    "        my_service.delete_blob(CONTAINER_NAME, blob.name)\n",
    "\n",
    "# upload the entire folder into blob storage\n",
    "for name in glob.iglob('featureengineering_files.parquet/*'):\n",
    "    print(os.path.abspath(name))\n",
    "    my_service.create_blob_from_path(CONTAINER_NAME, name, name)\n",
    "\n",
    "print(\"Feature engineering final dataset files saved!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Hack2 docker",
   "language": "python",
   "name": "hack2_docker"
  },
  "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}

{'fraction_non_alphanumeric': 0.17012462146371118, 'fraction_numerical': 0.01506264845939952, 'mean_word_length': 2.5770949945636854, 'pattern_counts': {'":': 598, '<': 12, '<?xml version=': 0, '>': 22, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 354, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6412988', 'n_tokens_mistral': 30686, 'n_tokens_neox': 29543, 'n_words': 5678}