Combing Pig and Python to explore raw datasets via Pydoop¶
Taking the raw data processing capabilities of Hadoop and seamlessly integrating that into an python notebook to start doing from exploratory data analysis directly from my raw electricity dataset.
Loading the libaries¶
import pandas as pd
import sys
import numpy as np
import math
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = "all"
Write the pig script out to the server (this gets saved in the working directory of the ipython notebook). The purpose of this script is to take the raw json files that get generated every day from my logger (one for each day) and turns it into something that can be read into python. I also convert the date into something useful and create columns to represnt the hour and day of week to help the anlaysis. It then writes this all out to csv file(s). This will create a directory called electricty.csv and in that directory a number of csv files 1 for each split in the map reduce.
%%writefile ElectricityReadings.pig
REGISTER '/edge/elephant-bird-pig-4.3.jar';
REGISTER '/edge/elephant-bird-hadoop-compat-4.3.jar';
REGISTER '/edge/json-simple-1.1.jar';
A = LOAD '/user/hive/warehouse/sarafand.db/stg_raw_loop/*.json'
USING com.twitter.elephantbird.pig.load.JsonLoader() as (json:map[]);
B = FOREACH A GENERATE ToDate((long)ROUND_TO((double)$0#'Current Time'*1000,0)) as TimeStamp,
$0#'Type' AS Type,
$0#'Electricity Reading' AS ElectricityReading;
C = FILTER B By ((chararray)Type == 'Electricity');
D = FOREACH C GENERATE
TimeStamp,
GetHour(TimeStamp) as hour,
ToString( TimeStamp, 'EEE' ) as DayofWeek,
ElectricityReading;
store D INTO '/user/olloyd/Pig Learning/electricty.csv'
USING PigStorage(',');
The job will fail if a previous run has been performed so we need to delete these. This just runs a command on the analysis server to delete the electricty.csv directory from the hadoop file system.
%%bash
hdfs dfs -rm -r '/user/olloyd/Pig Learning/electricty.csv'
Now run the ElectricityReadings pig scipt created above. The second block of code is optional and allows you to see the progress of the job as it runs.
%%bash --err pig_out --bg
pig -f ElectricityReadings.pig
while True:
line = pig_out.readline()
if not line:
break
sys.stdout.write("%s" % line)
sys.stdout.flush()
Pydoop has a built in funtion that allows us to read in data directly from the hadoop file system. As the map reduce program outputs a number of files rather than a single csv we need to loop through them for each one append them into a pandas dataframe ready for anlaysis in python.
import pydoop.hdfs as hdfs
def read_csv_from_hdfs(path, cols, col_types=None):
files = hdfs.ls(path);
pieces = []
for f in files:
fhandle = hdfs.open(f)
pieces.append(pd.read_csv(fhandle, names=cols, dtype=col_types))
fhandle.close()
return pd.concat(pieces, ignore_index=True)
ElecReadings = read_csv_from_hdfs('/user/olloyd/Pig Learning/electricty.csv',["TimeStamp","Hour","DayofWeek","ElectricityReading"])
As this step can take a few minutes and I might want to pick up this analysis again later I write out the data to the pickle jar and then another statement to read it. OK I know its not really called the pickle jar but seriously.. it should be.
ElecReadings.to_pickle('ElecReadings')
ElecReadings = pd.read_pickle('ElecReadings')
Now we can take a quick look and see what we are dealing with. My pandas dataframe isnt too large - just over 860K records with just 4 columns.
ElecReadings.shape
To take a look at the column names we can either list them out or use the .columns attribute in pandas.
list(ElecReadings)
ElecReadings.columns
No we want to start taking a quick look at the data itself. We can the .values attribute we can get a Numpy representation of the data frame.
ElecReadings.values
I personally dont find that very useful when trying to get a feel for the data. Instead I use the commands below. Head and Tail show the top and bottom n records where as sample will show as a random number of records taken from across the whole dataframe,
ElecReadings.head(5)
ElecReadings.tail(10)
ElecReadings.sample(10)
We can also get some basic statistics
ElecReadings.describe()
These figures are shown as watts per hour. So my maximum usage in my house so far has been just over 8kwH. We must have had every appliance on at the time! We can filter the records to take a look.
ElecReadings.query('ElectricityReading > 8e+3')
We could equally have run the below. Results are excactly the same.
ElecReadings[ElecReadings.ElectricityReading > 8e+3]
Null values can cause a lot of issues so I just want to check to see if my dataset has any that I need to deal with.
ElecReadings[ElecReadings.ElectricityReading.isnull()]
Luckly the data I have collected so far doesnt have any.
From the describe a few steps about it looks like 75% of my readings are below 500wH but there is a large gap between this and my maximum reading of 8kwH. I will create a subset of data just looking above the 500W mark to get some flavour of the higer readings.
HighElecReadings = ElecReadings.query('ElectricityReading > 5e+2')
HighElecReadings.sample(10)
HighElecReadings.describe()
So it looks like the higer readings are quite rare with the distibution definitely skewed towards the lower end. Not unsurprising if you think about it - how many times a day do you actually boil a kettle?
Basic statistics can help a little bit but really we need to plot some of this out to get a real understanding of the data. matplotlib is a good libary for doing 2D plots and as I also use ggplot within R I'm really happy that is it able to mirror the style.
import matplotlib.pyplot as plt
plt.style.use('ggplot')
%matplotlib inline
fig1 = plt.hist((ElecReadings['ElectricityReading']), bins= np.arange(min(ElecReadings['ElectricityReading'])
,max(ElecReadings['ElectricityReading']) + 100,100))
Yup definitely skewed! We need to split this into 2 charts. The first looking in more detail at the lower readings by limiting the x axis to values between 0 and 1kwH and shrinking the bin size to 10. The other looking at values over 1000kwH
fig2 = plt.hist((ElecReadings['ElectricityReading']), bins= np.arange(min(ElecReadings['ElectricityReading'])
,max(ElecReadings['ElectricityReading']) + 10,10));plt.xlim(0,1000)
plt.show()
fig3= plt.hist((ElecReadings['ElectricityReading']), bins= np.arange(min(ElecReadings['ElectricityReading'])
,max(ElecReadings['ElectricityReading']) + 100,100));plt.xlim(1000);plt.ylim(0,8000)
plt.show()
The second chart we start seeing a series of peaks every 1.5 KwH or so. We could get a little bit more insight maybe by change the y axis scale to a logarithmic one.
fig4 = plt.hist((ElecReadings['ElectricityReading']), bins= np.arange(min(ElecReadings['ElectricityReading'])
,max(ElecReadings['ElectricityReading']) + 100,100));plt.yscale('log')
There is definitely a pattern of peaks every 1.5 KwH. 1.5, 3, 4.5, 6 and just below 7.5 KwH. It just so happens that our kettle is a 1.5KwH one.. And Anita does like her Tea. I suspect if I look around we might find a number of our appliances are around the 1.5 KwH mark.
Summary¶
What we have done here is to take raw data directly from my Utility Sensor log files, process them on the hadoop cluster using pig and do some basic exploratory data anlysis using python and the matplotlib libaries. We have identified a pattern in the data with peaks every 1.5kwH that I balme on Anita's Tea habit.In another post I will look in a little more detail as well as use Machine Learning to understand what variables have an impact on our electricity usage. Hopefully settle an age old argument in the house about uses the most energy.. or most likely reignite it!