1. Define Spark Context

sc

<pyspark.context.SparkContext at 0x102cea1d0>

2. Start H2O Context

    from pysparkling import *
    sc
    hc= H2OContext(sc).start()

    Warning: Version mismatch. H2O is version 3.6.0.2, but the python package is version 3.7.0.99999.

3. Define H2O Context

    hc

    H2OContext: ip=172.16.2.98, port=54329

4. Import H2O Python library

import h2o

5. View all available H2O Python functions

#dir(h2o)

6. Parse Chicago Crime dataset into H2O

column_type = ['Numeric','String','String','Enum','Enum','Enum','Enum','Enum','Enum','Enum','Numeric','Numeric','Numeric','Numeric','Enum','Numeric','Numeric','Numeric','Enum','Numeric','Numeric','Enum']
f_crimes = h2o.import_file(path ="../data/chicagoCrimes10k.csv",col_types =column_type)

print(f_crimes.shape)
f_crimes.summary()


Parse Progress: [##################################################] 100%
(9999, 22)

7. Look at the distribution of the IUCR column

f_crimes["IUCR"].table()

8. Look at the distribution of the Arrest column

f_crimes["Arrest"].table()

9. Modify column names to replace blank spaces with underscores

col_names = map(lambda s: s.replace(' ', '_'), f_crimes.col_names)
f_crimes.set_names(col_names)

10. Set time zone to UTC for date manipulation

h2o.set_timezone("Etc/UTC")

11. Refine the date column

def refine_date_col(data, col, pattern):
    data[col]         = data[col].as_date(pattern)
    data["Day"]       = data[col].day()
    data["Month"]     = data[col].month()    # Since H2O indexes from 0
    data["Year"]      = data[col].year()
    data["WeekNum"]   = data[col].week()
    data["WeekDay"]   = data[col].dayOfWeek()
    data["HourOfDay"] = data[col].hour()

    # Create weekend and season cols
    data["Weekend"] = (data["WeekDay"] == "Sun" or data["WeekDay"] == "Sat").ifelse(1, 0)[0]
    data["Season"] = data["Month"].cut([0, 2, 5, 7, 10, 12], ["Winter", "Spring", "Summer", "Autumn", "Winter"])

refine_date_col(f_crimes, "Date", "%m/%d/%Y %I:%M:%S %p")
f_crimes = f_crimes.drop("Date")

12. Parse Census data into H2O

f_census = h2o.import_file("../data/chicagoCensus.csv",header=1)

## Update column names in the table
col_names = map(lambda s: s.strip().replace(' ', '_'), f_census.col_names)
f_census.set_names(col_names)
f_census = f_census[1:78,:]
print(f_census.dim)
#f_census.summary()


Parse Progress: [##################################################] 100%
[77, 9]

13. Parse Weather data into H2O

f_weather = h2o.import_file("../data/chicagoAllWeather.csv")
f_weather = f_weather[1:]
print(f_weather.dim)
#f_weather.summary()


Parse Progress: [##################################################] 100%
[5162, 6]

14. Look at all the null entires in the Weather table

f_weather[f_weather["meanTemp"].isna()]

15. Look at the help on as_h2o_frame

hc.as_spark_frame?
f_weather

H2OContext: ip=172.16.2.98, port=54329

16. Copy data frames to Spark from H2O

df_weather = hc.as_spark_frame(f_weather,)
df_census = hc.as_spark_frame(f_census)
df_crimes = hc.as_spark_frame(f_crimes)

17. Look at the weather data as parsed in Spark

(only showing top 2 rows)

df_weather.show(2)

+-----+---+----+-------+--------+-------+
|month|day|year|maxTemp|meanTemp|minTemp|
+-----+---+----+-------+--------+-------+
|    1|  1|2001|     23|      14|      6|
|    1|  2|2001|     18|      12|      6|
+-----+---+----+-------+--------+-------+

18. Join columns from Crime, Census and Weather DataFrames in Spark

## Register DataFrames as tables in SQL context
sqlContext.registerDataFrameAsTable(df_weather, "chicagoWeather")
sqlContext.registerDataFrameAsTable(df_census, "chicagoCensus")
sqlContext.registerDataFrameAsTable(df_crimes, "chicagoCrime")


crimeWithWeather = sqlContext.sql("""SELECT
a.Year, a.Month, a.Day, a.WeekNum, a.HourOfDay, a.Weekend, a.Season, a.WeekDay,
a.IUCR, a.Primary_Type, a.Location_Description, a.Community_Area, a.District,
a.Arrest, a.Domestic, a.Beat, a.Ward, a.FBI_Code,
b.minTemp, b.maxTemp, b.meanTemp,
c.PERCENT_AGED_UNDER_18_OR_OVER_64, c.PER_CAPITA_INCOME, c.HARDSHIP_INDEX,
c.PERCENT_OF_HOUSING_CROWDED, c.PERCENT_HOUSEHOLDS_BELOW_POVERTY,
c.PERCENT_AGED_16__UNEMPLOYED, c.PERCENT_AGED_25__WITHOUT_HIGH_SCHOOL_DIPLOMA
FROM chicagoCrime a
JOIN chicagoWeather b
ON a.Year = b.year AND a.Month = b.month AND a.Day = b.day
JOIN chicagoCensus c
ON a.Community_Area = c.Community_Area_Number""")

19. Print the crimeWithWeather data table from Spark

crimeWithWeather.show(2)

+----+-----+---+-------+---------+-------+------+-------+----+-----------------+--------------------+--------------+--------+------+--------+----+----+--------+-------+-------+--------+--------------------------------+-----------------+--------------+--------------------------+--------------------------------+---------------------------+--------------------------------------------+
|Year|Month|Day|WeekNum|HourOfDay|Weekend|Season|WeekDay|IUCR|     Primary_Type|Location_Description|Community_Area|District|Arrest|Domestic|Beat|Ward|FBI_Code|minTemp|maxTemp|meanTemp|PERCENT_AGED_UNDER_18_OR_OVER_64|PER_CAPITA_INCOME|HARDSHIP_INDEX|PERCENT_OF_HOUSING_CROWDED|PERCENT_HOUSEHOLDS_BELOW_POVERTY|PERCENT_AGED_16__UNEMPLOYED|PERCENT_AGED_25__WITHOUT_HIGH_SCHOOL_DIPLOMA|
+----+-----+---+-------+---------+-------+------+-------+----+-----------------+--------------------+--------------+--------+------+--------+----+----+--------+-------+-------+--------+--------------------------------+-----------------+--------------+--------------------------+--------------------------------+---------------------------+--------------------------------------------+
|2015|    1| 23|      4|       22|      0|Winter|    Fri|143A|WEAPONS VIOLATION|               ALLEY|            31|      12|  true|   false|1234|  25|      15|     29|     31|      30|                            32.6|            16444|            76|         9.600000000000001|                            25.8|                       15.8|                                        40.7|
|2015|    1| 23|      4|       19|      0|Winter|    Fri|4625|    OTHER OFFENSE|            SIDEWALK|            31|      10|  true|   false|1034|  25|      26|     29|     31|      30|                            32.6|            16444|            76|         9.600000000000001|                            25.8|                       15.8|                                        40.7|
+----+-----+---+-------+---------+-------+------+-------+----+-----------------+--------------------+--------------+--------+------+--------+----+----+--------+-------+-------+--------+--------------------------------+-----------------+--------------+--------------------------+--------------------------------+---------------------------+--------------------------------------------+
only showing top 2 rows

20. Copy table from Spark to H2O

hc.as_h2o_frame?
crimeWithWeatherHF = hc.as_h2o_frame(crimeWithWeather,framename="crimeWithWeather")

H2OContext: ip=172.16.2.98, port=54329



crimeWithWeatherHF.summary()

21. Assign column types to the CrimeWeatherHF data table in H2O

crimeWithWeatherHF["Season"]= crimeWithWeatherHF["Season"].asfactor()
crimeWithWeatherHF["WeekDay"]= crimeWithWeatherHF["WeekDay"].asfactor()
crimeWithWeatherHF["IUCR"]= crimeWithWeatherHF["IUCR"].asfactor()
crimeWithWeatherHF["Primary_Type"]= crimeWithWeatherHF["Primary_Type"].asfactor()
crimeWithWeatherHF["Location_Description"]= crimeWithWeatherHF["Location_Description"].asfactor()
crimeWithWeatherHF["Arrest"]= crimeWithWeatherHF["Arrest"].asfactor()
crimeWithWeatherHF["Domestic"]= crimeWithWeatherHF["Domestic"].asfactor()
crimeWithWeatherHF["FBI_Code"]= crimeWithWeatherHF["FBI_Code"].asfactor()
crimeWithWeatherHF["Season"]= crimeWithWeatherHF["Season"].asfactor()



crimeWithWeatherHF.summary()

22. Split final H2O data table into train test and validation sets

ratios = [0.6,0.2]
frs = crimeWithWeatherHF.split_frame(ratios,seed=12345)
train = frs[0]
train.frame_id = "Train"
valid = frs[2]
valid.frame_id = "Validation"
test = frs[1]
test.frame_id = "Test"

23. Import Model Builders from H2O Python

from h2o.estimators.gbm import H2OGradientBoostingEstimator
from h2o.estimators.deeplearning import H2ODeepLearningEstimator

24. Inspect the availble GBM parameters

H2OGradientBoostingEstimator?

25. Define Predictors

predictors = crimeWithWeatherHF.names[:]
response = "Arrest"
predictors.remove(response)

26. Create a Simple GBM model to Predict Arrests

model_gbm = H2OGradientBoostingEstimator(ntrees         =50,
                                        max_depth      =6,
                                        learn_rate     =0.1, 
                                        #nfolds         =2,
                                        distribution   ="bernoulli")

model_gbm.train(x               =predictors,
               y               ="Arrest",
               training_frame  =train,
               validation_frame=valid
               )

27. Create a Simple Deep Learning model to Predict Arrests

model_dl = H2ODeepLearningEstimator(variable_importances=True,
                                   loss                ="Automatic")

model_dl.train(x                =predictors,
              y                ="Arrest",
              training_frame  =train,
              validation_frame=valid)


gbm Model Build Progress: [##################################################] 100%

deeplearning Model Build Progress: [##################################################] 100%

28. Print confusion matrices for the training and validation datasets

print(model_gbm.confusion_matrix(train = True))
print(model_gbm.confusion_matrix(valid = True))

Confusion Matrix (Act/Pred) for max f1 @ threshold = 0.335827722991:

Confusion Matrix (Act/Pred) for max f1 @ threshold = 0.432844055866:

print(model_gbm.auc(train=True))
print(model_gbm.auc(valid=True))
model_gbm.plot(metric="AUC")

0.974667176776
0.92596751276

29. Print variable importances

model_gbm.varimp(True)

30. Inspect Deep Learning model output

model_dl

Model Details
=============
H2ODeepLearningEstimator :  Deep Learning
Model Key:  DeepLearning_model_python_1446861372065_4

Status of Neuron Layers: predicting Arrest, 2-class classification, bernoulli distribution, CrossEntropy loss, 118,802 weights/biases, 1.4 MB, 72,478 training samples, mini-batch size 1

ModelMetricsBinomial: deeplearning

Reported on train data.

MSE: 0.0737426129728
R^2: 0.642891439669
LogLoss: 0.242051500943
AUC: 0.950131166302
Gini: 0.900262332604

Confusion Matrix (Act/Pred) for max f1 @ threshold = 0.343997370612:

Maximum Metrics: Maximum metrics at their respective thresholds

ModelMetricsBinomial: deeplearning
** Reported on validation data. **

MSE: 0.0843305429737
R^2: 0.593831388139
LogLoss: 0.280203809486
AUC: 0.930515181213
Gini: 0.861030362427

Confusion Matrix (Act/Pred) for max f1 @ threshold = 0.493462351545:

Maximum Metrics: Maximum metrics at their respective thresholds

Scoring History:

Variable Importances:

31. Predict on the test set using the GBM model

predictions = model_gbm.predict(test)
predictions.show()

32. Look at test set performance (if it includes true labels)

test_performance = model_gbm.model_performance(test)
test_performance


ModelMetricsBinomial: gbm
** Reported on test data. **

MSE: 0.0893676876445
R^2: 0.57094394422
LogLoss: 0.294019576922
AUC: 0.922152238508
Gini: 0.844304477016

Confusion Matrix (Act/Pred) for max f1 @ threshold = 0.365461652105:

Maximum Metrics: Maximum metrics at their respective thresholds

33. Create Plots of Crime type vs Arrest Rate and Proportion of reported Crime

# Create table to report Crimetype, Arrest count per crime, total reported count per Crime  
sqlContext.registerDataFrameAsTable(df_crimes, "df_crimes")
allCrimes = sqlContext.sql("""SELECT Primary_Type, count(*) as all_count FROM df_crimes GROUP BY Primary_Type""")
crimesWithArrest = sqlContext.sql("SELECT Primary_Type, count(*) as crime_count FROM chicagoCrime WHERE Arrest = 'true' GROUP BY Primary_Type")

sqlContext.registerDataFrameAsTable(crimesWithArrest, "crimesWithArrest")
sqlContext.registerDataFrameAsTable(allCrimes, "allCrimes")

crime_type = sqlContext.sql("Select a.Primary_Type as Crime_Type, a.crime_count, b.all_count \
FROM crimesWithArrest a \
JOIN allCrimes b \
ON a.Primary_Type = b.Primary_Type ")

crime_type.show(12)

+--------------------+-----------+---------+
|          Crime_Type|crime_count|all_count|
+--------------------+-----------+---------+
|       OTHER OFFENSE|        183|      720|
|   WEAPONS VIOLATION|         96|      118|
|  DECEPTIVE PRACTICE|         25|      445|
|            BURGLARY|         14|      458|
|             BATTERY|        432|     1851|
|             ROBBERY|         17|      357|
| MOTOR VEHICLE THEFT|         17|      414|
|        PROSTITUTION|        106|      106|
|     CRIMINAL DAMAGE|         76|     1003|
|          KIDNAPPING|          1|        7|
|            GAMBLING|          3|        3|
|LIQUOR LAW VIOLATION|         12|       12|
+--------------------+-----------+---------+
only showing top 12 rows

34. Copy Crime_type table from Spark to H2O

crime_typeHF = hc.as_h2o_frame(crime_type,framename="crime_type")

35. Create Additional columns Arrest_rate and Crime_propotion

crime_typeHF["Arrest_rate"] = crime_typeHF["crime_count"]/crime_typeHF["all_count"]
crime_typeHF["Crime_proportion"] = crime_typeHF["all_count"]/crime_typeHF["all_count"].sum()
crime_typeHF["Crime_Type"] = crime_typeHF["Crime_Type"].asfactor()
# h2o.assign(crime_typeHF,crime_type)
crime_typeHF.frame_id = "Crime_type"


crime_typeHF

hc


H2OContext: ip=172.16.2.98, port=54329

36. Plot in Flow

plot (g) -> g(
  g.rect(
    g.position "Crime_Type", "Arrest_rate"
    g.fillColor g.value 'blue'
    g.fillOpacity g.value 0.75
  )
  g.rect(
    g.position "Crime_Type", "Crime_proportion"
    g.fillColor g.value 'red'
    g.fillOpacity g.value 0.65
  )
  g.from inspect "data", getFrame "Crime_type"
)


#hc.stop()