Introduction
More often than not, data doesn't come packaged exactly as we'd like it for analysis. Transformation,
match-merge operations, and a host of data munging tasks
are usually needed before we can extract insights from our Big Data sources. Few people find data
munging exciting, but it has
to be done. Once we've suffered that boredom, we should take steps
to automate
the process. We want codify our work into repeatable units and
create
workflows which we can leverage over and over again without having
to write new code. In this article,
we'll look at
how to use Oozie
to create a workflow for the
parallel machine learning task I described on Cloudera's site.
Hive Actions: Prepping for Pig
In my parallel machine learning article, I use data from the
National Climatic Data Center
to build weather models
on a state-by-state basis. NCDC makes the data freely available as gzipped files of day-over-day observations
stretching from the 1930s
to today. In reading that post, one might get the impression that the data came
in a handy, ready-to-model files with convenient delimiters. The truth of it is that I need
to perform
some parsing and projection on the dataset before it can be modeled. If I get more observations, I'll want
to retrain and test those models, which will require more parsing and projection. This is a good opportunity
to start building up a workflow with Oozie.
I store the data from the NCDC in HDFS and
create an external Hive table
partitioned by year. This gives me flexibility of Hive's query language when I want it, but let's me put
the dataset in a directory of my choosing in case I want
to treat the same data with Pig or MapReduce code.
CREATE EXTERNAL TABLE IF NOT EXISTS historic_weather(column 1, column2)
PARTITIONED BY (yr string)
STORED AS ...
LOCATION '/user/oracle/weather/historic';
As new weather data comes in from NCDC, I'll need
to add partitions
to my table. That's an action I should put in the workflow.
Similarly, the weather data requires parsing
in order
to be useful as a set of columns. Because of their long history, the weather data is broken up into fields of specific
byte lengths: x bytes for the station ID, y bytes for the dew point, and so on. The delimiting is consistent from year
to year,
so writing SerDe or a parser for transformation is simple. Once that's done, I want
to select columns on which
to train,
classify certain features, and place the training data in an HDFS directory for my Pig script
to access.
ALTER TABLE historic_weather ADD IF NOT EXISTS PARTITION (yr='2010')
LOCATION '/user/oracle/weather/historic/yr=2011';
INSERT OVERWRITE DIRECTORY '/user/oracle/weather/cleaned_history'
SELECT w.stn, w.wban, w.weather_year, w.weather_month,
w.weather_day, w.temp, w.dewp, w.weather FROM (
FROM historic_weather SELECT TRANSFORM(...)
USING '/path/to/hive/filters/ncdc_parser.py'
as stn, wban, weather_year, weather_month, weather_day, temp, dewp, weather
) w;
Since I'm going
to prepare training directories with at least the same frequency that I add partitions, I should
also add that
to my workflow. Oozie is going
to invoke these Hive actions using what's somewhat obviously referred
to as
a Hive action. Hive actions amount
to Oozie running a script file containing our query language statements,
so we can place them in a file called weather_train.hql.
Starting Our Workflow
Oozie offers two types of jobs: workflows and coordinator jobs. Workflows are straightforward: they define a set of actions
to perform as a sequence or directed acyclic graph. Coordinator jobs can take all the same actions of Workflow jobs, but
they can be automatically started either periodically or when new data arrives in a specified location.
To keep things
simple we'll make a workflow job; coordinator jobs simply require another XML file for scheduling. The bare minimum
for workflow XML defines a name, a starting point, and an end point:
<workflow-app name="WeatherMan" xmlns="uri:oozie:workflow:0.1">
<start to="ParseNCDCData"/>
<end name="end"/>
</workflow-app>
To this we need
to add an action, and within that we'll specify the hive parameters Also, keep in mind that actions require
<ok> and <error> tags
to direct the next action on success or failure.
<action name="ParseNCDCData">
<hive xmlns="uri:oozie:hive-action:0.2">
<job-tracker>localhost:8021</job-tracker>
<name-node>localhost:8020</name-node>
<configuration>
<property>
<name>oozie.hive.defaults</name>
<value>/user/oracle/weather_ooze/hive-default.xml</value>
</property>
</configuration>
<script>ncdc_parse.hql</script>
</hive>
<ok to="WeatherMan"/>
<error to="end"/>
</action>
There are a couple of things
to note here:
I have
to give the FQDN (or IP) and port of my JobTracker and NameNode.
I have
to include a hive-default.xml file.
I have
to include a script file.
The hive-default.xml and script file must be stored in HDFS
That last point is particularly important. Oozie doesn't make assumptions about where a given workflow is being run. You might
submit workflows against different clusters, or have different hive-defaults.xml on different clusters (e.g. MySQL or Postgres-backed metastores).
A quick way
to ensure that all the assets end up in the right place in HDFS is just
to make a working directory locally, build your workflow.xml
in it, and copy the assets you'll need
to it as you add actions
to workflow.xml.
At this point, our local directory should contain:
workflow.xml
hive-defaults.xml (make sure this file contains your metastore connection data)
ncdc_parse.hql
Adding Pig
to the Ooze
Adding our Pig script as an action is slightly simpler from an XML standpoint. All we do is add an action
to workflow.xml as follows:
<action name="WeatherMan">
<pig>
<job-tracker>localhost:8021</job-tracker>
<name-node>localhost:8020</name-node>
<script>weather_train.pig</script>
</pig>
<ok to="end"/>
<error to="end"/>
</action>
Once we've done this, we'll copy weather_train.pig
to our working directory. However, there's a bit of a "gotcha" here.
My pig script registers the Weka Jar and a chunk of jython. If those aren't also in HDFS, our action will fail from the outset --
but where do we put them? The Jython script goes into the working directory at the same level as the pig script, because pig attempts
to
load Jython files in the directory from which the script executes. However, that's not where our Weka jar goes.
While Oozie doesn't assume much, it does make an assumption about the Pig classpath. Anything under working_directory/lib
gets automatically added
to the Pig classpath and no longer requires a REGISTER statement in the script. Anything that uses a REGISTER statement
cannot be in the working_directory/lib directory. Instead, it needs
to be in a different HDFS directory and attached
to the pig action with an
<archive> tag.
Yes, that's as confusing as you think it is.
You can get the exact rules for adding Jars
to the
distributed cache from Oozie's Pig Cookbook.
Making the Workflow Work
We've got a workflow defined and have collected all the components we'll need
to run. But we can't run anything yet, because we
still have
to define some properties about the job and submit it
to Oozie. We need
to start with the job properties, as this is essentially the
"request" we'll submit
to the Oozie server. In the same working directory, we'll make a file called job.properties as follows:
nameNode=hdfs://localhost:8020
jobTracker=localhost:8021
queueName=default
weatherRoot=weather_ooze
mapreduce.jobtracker.kerberos.principal=foo
dfs.namenode.kerberos.principal=foo
oozie.libpath=${nameNode}/user/oozie/share/lib
oozie.wf.application.path=${nameNode}/user/${user.name}/${weatherRoot}
outputDir=weather-ooze
While some of the pieces of the properties file are familiar (e.g., JobTracker address), others take a bit of explaining.
The first is weatherRoot: this is essentially an environment variable for the script (as are jobTracker and queueName). We're simply
using them
to simplify the directives for the Oozie job. The oozie.libpath pieces is extremely important. This is a directory
in HDFS which holds Oozie's shared libraries: a collection of Jars necessary for invoking Hive, Pig, and other actions. It's a good idea
to make sure
this has been installed and copied up
to HDFS. The last two lines are straightforward: run the application defined by workflow.xml at the application path listed and write the output
to the output directory.
We're finally ready
to submit our job! After all that work we only need
to do a few more things:
Validate our workflow.xml
Copy our working directory
to HDFS
Submit our job
to the Oozie server
Run our workflow
Let's do them in order. First validate the workflow:
oozie validate workflow.xml
Next, copy the working directory up
to HDFS:
hadoop fs -put working_dir /user/oracle/working_dir
Now we submit the job
to the Oozie server. We need
to ensure that we've got the correct URL for the Oozie server, and we need
to
specify our job.properties file as an argument.
oozie job -oozie http://url.
to.oozie.server:port_number/ -config /path/to/working_dir/job.properties -submit
We've submitted the job, but we don't see any activity on the JobTracker? All I got was this funny bit of output:
14-20120525161321-oozie-oracle
This is because submitting a job
to Oozie creates an
entry for the job and places it in PREP status. What we got back, in essence, is a ticket for our workflow
to ride the
Oozie train. We're responsible for redeeming our ticket and running the job.
oozie -oozie http://url.
to.oozie.server:port_number/ -start 14-20120525161321-oozie-oracle
Of course, if we really want
to run the job from the outset, we can change the "-submit" argument above
to "-run." This will prep and run the workflow immediately.
Takeaway
So, there you have it: the somewhat laborious process of building an Oozie workflow. It's a bit tedious the first time out, but it
does present a pair of real benefits
to those of us who spend a great deal of time data munging. First, when new data arrives that requires the same processing,
we already have the workflow defined and ready
to run. Second, as we build up a set of useful action definitions over time, creating new workflows becomes quicker and
quicker.
