<jupyter_start><jupyter_text>Import required python package and set the Cloudant credentialsflightPredict is a helper package used to train and run Spark MLLib models for predicting flight delays based on Weather data<jupyter_code>sc.addPyFile("https://github.com/ibm-watson-data-lab/simple-data-pipe-connector-flightstats/raw/master/flightPredict/training.py")
sc.addPyFile("https://github.com/ibm-watson-data-lab/simple-data-pipe-connector-flightstats/raw/master/flightPredict/run.py")
import training
import run
%matplotlib inline
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.linalg import Vectors
from numpy import array
import numpy as np
import math
from datetime import datetime
from dateutil import parser

sqlContext=SQLContext(sc)
training.sqlContext = sqlContext
training.cloudantHost='dtaieb.cloudant.com'
training.cloudantUserName='weenesserliffircedinvers'
training.cloudantPassword='<PASSWORD>'
training.weatherUrl='https://4b88408f-11e5-4ddc-91a6-fbd442e84879:p6hxeJsfIb@twcservice.mybluemix.net'<jupyter_output><empty_output><jupyter_text>load data from training data set and print the schema<jupyter_code>dbName="pycon_flightpredict_training_set"
%time cloudantdata = training.loadDataSet(dbName,"training")
%time cloudantdata.printSchema()
%time cloudantdata.count()<jupyter_output>Successfully cached dataframe
Successfully registered SQL table training
CPU times: user 12.5 ms, sys: 2.93 ms, total: 15.4 ms
Wall time: 1min 17s
root
 |-- _id: string (nullable = true)
 |-- _rev: string (nullable = true)
 |-- actualRunwayArrival: struct (nullable = true)
 |    |-- dateLocal: string (nullable = true)
 |    |-- dateUtc: string (nullable = true)
 |-- actualRunwayDeparture: struct (nullable = true)
 |    |-- dateLocal: string (nullable = true)
 |    |-- dateUtc: string (nullable = true)
 |-- arrivalAirportFsCode: string (nullable = true)
 |-- arrivalTerminal: string (nullable = true)
 |-- arrivalTime: string (nullable = true)
 |-- arrivalWeather: struct (nullable = true)
 |    |-- blunt_phrase: string (nullable = true)
 |    |-- class: string (nullable = true)
 |    |-- clds: string (nullable = true)
 |    |-- day_ind: string (nullable = true)
 |    |-- dewPt: long (nullable = true)
 |    |-- expire_time_gmt: long (nullable = true)
 |    |-- feels_like: long (nullable = [...]<jupyter_text>Visualize classes in scatter plot based on 2 features<jupyter_code>training.scatterPlotForFeatures(cloudantdata, \
     "departureWeather.temp","arrivalWeather.temp","Departure Airport Temp", "Arrival Airport Temp")
training.scatterPlotForFeatures(cloudantdata,\
     "departureWeather.pressure","arrivalWeather.pressure","Departure Airport Pressure", "Arrival Airport Pressure")
training.scatterPlotForFeatures(cloudantdata,\
 "departureWeather.wspd","arrivalWeather.wspd","Departure Airport Wind Speed", "Arrival Airport Wind Speed")<jupyter_output><empty_output><jupyter_text>Load the training data as an RDD of LabeledPoint<jupyter_code>trainingData = training.loadLabeledDataRDD("training")
trainingData.take(5)<jupyter_output><empty_output><jupyter_text>Train multiple classification models<jupyter_code>from pyspark.mllib.classification import LogisticRegressionWithLBFGS
logRegModel = LogisticRegressionWithLBFGS.train(trainingData.map(lambda lp: LabeledPoint(lp.label,\
      np.fromiter(map(lambda x: 0.0 if np.isnan(x) else x,lp.features.toArray()),dtype=np.double )))\
      , iterations=1000, validateData=False, intercept=False)
print(logRegModel)
from pyspark.mllib.classification import NaiveBayes
#NaiveBayes requires non negative features, set them to 0 for now
modelNaiveBayes = NaiveBayes.train(trainingData.map(lambda lp: LabeledPoint(lp.label, \
                    np.fromiter(map(lambda x: x if x>0.0 else 0.0,lp.features.toArray()),dtype=np.int)\
               ))\
          )

print(modelNaiveBayes)
from pyspark.mllib.tree import DecisionTree
modelDecisionTree = DecisionTree.trainClassifier(trainingData.map(lambda lp: LabeledPoint(lp.label,\
      np.fromiter(map(lambda x: 0.0 if np.isnan(x) else x,lp.features.toArray()),dtype=np.double )))\
      , numClasses=training.getNumClasses(), categoricalFeaturesInfo={})
print(modelDecisionTree)
from pyspark.mllib.tree import RandomForest
modelRandomForest = RandomForest.trainClassifier(trainingData.map(lambda lp: LabeledPoint(lp.label,\
      np.fromiter(map(lambda x: 0.0 if np.isnan(x) else x,lp.features.toArray()),dtype=np.double )))\
      , numClasses=training.getNumClasses(), categoricalFeaturesInfo={},numTrees=100)
print(modelRandomForest)<jupyter_output>TreeEnsembleModel classifier with 100 trees<jupyter_text>Load Test data from Cloudant database and compute accuracy metrics<jupyter_code>dbTestName="pycon_flightpredict_test_set"
testCloudantdata = training.loadDataSet(dbTestName,"test")
testCloudantdata.count()
testData = training.loadLabeledDataRDD("test")
training.displayConfusionTable=True
training.runMetrics(testData,modelNaiveBayes,modelDecisionTree,logRegModel,modelRandomForest)<jupyter_output><empty_output><jupyter_text>Accuracy analysis and model refinement* Run Histogram to refine classification<jupyter_code>rdd = sqlContext.sql("select deltaDeparture from training").map(lambda s: s.deltaDeparture)\
    .filter(lambda s: s < 50 and s > 12)
    
print(rdd.count())

histo = rdd.histogram(50)
    
#print(histo[0])
#print(histo[1])
%matplotlib inline
import matplotlib
import matplotlib.pyplot as plt
import numpy as np
bins = [i for i in histo[0]]

params = plt.gcf()
plSize = params.get_size_inches()
params.set_size_inches( (plSize[0]*2.5, plSize[1]*2) )
plt.ylabel('Number of records')
plt.xlabel('Bin')
plt.title('Histogram')
intervals = [abs(j-i) for i,j in zip(bins[:-1], bins[1:])]
values=[sum(intervals[:i]) for i in range(0,len(intervals))]
plt.bar(values, histo[1], intervals, color='b', label = "Bins")
plt.xticks(bins[:-1],[int(i) for i in bins[:-1]])
plt.legend()

plt.show()<jupyter_output>19095<jupyter_text>* Customize classification using Training Handler class extension* Add new features to the model* Re-build the models* Re-compute accuracy metrics<jupyter_code>class customTrainingHandler(training.defaultTrainingHandler):
    def getClassLabel(self, value):
        if ( int(value)==0 ):
            return "Delayed less than 13 minutes"
        elif (int(value)==1 ):
            return "Delayed between 13 and 41 minutes"
        elif (int(value) == 2 ):
            return "Delayed more than 41 minutes"
        return value
    
    def numClasses(self):
        return 3
    
    def computeClassification(self, s):
        return 0 if s.deltaDeparture<13 else (1 if s.deltaDeparture < 41 else 2)
    
    def customTrainingFeaturesNames(self ):
        return ["departureTime"]
    
    def customTrainingFeatures(self, s):
        dt=parser.parse(s.departureTime)
        print(dt)
        features=[]
        for i in range(0,7):
            features.append(1 if dt.weekday()==i else 0)
        return features

training.customTrainingHandler=customTrainingHandler()

#reload the training labeled data RDD
trainingData = training.loadLabeledDataRDD("training")

#recompute the models
logRegModel = LogisticRegressionWithLBFGS.train(trainingData.map(lambda lp: LabeledPoint(lp.label,\
      np.fromiter(map(lambda x: 0.0 if np.isnan(x) else x,lp.features.toArray()),dtype=np.double )))\
      , iterations=1000, validateData=False, intercept=False)
modelNaiveBayes = NaiveBayes.train(trainingData.map(lambda lp: LabeledPoint(lp.label, \
                    np.fromiter(map(lambda x: x if x>0.0 else 0.0,lp.features.toArray()),dtype=np.int)\
               ))\
          )
modelDecisionTree = DecisionTree.trainClassifier(trainingData.map(lambda lp: LabeledPoint(lp.label,\
      np.fromiter(map(lambda x: 0.0 if np.isnan(x) else x,lp.features.toArray()),dtype=np.double )))\
      , numClasses=training.getNumClasses(), categoricalFeaturesInfo={})

modelRandomForest = RandomForest.trainClassifier(trainingData.map(lambda lp: LabeledPoint(lp.label,\
      np.fromiter(map(lambda x: 0.0 if np.isnan(x) else x,lp.features.toArray()),dtype=np.double )))\
      , numClasses=training.getNumClasses(), categoricalFeaturesInfo={},numTrees=100)

#reload the test labeled data
testData = training.loadLabeledDataRDD("test")

#recompute the accuracy metrics
training.displayConfusionTable=True
training.runMetrics(testData,modelNaiveBayes,modelDecisionTree,logRegModel,modelRandomForest)<jupyter_output><empty_output><jupyter_text>Run the predictive modelrunModel(departureAirportCode, departureDateTime, arrivalAirportCode, arrivalDateTime)  Note: all DateTime must use UTC format<jupyter_code>run.useModels(modelNaiveBayes,modelDecisionTree,logRegModel,modelRandomForest)
run.runModel('BOS', "2016-05-18 20:15-0500", 'AUS', "2016-05-18 22:30-0800" )<jupyter_output>2016-05-18 20:15:00-05:00
{'fraction_non_alphanumeric': 0.11837655016910936, 'fraction_numerical': 0.01939120631341601, 'mean_word_length': 5.32287954383464, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 37, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '609596', 'license': 'Apache-2.0', 'n_tokens_mistral': 3009, 'n_tokens_neox': 2864, 'n_words': 597, 'path': 'notebook/Flight Predict PyCon 2016.ipynb', 'repo_name': 'ibm-watson-data-lab/simple-data-pipe-connector-flightstats', 'size': '8900'}