Speed up hadoop development with progressive testing

Debugging Hadoop jobs can be a huge pain.  The cycle time is slow, and error messages are often uninformative --- especially if you're using Hadoop streaming, or working on EMR.

I once found myself trying to debug a job that took a full six hours to fail.  It took more than a week -- a whole week! -- to find and fix the problem.  Of course, I was doing other things at the same time, but the need to constantly check up on the status of the job was a huge drain on my energy and productivity.  It was a Very Bad Week.

Painful experiences like this have taught me to follow a test-driven approach to hadoop development.  Whenever I'm working on a new hadoop-based data pipe, my goal is to isolate six distinct kinds of problems that arise in hadoop development.

1. Explore the data: The pipe must accept data from a given format, which might not be fully understood at the outset.
2. Test basic logic: The pipe must execute the intended data transformation for "normal" data. 
3. Test edge cases: The pipe must deal gracefully with edge cases, missing or misformatted fields, rare divide-by-zeroes, etc. 
4. Test deployment parameters: The pipe must be deployable on hadoop, with all the right filenames, code dependencies, and permissions.
5. Test cluster performance: For big enough jobs, the pipe must run efficiently.  If not, we need to tune or scale up the cluster.
6. Test scheduling parameters: Once pipes are built, routine jobs must be scheduled and executed.

Each of these steps requires different test data and different methods for trapping and diagnosing errors.  Therefore, the goal is to make sure to (1) tackle problems one at a time, and (2) solve each kind of problem in the environment with the fastest cycle time.

Steps 1 through 3 should be solved locally, using progressively larger data sets.  Steps 4 and 5 must be run remotely, again using progressively larger data sets.

Step 6 depends on your scheduling system and has a very slow cycle time (i.e. you must wait a day to test whether your daily jobs run on the proper schedule.).  However, it's independent of hadoop, so you can build, test, and deploy it separately.  (There may be some crossover with #4, but you can test this with small data sets.)

Going through six different rounds of testing may seem like overkill, but in my experience it's absolutely worth it.  Very likely, you'll encounter at least one new bug/mistake/unanticipated case at each stage.  Progressive testing ensures that each bug is dealt with as quickly as possible, and prevents them from ganging up on you.

Other suggestions:

• Definitely use an abstraction layer that allows you to seamlessly deploy local code to your staging and production clusters.  Cascalog and mrJob are good examples.  Otherwise, you'll find yourself solving steps 2 and 3 all over again in deployment.

• Config files and object-oriented code can reduce a lot of headaches in step 4.  Most of your deployment hooks can be written once and saved in a config file.  If you have strong naming conventions, then most of your filenames can be constructed (and tested) programmatically.  It's amazing how many hours you can waste debugging a simple typo in hadoop.  Good OOP will spare you many of these headaches.

• Part of the beauty of Hive and HBase is that they abstract away most of the potential pitfalls on the deployment side, especially in step 4.  By the same token, tools like Azkaban and Oozie can take a lot of the pain out of step 6.  (Be careful, though -- each of these scheduling tools has its limitations.)

Hacking Scrabble with Cascalog

We've barely met, but I am in love with Cascalog.  So elegant.  So powerful.  So easy to ship from testing to staging to production.  So perfect for the workflow abstraction where most data science happens.  Let me count the ways...

The only downside is the shortage of documentation and examples. There's an active google group, yes, but only a handful of cascalog questions on StackOverflow.  So I thought I'd pay things forward by tossing out a bunch of toy examples that I used in my early experiments with the language.  These examples cover clojure's basic syntax and regular expressions, plus cascalog's filtering and basic aggregations.  I'll save tests, joins, etc. for a later post.

All these examples use the Scrabble Word dictionary, which provides a nice bite-sized playground for lots of MapReduced fun.  The Tournament Word List is a list of all the legal words for Scrabble in U.S. tournament play.  Words are in all caps and separated by line breaks.  I downloaded and saved the file as /Users/agong/Data/scrabble-list-twl06.txt.  To speed up testing, I also sampled rows at random to create a 10% sample (/Users/agong/Data/scrabble-list-twl06-1pct.txt) and a 1% sample (/Users/agong/Data/scrabble-list-twl06-1pct.txt).

First, let's try a vanilla query: just print all the lines in the file. This will help make sure clojure and hadoop are set up properly.

; vanilla query: print all lines
(?<- (stdout) [?word]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-1pct.txt") ?word))

If that works, we can do some basic filtering.

; basic filter: print lines starting with 'z'
(?<- (stdout) [?word]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-1pct.txt") :> ?word)
(clojure.core/re-matches #"Z.*" ?word) ))

So far so good. We have Hello world, and basic filters check out.  Let's use the count aggregation to count the number of words in the sample:

; count words
(?<- (stdout) [?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-1pct.txt") ?word)
(cascalog.ops/count ?count) ))

Now let's build up to letter counts (a twist on the classic MapReduce word count example.)  To get there, we need to define a map cat operator:

; split words into letters
; https://github.com/sritchie/cascalog-class/blob/master/src/cascalog_class/core.clj
(defmapcatop split
 "Accepts a word and emits a single 1-tuple for each letter."
 [word]
 (clojure.core/re-seq #"." word))

; count letters
(require `cascalog.ops)
(?<- (stdout) [?letter ?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-1pct.txt") ?word)
(split ?word :> ?letter) (cascalog.ops/count ?count))

With these very simple tools, we can do a surprising number of interesting things. Let's create a function to do n-gram counting, modified slightly from this example.

; count ngrams
(defmapcatop ngrams
 "Accepts a word and n-parameter and emits a single 1-tuple for each n-gram."
 [word n]
 (map my-join (partition n 1 word)))

Now we can count bigrams and trigrams:

; Character bigrams
(?<- (stdout) [?letter ?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-1pct.txt") ?word)
(ngrams ?word 2 :> ?letter) (cascalog.ops/count ?count))

; Character trigrams - no sampling
(?<- (stdout) [?letter ?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06.txt") ?word)
(ngrams ?word 3 :> ?letter) (cascalog.ops/count ?count))

Let's get word lengths...

; get word lengths
(defmapcatop get-len
 "Accepts a word and emits a single 1-tuple with its length."
 [word]
 [(.length word)])

; distribution of word lengths
(?<- (stdout) [?length ?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-1pct.txt") ?word)
(get-len ?word :> ?length)
(cascalog.ops/count ?count))

I guess it makes sense that the longest words in scrabble are 15 letters long...

Now let's combine filters and aggregators.  We need to create a filter operation for this...

(deffilterop len-n? [word n]
 "Keep only words not of length n"
 (= (.length word) n))

; distribution of lengths for 7-letter words (a silly example to make sure it worked)
(?<- (stdout) [?length ?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-1pct.txt") :> ?word)
(len-n? ?word 7)
(get-len ?word :> ?length)
(cascalog.ops/count ?count))

Vowel dumps are an important part of scrabble tactics: words that let you get rid of extra vowels without wasting a turn to exchange your hand.  First, we can do pure vowel dumps -- words that include no consonants at all.  (We'll grant Y vowel status, even though it only qualifies sometimes.)

; vowel dumps
(deffilterop pure-vowel-dump? [word]
 "Keep only words containing only vowels (and sometimes y)"
 (every?
  (into #{} (clojure.core/re-seq #"." "AEIOUY"))
  (clojure.core/re-seq #"." word)))

(?<- (stdout) [?word]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06.txt") :> ?word)
(pure-vowel-dump? ?word))

Hm. There aren't very many of these pure vowel dumps.  How about a more flexible function that calculates the proportion of vowels in the word?

(defmapop vowel-ratio [word]
(/
    (.length (clojure.string/replace word #"[AEIOUY]" ""))
    (.length word)))

Now we can look up words with 70% or more vowels.  For good measure, let's show the ratio of vowels in each.

;Ratios for vowel dumps
(?<- (stdout) [?word ?ratio]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-1pct.txt") :> ?word)
(vowel-ratio ?word :> ?ratio )
(> ?ratio 7/10))

Among these vowel dumps, what's the distribution of lengths?  The distribution of letters?

;Length distribution for vowel dumps
(?<- (stdout) [?length ?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06.txt") :> ?word)
(vowel-ratio ?word :> ?ratio )
(> ?ratio 7/10)
(get-len ?word :> ?length)
(cascalog.ops/count :> ?count))

;Letter distribution for vowel dumps
(?<- (stdout) [?letter ?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06.txt") :> ?word)
(vowel-ratio ?word :> ?ratio )
(> ?ratio 7/10)
(split ?word :> ?letter)
(cascalog.ops/count :> ?count))

Okay, a few more fun examples.  Palindromes...

;Palindromes
(deffilterop palindrome? [word]
 "Return true for palindromes"
 (= word (clojure.string/reverse word)) )

(?<- (stdout) [?word]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06.txt") :> ?word)
(palindrome? ?word))

;Distribution of letters in palindromes
(?<- (stdout) [?letter ?count]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06.txt") :> ?word)
(palindrome? ?word)
(split ?word :> ?letter)
(cascalog.ops/count ?count))

Not a whole lot of these either.  What about vowel-consonant palindromes?  That is, words where the back-to-front and front-to-back ordering of vowel consonants is the same?

;Vowel-consonants palindromes
(deffilterop vc-palindrome? [word]
 "Return true for vowel-consonant palindromes"
 (let [vc-word (clojure.string/replace (clojure.string/replace word #"[AEIOUY]" "A") #"[^A]" "B")]
  (= vc-word (clojure.string/reverse vc-word)) ))

(?<- (stdout) [?word]
((lfs-textline "/Users/agong/Data/scrabble-list-twl06-10pct.txt") :> ?word)
(vc-palindrome? ?word))

There we go!  From ANALYZE to ZYMOSIS.  (Your words will probably be different, because of the sampling.  But they will still be v/c palindromes. :) )

I had fun spending an afternoon putting together these examples.  It was a great, well-bounded way to get my feet wet with clojure and cascalog.

HTH

Python mapreduce on EC2

Last week, I wrote about getting AWS public datasets onto an EC2 cluster, and then into HDFS for MapReduce.  Now let's get to hello world (or rather, countWords) with python scripts.

#!/usr/bin/env python
# mapper2.py

import sys, re
 
for line in sys.stdin:
    line = line.lower()
    words = line.split()
 
    #--- output tuples [word, 1] in tab-delimited format---
    for word in words: 
        print '%s\t%s' % (word, "1")

Here's the reducer script....

#!/usr/bin/env python
# reducer.py
 
import sys
 
# maps words to their counts
word2count = {}
 
# input comes from STDIN
for line in sys.stdin:
    # remove leading and trailing whitespace
    line = line.strip()
 
    # parse the input we got from mapper.py
    word, count = line.split('\t', 1)
    # convert count (currently a string) to int
    try:
        count = int(count)
    except ValueError:
        continue
 
    try:
        word2count[word] = word2count[word]+count
    except:
        word2count[word] = count
 
# write the tuples to stdout
# Note: they are unsorted
for word in word2count.keys():
    print '%s\t%s'% ( word, word2count[word] )

The command to execute all this in hadoop is a bit of a monster, mainly because of all the filepaths.  Note the usage of the -file parameter, which tells hadoop to load files for use in the -mapper and -reducer arguments. Also, I set -jobconf compression to false, because I didn't have a handy LZO decompresser installed.

bin/hadoop jar contrib/streaming/hadoop-0.19.0-streaming.jar -input wex-data -output output/run9 -file /usr/local/hadoop-0.19.0/my_scripts/mapper2.py -file /usr/local/hadoop-0.19.0/my_scripts/reducer.py -mapper mapper2.py -reducer reducer.py -jobconf mapred.output.compress=false

NB: As I dug into this task, I discovered several pretty good python/hadoop-streaming tutorials online.  The scripts here were modified from:  http://cs.smith.edu/dftwiki/index.php/Hadoop_Tutorial_3.2_--_Using_Your_Own_Streaming_WordCount_program


Other sources:

    http://www.protocolostomy.com/2008/03/20/hadoop-ec2-s3-and-me/
    http://www.michael-noll.com/tutorials/writing-an-hadoop-mapreduce-program-in-python/
    http://www.larsgeorge.com/2010/10/hadoop-on-ec2-primer.html
    http://www.princesspolymath.com/princess_polymath/?p=137
    http://arunxjacob.blogspot.com/2009/04/configuring-hadoop-cluster-on-ec2.html

    http://wiki.apache.org/hadoop/AmazonS3

    http://oobaloo.co.uk/articles/2010/1/20/processing-xml-in-hadoop.html

    http://www.cloudera.com/blog/2009/07/advice-on-qa-testing-your-mapreduce-jobs/
    http://hadoop.apache.org/common/docs/r0.20.2/streaming.html

Running mapreduce on Amazon's publicly available datasets with python

on Monday, I had a preliminary interview at a really interesting tech startup. In the course of the conversation, the interviewer mentioned that he'd used some of the technical notes from compSocSci in his own work.  And I thought nobody was reading!

Anyway, I've been sitting on some old EC2/hadoop/python notes for a while.  The talk gave me the motivation to clean up and post them, just in case they can help somebody else.  The goal here is threefold:

1. Fire up a hadoop cluster on EC2
2. Import data from an EBS volume with one of AWS' public data sets
3. Use hadoop streaming and python for quick scripting

In other words, we want to set up a tidy, scalable data pipeline as fast as possible.  My target project is to do word counts on wikipedia pages -- the classic "hello world" of mapReduce.  This isn't super-hard, but I haven't seen a good soup-to-nuts guide that brings all of these things together.

Phase 1:
Follow the notes below to get to the digits-of-pi test.  Except for a little trouble with AWS keys, this all went swimmingly, so I see no need to duplicate.  If you run into trouble with this part, we can troubleshoot in the comments.

    http://wiki.apache.org/hadoop/AmazonEC2#Running_a_job_on_a_cluster
    http://www.michael-noll.com/tutorials/writing-an-hadoop-mapreduce-program-in-python/

Phase 2:
Now let's attach an external dataset.  Here's the dataset we'll use:  Wikipedia Extraction (WEX).  It's a processed dump of the English language Wikipedia, hosted publicly on Amazon Web Services under snapshot ID snap-­1781757e.

This dataset contains a dump of 1,000 popular English wikipedia articles.  It's about 70GB.  At Amazon's $.12/GB rate, maintaining this volume costs about $8 for a whole month -- cheap!  If you want to scale up to full-size wikipedia (~500GB), you can do that too.  After all, we're in big data land.

Here's the command sequence to create an EBS volume for this snapshot and attach it to an instance. You can look up the ids using ec2-describe-volumes and ec2-describe-instances, or get them from the AWS console at https://console.aws.amazon.com.  (Hint: they're not vol-aaaaaaaa and i-bbbbbbbbb.)

    ec2-create-volume -snapshot snap-1781757e -z us-east-1a
    ec2-attach-volume vol-aaaaaaaa -i i-bbbbbbbb -d /dev/sdf

It took a while for these commands to execute.  Attaching the volume got stuck in "attaching" status for several minutes.  I finally got tired of waiting and mounted the volume, and then the status switched right away.  Can't say whether that was cause-and-effect or coincidence, but it worked.

Once you've attached the EBS volume, login to the instance (instructions here) and mount the volume as follows. This should be pretty much instantaneous.


    mkdir /mnt/wex_data
    mount /dev/sdf /mnt/wex_data

Now import the data into the Hadoop file system:

    cd /usr/local/hadoop/
    hadoop fs -copyFromLocal /mnt/wex_data/rawd/freebase-wex-2009-01-12-articles.tsv wex-data

If you want, you can now remove and delete the EBS volume.  The articles file is stored in the distributed filesystem across  your EC2 instances in  you hadoop cluster.  The nice thing is that you can get to this point within less than an hour, meaning that you only have to pay a tiny fraction of the monthly storage cost.


    ec2-detach-volume vol-aaaaaaaa -i i-bbbbbbbbb -d /dev/sdf
    ec2-delete-volume vol-aaaaaaaa

I had some trouble detaching volumes until I used the force flag: -f.  Maybe I was just being impatient again.

That's enough for the moment.  I'll tackle python in my next post.

Key skills for job-hunting data scientists

There's a lot of buzz around data science, but (as I've posted about previously) the term is still murky.

One way to get a look at the emerging definition of data science is to search for "data science" jobs, and see what they have in common: What are the key skills for data scientists?

This wordle sounds like Romney: "jobs jobs jobs..."

I took a few minutes today to run that search -- automated, of course.  Nothing rigorous or scientific, but the results are still plenty interesting.

Methods:  I searched "data scientist" on simplyhired.com, then scraped the ~250 resulting links.  All of the non-dead links (there weren't many dead ones) returned a job posting, usually on a company page, occasionally on another aggregator.  I grabbed the html of each of these pages, and cleaned the html to get rid of scripts, styles, etc.  I didn't do any fancy chrome or ad scraping, so take the results with a grain of salt.

First, I generated the obligatory word cloud.  Thank you, wordle.


Then I skimmed a dozen of the pages, looking for keywords that seemed to pop up a lot: java, hadoop, python.  For the most part, I focused on specific skills that companies are explicitly hiring for.  I also tossed in a few other terms, just to see what would happen.

Here are counts of jobs mentioning keywords (Not keyword counts -- the number of separate job postings that include at least one reference to a given keyword.):

198	data
131	statist
130	java
107	hadoop
85	mining
68	python
49	visuali
39	cloud
37	mapreduce
35	c\+\+
24	amazon
22	ruby
18	bayes
15	ec2
13	jquery
13	fun
3	estimat

Evidently, data science postings put equal value on "fun" and "jquery."

Also, at a glance, Java beats python, beats C++ in terms of employability.  It kind of makes me wish I'd been nicer to Java all these years.

One clear finding is that hadoop and MapReduce skills are in high demand.  That's not news to anyone working in this area, but I was surprised at just how many jobs were looking for these skills.  Almost half (107 of 238, 45%) of total job postings explicitly mention hadoop.

That percentage seems slightly out of whack to me, because there are plenty of valuable ways to mine data without using a MapReduce algorithm.  Maybe Hadoop is a pinch point in the job market because there just aren't enough MapReduce-literate data miners out there?  If that's the case, I would expect demand to come down (relative to supply) in the not-too-distant future -- MapReduce isn't that hard to learn.

Alternatively, there could be some bandwaggoning going on:

"Data is the Next Big Thing.  We need to hire a data person."

"What exactly is a data person?"

"I don't know, but I hear they know how to program in 'ha-doop,' so put that in the posting."

As a third explanation, it may be that the meaning of "data science" is narrowing.  Instead of encompassing all the things that can be done with data, perhaps it's coming to mean "mapreduce."  If that's the case, then "data science" jobs would naturally include hadoop/mapreduce skills.  IMO, that would be sad, because it would be an opportunity missed to change the way data flows into decisions in a more systemic way.

I'd be interested in hearing other explanations for the dominance of hadoop.  Also, if you have other queries to run against this data set, I'm happy to try them out.  What I've put up so far is just back-of-envelope coffee-break stuff.