We’re on a quest to write a scalable web crawler. Our goal is to build a web crawler that will download a billion pages a week. We’ve calculated that to download a billion pages in a week we need to sustain a rate of 1653 pages per second .

The problem with these kinds of numbers is that, unless you are familiar with web-crawling, it is not obvious how fast that really is. How fast can a simple crawler go? 10 pages per second? A thousand?

We set out to benchmark the simplest thing that could possibly work: wget and xargs.

Our Tools

wget is a popular tool used for downloading files from the web. It has a flexible set of options and built in support for crawling.

xargs is used to run a command repeatedly over a given set of inputs. In our case, we’re using a fixed URL list as our input. We use xargs as our “thread-pool” (it’s actually a “process-pool”). Using the -P <numprocs> option. xargs will run through the input file of URLs and each wget process will take a URL off the stack and run until it finishes the crawl for that domain. The number of concurrent processes is limited by <numprocs>.

Napkin Calculations

Before we actually run our jobs, let’s try to predict the kind of results we’ll get. I’m running the jobs on my MacBookPro Intel Core 2 Duo with 4GB RAM. I’m on my home network where I have AT&T U-Verse with advertised download speed of 18Mbps (mega bits per second).

wget measures rate limiting in kilobytes rather than kilobits, so we’ll use bytes rather than bits:

18Mbps = 2.25 megabytes per second =~ 2300 kilobytes/s

We’re just doing rough calculations at this point, so lets just guess that the mean size of each page is 10KB. At this page size, the absolute best number we can expect to get is around 230 pages/second before we saturate my connection.

Politeness

While we want our crawler as a system to go as fast as possible, we don’t want to hit any one server too many times. Not only might we get banned, but we it isn’t kind to the site owners. Many smaller servers can’t handle the load of a crawler thrown against it at full speed.

So if we want to get to 200+ pages/sec we’re going to have to have many concurrent connections. wget has a number of options that we can set to be nicer to each individual server. So our strategy will be to crawl many servers concurrently, but only hit a particular server lightly.

Here are a few of the relevant wget options we will set:

• --wait=2 and --random-wait – wait a random amount of time between requests averaging 2 seconds. The waiting time is for the servers benefit but the random time is for ours. Given that we are going to be running a large number of processes in parallel, we’d rather have them be out of step with each other.
• --tries=5 – only retry 5 times
• --timestamping – If the file exists on disk, send the server a HEAD request and check the Last-Modified header. If the file on disk has a timestamp greater than or equal to the Last-Modified date, we don’t request the whole page. This extra HEAD request doesn’t really slow us down because wget will only request it if the file already exists on disk. This is just a little extra protection in case our separate processes start to visit the same sites.

DMOZ Sample Set

For multiple runs of our test we don’t want to hit one particular server repeatedly. We’re going to use DMOZ to get a random sample of URLs to test and use a few commands to extract some random URLs:

mkdir -p data/dmoz
curl -0 http://rdf.dmoz.org/rdf/content.rdf.u8.gz > data/dmoz/dmoz-content.rdf.u8.gz
cd data/dmoz
unzip data/dmoz/dmoz-content.rdf.u8.gz
cat dmoz-content.rdf.u8 | grep http | grep r:resource | \
    grep -o '<link r:resource=['"'"'"][^"'"'"']*['"'"'"]' | \
    sed -e 's/^<link r:resource=["'"'"']//' -e 's/["'"'"']$//' \
    > urls.txt
ruby random-lines.rb urls.txt 300 > random-urls.txt

(You can get the random-lines.rb script here).

The DMOZ file is around 300MB so this will take a few minutes. The DMOZ RDF file is well formed, so we’re just using grep and sed to extract the URLs.

Shaping wget

Our wget command is below. You can see we aren’t trying very hard to access a page that doesn’t respond quickly (the various timeout options). Also, we’re only looking 5 pages deep per URL. We are not visiting any “parent” pages, that is, we’re not crawling up any directories. We don’t want any images or binary files (the reject) options and we don’t care about invalid SSL certificates (no-check-certificate).

wget \
  --tries=5 \
  --dns-timeout=30 \
  --connect-timeout=5 \
  --read-timeout=5 \
  --timestamping \
  --directory-prefix=data/pages \
  --wait=2 \
  --random-wait \
  --recursive \
  --level=5 \
  --no-parent \
  --no-verbose \
  --reject *.jpg --reject *.gif \
  --reject *.png --reject *.css \
  --reject *.pdf --reject *.bz2 \
  --reject *.gz  --reject *.zip \
  --reject *.mov --reject *.fla \
  --reject *.xml \
  --no-check-certificate

DNS

A nice thing about our setup is that each wget process is assigned to one domain. wget caches the DNS lookup so we only need to make one DNS request per process. A problem with this setup is that wget uses gethostbyname (or getaddrinfo depending on your platform). A quick check on man gethostbyname shows that on my BSD-based Mac gethostbyname is thread-safe e.g. it is synchronized. The result is that there is going to be some resource starvation when we have hundreds of wget processes all trying to call BIND all at the same time. 1

We set a DNS timeout of 30 seconds here, but in practice I found that it didn’t matter much. All of the processes race to grab the DNS lookup lock at the beginning, a large number time-out (waiting for the lock), but the requests even out over a couple of minutes.

xargs

xargs acts as our thread-pool or, more specifically, our process-pool.

-P specifies the number of processes to use. -I <sub> is a substitution parameter. It means “for each line in STDIN run CMD substituting the current line for <sub>“. So below we substitute _URL_ with the actual URL contained in the URLS_FILE.

cat $URLS_FILE | xargs -P $CRAWLERS -I _URL_ $WGET_CMD _URL_

Code

Our crawler script looks like this:

#!/bin/bash
# a basic crawler in bash
# usage: crawl.sh urlfile.txt <num-procs>
URLS_FILE=$1
CRAWLERS=$2

mkdir -p data/pages

WGET_CMD="wget \
  --tries=5 \
  --dns-timeout=30 \
  --connect-timeout=5 \
  --read-timeout=5 \
  --timestamping \
  --directory-prefix=data/pages \
  --wait=2 \
  --random-wait \
  --recursive \
  --level=5 \
  --no-parent \
  --no-verbose \
  --reject *.jpg --reject *.gif \
  --reject *.png --reject *.css \
  --reject *.pdf --reject *.bz2 \
  --reject *.gz  --reject *.zip \
  --reject *.mov --reject *.fla \
  --reject *.xml \
  --no-check-certificate"

cat $URLS_FILE | xargs  -P $CRAWLERS -I _URL_ $WGET_CMD _URL_

I’ve put this code on github with a Rakefile so you can follow along.

git clone git://github.com/jashmenn/bashpider.git
cd bashpider
rake data:get_urls # downloads and parses DMOZ, will take a while
rake crawl:restart # this will run a crawl

If you want to monitor the downloads per second, in another window type the following:

rake crawl:watch

When you feel you’ve gathered enough data, CTRL-C to kill both windows and then type:

rake results:process

Results at Home

As you can see from the chart, on my home computer through u-verse we max out at about 150 agents at 27 pages/sec, far below our original estimate of 200 pages/sec.

procs pages/sec
10     3.9
25     8.9
50    15.8
75    19.7
100   25.8
125   26.1
150   27.3
175   22.3
200    6.0

First of all, our initial estimate of 10KB per page was too low. In reality we observed a mean page size of around 37KB. This means on our 2300KB connection we can only expect a best-case download rate of 62 pages/sec. Still, 27 pages/sec is not even half that speed.

The other problem could be DNS requests. My home router also serves as my DNS server. It’s good enough for home use, but I’m pretty sure it’s not up to this task. Come to think of it, I’m not even sure how fast this CAT5 cable is.

I think it’s time to try out this setup in a better environment.

In the Data Center

For this experiment we loaded our script onto a beefy 8-core machine with a fat bandwidth connection.

The results were much better:

procs  pages/sec
  150   54
  200   71
  300  107
  400  141
  500  178
  600  214
  700  244
  800  386
  900  327
 1000  203
 1100  222
 1200  392
 1300  202
 1500  249
 1600  485
 2000  577
 3000  679
 3500  459
 4000  336

Take these numbers as rough estimates. For each of these entries I only let them run for a few minutes.

Processes

When I started getting into the thousands of processes, I wondered if I would hit the user process limit. Jay Donnell pointed out to me that uname will also give the process limit:

 ulimit -a
 max user processes              (-u) 268287

So with 260k+ processes available, we have no problem there.

Files

Using wget process gets its own file, which is uncompressed. So we’ve got a lot of disk IO going on. We’d probably save a good amount of time if each process just opened one file and appended content to it. We’d also save the file system from creating hundreds of thousands of inodes.

Also, the decline we see around 3000 agents may be due, in part, to the max number of open files on our system:

$ ulimit -a
open files                      (-n) 1024

Each crawler waits an average of 2 seconds before making the next request, at which time it makes the request and then downloads the file. So each process is making 1 file every >2 seconds. This means the number of wget processes we can run should be at least twice the max number of open files (2048).

My theory is this: at around 3000 concurrent agents the time it takes to actually download the content means that the probability that we will have enough file descriptors available. However, once we have 4000 concurrent agents the probability that any two agents will need to write a file at the same time is much higher, we see a significant performance drop.

I think we’re going to need to look at the file system format. Currently we’re using ext3, but I’m not sure if we should switch to xfs. While monitoring the file count using find I kept getting the following error:

find: WARNING: Hard link count is wrong for <some file>: this may be a bug
    in your filesystem driver.  Automatically turning on find's -noleaf option.
    Earlier results may have failed to include directories that should have been searched.

Also kjournald seemed to be working very hard to keep up with all the file writes. I’m not sure if this is unavoidable or not. I’m going to leave this problem for future work.

Summary

This crawler is just a baseline to see what performance is possible of basic unix utilities. Obviously, this approach used a list of static URLs and in a “real” crawler you probably want to have a mechanism for communicating and prioritizing URLs throughout the system.

That said, if you already know the list of URLs you want to download, you could download tens-of-million pages over a 24-hour window. For instance, if we assume a sustained rate of 600 pages per second you could download 51.8 million pages in 24 hours.

So how long would it take to download a billion pages with xargs and wget?

If you had the list of URLs beforehand, according to these numbers it would
take 19 days .

To download a billion pages is a week we’re going to need to figure out a way to download at least 1000 more pages per second.

What we’ve learned:

• watch out for file limits
• append to a single file rather than creating thousands of tiny files
• run your own non-locking DNS server
• unix tools are handy and powerful

What do you think?

Any suggestions for cranking out more performance out of wget? Should I try increasing my open file limit and see what happens? Think these numbers are ridiculous? Leave a comment below!

Share:

I was trying some remote git repo tests last night, and when I went to push my changes, I received this warning message:

warning: updating the currently checked out branch; this may cause confusion, as the index and working tree do not reflect changes that are now in HEAD.

On remote server I created a directory and did “git init“. Then cloned it from my local machine, did changes, committed, and then push. All seemed straightforward there. Any thoughts?

The issue git is warning you about is that your remote has both the repository and a working copy. That is, on the remote server you have a directory project/ with files in it (the working copy) and the folder project/.git (the repository).

If you push from your local machine to the remote, you will only be updating files in the repository and not the working copy. That is, the non-git files will not be changed. This can be confusing because you might log into the remote after you push an expect the working copy to be different.

To deal with this possible confusion git init provides a --bare option. What this does is create the repository only (no working copy). You can then push and pull from the remote like you might a central svn server.

Let me show an example. Say I have an existing git repository on my local machine and I want to create a new remote to back it up. My workflow would look like this:

ssh me@myserver.com
mkdir ~/git/newproject.git
cd ~/git/newproject.git
git init --bare
exit
git remote add myserver me@myserver.com:/home/nmurray/git/newproject.git
git push myserver master

If you want, you could even chain these together as a single command:

ssh me@myserver.com "mkdir ~/git/newproject.git && cd ~/git/newproject.git && git init --bare" && echo git remote add myserver me@myservercom:/home/nmurray/git/newproject.git

Hope this helps!

Share:

I like to use Cascading (or cascalog) for my Hadoop jobs, but out of the box Cascading doesn’t support using SequenceFiles that were created outside of Cascading. That is to say, Cascading requires that your SequenceFiles values be an instance of Tuple.

The solution is to create your own Scheme that parses a SequenceFile according to your own format. In my case I just want to parse each line as the string list.

The code is simple but may not be obvious for a first-time Cascading user. I hope this will save someone a few minutes.

    package com.xcombinator;
 
    import java.io.IOException;
 
    import cascading.tap.Tap;
    import cascading.tuple.Fields;
    import cascading.tuple.Tuple;
    import cascading.tuple.TupleEntry;
    import cascading.tuple.Tuples;
    import cascading.scheme.SequenceFile;
    import org.apache.hadoop.mapred.JobConf;
    import org.apache.hadoop.mapred.OutputCollector;
    import org.apache.hadoop.mapred.SequenceFileInputFormat;
    import org.apache.hadoop.mapred.SequenceFileOutputFormat;
 
    /**
     * A SequenceFileAsText is a type of {@link SequenceFile}, however the
     * SequenceFile has been created outside of Cascading and is assumed to have a
     * value of a string.
     */
    public class SequenceFileAsText extends SequenceFile
      {
      /** Field serialVersionUID */
      private static final long serialVersionUID = 1L;
 
      /** Protected for use by TempDfs and other subclasses. Not for general consumption. */
      protected SequenceFileAsText()
        {
        super( null );
        }
 
      /**
       * Creates a new SequenceFileAsText instance that stores the given field names.
       *
       * @param fields
       */
      public SequenceFileAsText( Fields fields )
        {
        super( fields );
        }
 
      @Override
      public Tuple source( Object key, Object value )
      {
        if (value instanceof Tuple)
        {
          return (Tuple) value;
        }
        else if (value instanceof Comparable)
        {
          return new Tuple((Comparable) value);
        }
        else if (value != null)
        {
          return new Tuple(String.valueOf(value));
        }
        else
        {
          return new Tuple((Comparable)null);
        }
      }
 
    }

Share:

Git Cheat Sheet Preview

I realize there are a number of cheatsheets for git already. However, I wanted a simple, one-page sheet specifically for my attendees.

You can download it here:

• git cheatsheet pdf
• git cheatsheet LaTeX source

Like it? Hate it? Find a typo? Leave your feedback in the comments!

Here are my raw notes from the talk:

;; -*- mode: Markdown; -*-

# How to read:
commands are indented
actions to perform while presenting are marked with @
left to right

# Welcome
see progit.org
what is version control

why use it:

  * backup/restore
  * synchronization sharing
  * track changes
  * ownership
  * branching and merging

who has used subversion 

git
  * you've heard its distributed
  * b/c branching and merging

pace - slow, no slides

leave with practical understanding

# Install & Config

    sudo port install git-core +svn
    git config --global user.name "Nate Murray"
    git config --global user.email "nate@natemurray.com"

# Basic Commands

    cd ~
    mkdir -p projects/demo       # explain only a little
    cd projects/demo
    git init
    git status                   # nothing here
    ls -a                        # talk .git repository vs. working copy
    echo "version 1" > README.txt
    git status                   # untracked file
    git add README.txt
    git status                   # changes to be committed
    git commit -m "added version one of the file"
    git status                   # clean

stop, draw the picture of the local operation phases - e.g. svn vs. git

> Principle 1: (almost) everything is local

so now that you know about the staging area, lets do it again

    echo "new file" > sheep.rb
    git status                   # draw untracked
    git add sheep.rb
    git status                   # draw staged
    git commit -m "added"

    cat README.txt                 # draw unmodified
    echo "version 2" > README.txt
    git status                   # draw modified
    git commit -a -m "updated version" # shorthand for git add
    git status

Tips:

    git config --global alias.st status
    git st

# Git Internals

* Before we can talk about branching you *have* to understand how git (tried to avoid this)
* files and folders

three objects -  @ Draw first commit

  * blob        - raw data
  * tree        - folder (stores blobs and trees)
  * commit      - snapshot of the repo + meta 

You won't need to use `git cat-file` on a daily basis. however, understanding
the concepts we're going to talk about is really important for branching.

    git log # view the log
    git show ----  # first commit, whatever it is

    git cat-file -p  ---- # first commit
    git cat-file -p  ---- # tree
    git cat-file -p  ---- # blob

draw the rest using git `cat-file`

    git log           # show the log again
    git cat-file -p ---- # second commit

draw the picture. point out the parent connection.
note committer / author

    git cat-file -p ---- # tree

note here there are two blobs!

finish drawing out the second commit
* git stores reference to first file.
* snapshot of the *whole project*
* git stores each file once
* filename is in the `tree` 

draw the last commit

     git log
     git cat-file -p ---- # third commit

> Principle #2 : Git commits are snapshots

* A commit in git is a snapshot of the entire project, not just a list of diffs.
* snapshot is based on the SHA hash function. guarantees file integrity

# refs/branches

questions?

@ stop. redraw commits as *linear* . looking only at commits

ready to define a branch
a branch is a pointer to a commi
text file with a sha. thats it. 

start with one branch called `master`

    git branch

bash prompt

    # skip this
    tree .git/refs/
    cat .git/refs/heads/master
    git log
    # compare the SHAs

update diagram by adding a `ref` to our commit. (`master`). 

@ draw circle pointing to commit

create testing branch

# branching

So lets create another branch:

    git branch testing
    git branch

only created, didn't switch. just created a ref pointing to this
commit

@ update diagram

How does git know what branch we are "on"?

special ref called `HEAD` that points to the local branch
since we are still on master HEAD points to master

@ add HEAD

To switch working copy, use the `git checkout`

    git checkout testing
    git branch

HEAD moves from `master` to `testing`

@ update diagram

master and testing point to the same commit, working directory isn't changed

checkout means something different in git than it does in svn.
checkout in git to switch our working directory to a particular commit. 

now make changes:

    cat README.txt
    echo "we are on the testing branch!" > README.txt
    cat README.txt
    git commit -a -m "updated the readme"
    git log

@ update diagram, adding new commit. move the testing ref and the HEAD ref with it

add a "test"

    echo "this is a test" > test.rb
    git add test.rb                    # stage it for our commit
    git commit -m "added a test"       # now commit
    git log

@ update diagram - should have two commits

hotfix - scenario: you need to switch back to master

    git checkout master
    ls

@ move HEAD

so notice two things.
1) switching to this branch was fast - everything is local
2) our file test.rb is absent

and if we

    cat README.txt

it says 'version 2' just like we would expect

    echo "applying fix" >> sheep.rb
    cat sheep.rb
    git commit -a -m "applied important fix"
    git log
    git cat-file -p ---- # last commit

@ draw the new commit, and draw its reference back to the parent. move HEAD and master

now fixed, can push into production
and get back to work in `testing`

    git checkout testing
    cat README.txt
    cat test.rb

This is a general pattern:

> Principle #3: Branching is cheap, use it often

If you are working on a particular feature, create a branch. 

If you're coming from svn, making frequent branches might seem unnatural.
in svn, a branch is global -> namespace issues.
vs. git: private branches
name your branch 'test' and it won't collide with anyone elses

But branching itself isn't that useful unless its easy to merge.

* how many of you have merged a branch in svn?
* how many of you enjoyed it?

merging is one of git's strength and git makes it relatively easy

# merging

    cat sheep.rb

two branches: `master` and `testing` - need to merge

    git checkout master
    git merge testing
    git show HEAD

instead of a 'parent' we have a line that says 'merge'
a merge commit has more than one parent

@ draw the commit object
@ draw lines to the commits

    gitx

sometimes merging doesn't go as planned - conflicts

    git checkout -b breaker

this is shorthand for create and then checkout a new branch based on the
current HEAD

    vi sheep.rb # changing fix
    git commit -a -m "changed the fix"
    git checkout master
    vi sheep.rb # improving fix
    git commit -a -m "improved the fix"

@(update diagram, adding breaker and master refs)

    git merge breaker
    git status

there are many diff viewing tools.
* perforce
* opendiff - from apple

    git mergetool -t opendiff

I don't really like using the visual tools.
Sometimes you need character level editing

    vi sheep.rb
    git add sheep.rb
    git commit -a

talk about merge with conflicts

@ update diagram draw new merge commit

    gitx

Questions?

# Remotes

Everything so far on one machine. 

I work offline (I take the train)
If I break something I can rollback see where I was an hour ago 

want to share our changes.
might seem scary or messy because changes to totally independent lines of the code.
but in practice its not a problem.

svn version numbers are incremental - so two repos would get out of
step
no easy way of merging two separate repostories. 

git blob identifiers are a SHA of the content.
if the same content is created anywhere in the universe you'll still
have the same SHA

git doesn't care about where your commits come from or how you get them

Protocols:
  * ssh
  * git
  * http
  * local file system

sample project on our github

    cd ..
    open http://XXX/nmurray/simple-echo
    git clone git@XXX:nmurray/simple-echo.git
    cd simple-echo
    git log

svn checkout just HEAD
vs. git - whole repo

To be able to collaborate with others you have to manage 'remote repositories'.
When you clone a project, you have a default remote called 'origin'. 

    git remote -v

Remotes are pointers to other repositories that are _usually_ over the network.
'pull' and 'push' changes.

    vi README.mkd
    # make a change
    git commit -a -m "make a change"
    git push

If someone else makes a change:

    git pull origin master

This means pull from `origin` the branch `master` into local branch `master`. You can often to just

    git pull

which means pull from origin whatever branch Im on (i.e. HEAD) into this branch.

Now let's say someone pushes a change and I make a change
I can't push unless I pull first. This is good.

# remote forks

So that is while we are on the same line. What if were on different lines?

@(open up webbrowser again)

Bh also has forked my project. But when we say forked, all the means is he has
created his own development line from some of my commits

    git remote add bh git@XXX:bhenderson/simple-echo.git
    git remote -v

Now you shouldn't be surprised to learn that adding the remote doesn't change
anything. First we have to `fetch` hist changes

    git fetch bh

`fetch` brings his commits into my repo but again, doesnt change my working copy.

fetch brought branches + commits into repo
work with those branches just like any other branch.

    git branch -a

So you see here we have 

* `master`, which is our local master
* we have at the bottom `origin/master` which is the origin where we pulled from branch master
* and then we have `bh/master`, which is bhendersons master branch

These are all regular branches: they are just pointers to commits. We
can even checkout as branch 

    git checkout bh/master

scary message

    git checkout master

So how would we merge bhendersons changes with our own? I'm sure you could guess by now. Simply:

    git merge bh/master # don't press enter!!!

But lets take it up a notch.
say you didn't want to merge bh changes in your master branch.
real world, you might not know if his changes would merge cleanly
don't want to mess up your master branch.  

What we are going to do is
* create a new branch,
* merge bhs branch in THAT branch
* then we're going to merge to master.

It will make more sense when we do it. Lets try:

Okay we first want to create a new branch based on our master

    git checkout -b bh-merge
    git branch -a 

Now lets merge his changes

    cat simple-echo.rb
    git merge bh/master
    cat simple-echo.rb
    git log                  # see bh as the author of the commit

okay everything was clean! *phew* now lets go back to master

    git checkout master
    git merge bh-merge
    git log

and there we go! merged nicely.
now I don't need bhendersons merge branch anymore, so lets delete it

    git branch -d bh-merge
    git branch -a

git is distributed

Instead of one central server, that everyone has to sync to,
* independent lines of work can go on.
* If someone creates something good in their branch, they just tell people about it.
* permission-less 

you can see why it is so good for open-source development

questions about branching?

# Advanced

* tagging
* rebase
* cherry pick
* git bisect
* hooks
* tracking branches
* submodules
* interactive staging
* squashing commits
* git-svn
* setting up your own server
* patches via email
* gitjour

Share:

Scalability

The web is enormous and continually growing. A crawler should scale
linearly with the number of agent-machines that are added to the
system. This allows us to add more agents as our needs increase.

Speed

Speed is a significant issue at this scale. For example, if we want to crawl 1
billion pages in a week (this is less than 1/1000th of the web), our system
will have to sustain a rate of 1653 downloads per second.

To achieve this speed we need to employ a number of techniques such as
concurrent connections, data compression, dns caching, minimize disk seeks,
etc.

Politeness

While we need a high rate of download, we must be polite and not
overload one particular server. Najork et. al. propose limiting requests to
a single server by waiting 10 times the time it took to download the
last page. 1

Quality

We aim to build a crawler that visits “high-quality” or “relevant” pages
well-known quality metric. However, Najork et. al. found that a simple
breadth-first crawl tends to visit high-quality pages first 2.
We’ll eventually build a more intelligent page-selection mechanism, but for now
breadth-first will work.

Agents as a Distributed System

Given that the crawling problem cannot be solved by a single machine,
we are required to form our solution as a distributed
system. Distributed systems introduce more room for failures
and errors in coordination. Therefore we define the following desirable
features for our distributed system:

Fault Tolerance

Hardware failure is unavoidable. The failure of one node should not
prevent survivors from continuing to operate.

Even Partitioning

The URL frontier should be evenly distributed across all agents to
evenly assign the work. Many crawlers use a hashing function to distribute
the URLs among machines.

Minimize Overlap

Overlap is defined as (n-u)/u, where n is the total number of
crawled pages and u is the number of unique pages (sometimes u < n
because the same page has been erroneously fetched several times).
Optimally, we want an overlap of 0. 3

Agent churn

During the crawl we may want to add additional resources. The system
should support agents coming and leaving the group.

Next Steps

There are five major parts to a web crawler:

• The URL frontier
• IP address lookup
• Page download
• Page processing
• Tracking URLs encountered

Over the next few articles we will be designing a each of the five
components. The list of desirable features give us guidelines
that help shape the decisions about each component.

1. Najork M. High-Performance Web Crawling. Systems Research. 2001.
2. Najork M, Wiener JL. Breadth-First Search Crawling Yields High-Quality Pages. Systems Research. 2001:114-118.
3. Boldi P, Codenotti B, Santini M, Vigna S. UbiCrawler: a scalable fully distributed Web crawler. Software: Practice and Experience. 2004;34(8):711-726.

Share:

So far I’ve tried aleph, compjure/ring/jetty and saturnine.

The compojure stack is by far the cleanest, but it’s geared more
towards synchronous request/response cycles. I’d like to use
something like to EventMachine and saturnine seems the closest to
that goal. However, saturnine’s current HTTP implementation is
lacking.

Now before I could contribute to saturnine I first needed to understand netty. If you’re trying to learn netty, I recommend you first read the users guide and then jump straight to the API docs on ChannelPipeline.

Next, I needed to write a basic HTTP server using netty. This post on StackOverflow and this sample code on netty’s website helped me get a basic HTTP server up an running.

Below is a a nieve translation of the netty sample code into clojure. Note that this is not listed here as an example of sexy clojure code, but rather a starting point for someone looking to get dirty with the netty libraries in clojure.

    (ns xcombinator.netty.server
      (:gen-class)
      (:use clojure.contrib.import-static)
      (:import
         [java.net InetSocketAddress]
         [java.util.concurrent Executors]
         [org.jboss.netty.bootstrap ServerBootstrap]
         [org.jboss.netty.channel Channels ChannelPipelineFactory
           SimpleChannelHandler SimpleChannelUpstreamHandler]
         [org.jboss.netty.channel.socket.nio NioServerSocketChannelFactory]
         [org.jboss.netty.buffer ChannelBuffers]
         [org.jboss.netty.handler.codec.http HttpRequestDecoder
           HttpResponseEncoder DefaultHttpResponse]
    ))

    (import-static org.jboss.netty.handler.codec.http.HttpVersion HTTP_1_1)
    (import-static org.jboss.netty.handler.codec.http.HttpResponseStatus OK)
    (import-static org.jboss.netty.handler.codec.http.HttpHeaders$Names CONTENT_TYPE)

    (declare make-handler)

    (defrecord Server [#^ServerBootstrap bootstrap channel])

    (defn start
      "Start a Netty server. Returns the pipeline."
      [port handler]
      (let [channel-factory (NioServerSocketChannelFactory.
                              (Executors/newCachedThreadPool)
                              (Executors/newCachedThreadPool))
            bootstrap (ServerBootstrap. channel-factory)
            pipeline (.getPipeline bootstrap)]
        (.addLast pipeline "decoder" (new HttpRequestDecoder))
        (.addLast pipeline "encoder" (new HttpResponseEncoder))
        (.addLast pipeline "handler" (make-handler))
        (.setOption bootstrap "child.tcpNoDelay", true)
        (.setOption bootstrap "child.keepAlive", true)
        (new Server bootstrap (.bind bootstrap (InetSocketAddress. port))))) 

    (defn stop-server
      {:doc "Stops a Server instance"
       :arglists '([server])}
      [{bootstrap :bootstrap channel :channel}]
      (do (.unbind channel)
          (.releaseExternalResources bootstrap)))

    (defn http-response
      [status]
      (doto (DefaultHttpResponse. HTTP_1_1 status)
        (.setHeader CONTENT_TYPE "text/plain; charset=UTF-8")
        (.setContent (ChannelBuffers/copiedBuffer
                       (str "Success: " status) "UTF-8"))))

    (defn make-handler
      "Returns a Netty handler."
      []
      (proxy [SimpleChannelUpstreamHandler] []
        (messageReceived [ctx e]
          (let [c (.getChannel e)
                cb (.getMessage e)
                ]
            (println "HTTP request from" c)
            (.write c (http-response OK))
            (-> e .getChannel .close)))

        (exceptionCaught
          [ctx e]
          (let [throwable (.getCause e)]
            (println "@exceptionCaught" throwable))
          (-> e .getChannel .close))))

    (comment
      (def *server* (start 3335 make-handler))
      (stop-server *server*)
    )

Here’s a project.clj that will load up the right dependencies:

    (defproject server "0.0.1"
      :description "Simple Netty HTTP server"
      :repositories [["JBoss" "http://repository.jboss.org/maven2"]]
      :dependencies
        [[org.clojure/clojure "1.2.0-beta1"]
        [org.clojure/clojure-contrib "1.2.0-beta1"]
        [org.jboss.netty/netty "3.2.0.BETA1"]
        [log4j/log4j           "1.2.14"]]
      :dev-dependencies [[autodoc              "0.7.0"]
                         [lein-clojars         "0.5.0-SNAPSHOT"]
                         [lein-run "1.0.0-SNAPSHOT"]
                         [swank-clojure "1.2.1"]]

      :namespaces [xcombinator.netty.server])

Then lein swank and evaluate the (def *server*... line in your
REPL. You’ll have a simple netty HTTP server running on port 3335.

Share:

I was looking for an easy way to replicate ruby’s fileutils library in clojure and I came across a version by Mark McGranaghan in his clj-garden project which wraps around the Apache Commons IO library.

I extended Mark’s library by using multimethods to allow the use of strings (rather than requiring File objects) for parameters.

Usage

user=> (use 'clj-file-utils.core)
nil
user=> (exist "foo.txt")
false
user=> (touch "foo.txt")
nil
user=> (exist "foo.txt")
true
user=> (rm "foo.txt")
nil
user=> (file "foo.txt")
#<File foo.txt>
user=> (.getParent (file "/path/to/foo.txt"))
"/path/to"

As A Dependency

leiningen

[clj-file-utils "0.1.1"]

maven

<dependency>
  <groupId>clj-file-utils</groupId>
  <artifactId>clj-file-utils</artifactId>
  <version>0.1.1</version>
</dependency>

Code

• jashmenn/clj-file-utils on github

Share:

The tf-idf weight (term frequency-inverse document frequency) is a weight often used in information retrieval and text mining. This weight is a statistical measure used to evaluate how important a word is to a document in a collection or corpus. [1]

To calculate tf-idf we need the following four values:

• The number of times a term appears in a given document (n_i,j)
• The total number of terms in a given document (sum k, n_k,j)
• The number of documents that contain a given term (|{d : t_i E d}|)
• The total number of documents in the corpus (D)

Mathematical Details

We want to score the importance of term t_i in document d_j.

Term frequency is defined by:

Where n_i,j is the number of occurrences of term t_i in document d_j.

Inverse document frequency is defined by:

Where:

• D is the total number of documents in the corpus and
• |{d : t_i E d}| is the number of documents in which the term t_i appears.

Then:

Refer to [2] for more information on tf-idf.

Operation Input

Last time we discussed the technique of taking a group of records, calculating a value from that group and emitting each record with the calculated value attached. We called this operation a BufferedSum. We’re going to build on our previous work and create a reusable component (called a SubAssembly) for calculating tf-idf using Cascading on Hadoop.

To make our tf-idf operation we need to decide what the input arguments will be. Last time, we used an input corpus of the format (document_id, body) and emitted (document_id, term, term_count_in_document) for all terms in each document. This last tuple will be the input format to our tf-idf operation.

Creating a SubAssembly

The general format of a SubAssembly in Cascading is as follows:

public class MySubAssembly extends SubAssembly {
  public MySubAssembly(Pipe pipe) {
    // do something with `pipe`
    setTails(pipe); // must register all assembly tails
  }
}

In our operation we are assuming that the total number of documents in the corpus is known or could be found with a simple MapReduce job. Given that we have the total number of documents we take that number as the input to our SubAssembly:

public class TfIdfIndexSubAssmbly extends SubAssembly {
  private Integer totalNumberOfDocuments;
  public TfIdfIndexSubAssmbly(Pipe pipe, Integer totalNumberOfDocuments) {
    this.totalNumberOfDocuments = totalNumberOfDocuments;
    // do something with pipe
    setTails(pipe); // must register all assembly tails
  }
}

Gathering Variables

To compute our final tf-idf score, we first need to compute the intermediate variables.

total_terms_in_document

Given our input is (document_id, term, term_count_in_document) then we already have the first variable n_i,j. We can now calculate the total number of terms in each document:

// input: (document_id, term, term_count_in_document)
// emits: (document_id, term, term_count_in_document, total_terms_in_document)
pipe = new GroupBy(pipe, new Fields("document_id"));
pipe = new Every(pipe,
    new BufferedSum(new Fields("total_terms_in_document"),
                   new Fields("term_count_in_document"),
                   new Fields("document_id", "term", "term_count_in_document")),
    Fields.SWAP);

Remember that BufferedSum takes three arguments:

• The name of the Field to output
• The name of the Field to sum
• The other Fields to “pull through” the operation

So here we are grouping by document_id, and summing term_count_in_document for each group and placing the value into the field total_terms_in_document.

number_of_documents_containing_term

Next we need to calculate the number of documents that contain each term. We’ve already grouped by document_id and term, therefore we know we only have one record for a given document_id/term pair.

Rather than counting the number of document_id/term pairs directly we are simply going to assign a count of 1 to each record and then sum that value. This allows us to reuse the code we’ve written for BufferedSum.

// calculate the number of documents containing each term
// input: (document_id, term, term_count_in_document, total_terms_in_document)
// emit:  (document_id, term, term_count_in_document, total_terms_in_document, number_of_documents_containing_term)
pipe = new Each(pipe, new Insert(new Fields("term_in_doc"), 1), Fields.ALL); // we're going to sum these, easier than creating BufferedCount
pipe = new GroupBy(pipe, new Fields("term"));
pipe = new Every(pipe,
    new BufferedSum(new Fields("number_of_documents_containing_term"),
                   new Fields("term_in_doc"),
                   new Fields("document_id", "term", "term_count_in_document", "total_terms_in_document", "term_in_doc")),
    Fields.SWAP);

Here we group on term and for every term group we calculate the number of documents that contain that term. Note that if you have a very large corpus some groups may become memory constrained as very common words such as “the” have groups containing nearly the entire corpus (it would be a good idea to remove stop-words during pre-processing).

After we’ve calculated the value for number_of_documents_containing_term we don’t need the term_in_doc field any longer. Using Cascading’s Identity operation we can reorder and discard fields.

 pipe = new Each(pipe, //  reorder and rm fields
    new Fields("document_id", "term", "term_count_in_document", "total_terms_in_document", "number_of_documents_containing_term"),
    new Identity());

total_documents

Next we do a hard insert of the number of documents. Again, you can calculate this value with a relatively simple MapReduce job (e.g. use a counter), but here probably not the best place to do it. Because you have to count every document in the corpus, it would be better to calculate the number of documents when you are generating the input documents file.

int D = this.totalNumberOfDocuments;
pipe = new Each(pipe, new Insert(new Fields("total_documents"), D), Fields.ALL);

The Insert operation simply inserts the number of documents into the tuple stream.

Calculating Tf-idf with a Custom Operation

Now that we have all four values we can calculate tf-idf. We are going to create a custom operation to do this. We will perform this operation on each Tuple in the Pipe with the following:

// calculate tf * idf
// input:  (document_id, term, term_count_in_document, total_terms_in_document, number_of_documents_containing_term, total_documents)
// emit:   (document_id, term, term_count_in_document, total_terms_in_document, number_of_documents_containing_term, total_documents, tf, idf, tfidf)
pipe = new Each(pipe, new TfIdfOperation(), Fields.ALL);

To create an operation in Cascading you simply subclass BaseOperation. For example:

private static class MyOperation extends BaseOperation implements Function {
  public MyOperation() {
    super(new Fields("out_field_1", "out_field_2"));
  }
  public void operate(FlowProcess flowProcess, FunctionCall functionCall)
  {
    TupleEntry inputTuple = functionCall.getArguments();
    // take values form inputTuple
    // transform them to make outputTuple
    Tuple outputTuple = new Tuple("a value 1", "a value 2");
    functionCall.getOutputCollector().add(outputTuple);
  }
}

Here are the key things you need to do to create a Cascading operation:

• subclass BaseOperation and implement Function (there are other types of operations)
• call super and declare the names of the Fields this operation will be emitting. See BaseOperation for details
• functionCall.getArguments() returns a TupleEntry containing the input Tuple and input Fields.
• functionCall.getOutputCollector() is the OutputCollector you can use to emit Tuples from this operation.
• call outputCollector.add() to emit a Tuple. You can emit 0..n Tuples.

For our tf-idf operation we want to emit three Fields: tf, idf, and tfidf. We do this with the following code.

// note that we're using a private nested class. This is not required.
private static class TfIdfOperation extends BaseOperation implements Function {
  public TfIdfOperation() { super(new Fields("tf", "idf", "tfidf")); }
  public void operate(FlowProcess flowProcess, FunctionCall functionCall)
  {
    TupleEntry arguments = functionCall.getArguments();
    // tf
    Double termCount  = arguments.getDouble("term_count_in_document");
    Double totalTerms = arguments.getDouble("total_terms_in_document");
    BigDecimal tf = new BigDecimal(termCount / totalTerms);
    // idf
    Double totalDocuments  = arguments.getDouble("total_documents");
    Double td = arguments.getDouble("number_of_documents_containing_term");
    BigDecimal idf = new BigDecimal(Math.log(totalDocuments / (1 + td)));
    // tfidf
    BigDecimal tfidf = tf.multiply(idf);
    functionCall.getOutputCollector().add(
        new Tuple( tf.toPlainString(), idf.toPlainString(), tfidf.toPlainString() ));
  }
}

When we convert, say, a Double to a string then we often get an exponent (e.g. 0.02948E7). The exponent can be cumbersome to work with so we use BigDecimal to convert the number into a string without an exponent using toPlainString().

Building the Index

Now that we have the term, document_id, and tfidf score we can build our index. First we strip out the unnecessary fields and move the term to the front.

// input:  (document_id, term, term_count_in_document, total_terms_in_document, number_of_documents_containing_term, total_documents, tf, idf, tfidf)
// emit:   (term, document_id, tfidf)
pipe = new Each(pipe, // reorder and rm some fields
    new Fields("term", "document_id", "tfidf"),
    new Identity());

In our last step we want to build a single row that tells us “given a term, what documents are most relevant to it”. We format this list of (document_id, score) pairs as a JSON hash (JSONObject). (Formatting our records in this way is called using “stripes”.)

pipe = new GroupBy(pipe, new Fields("term"));
// "stripe" our group e.g.:
// input: (term,
//          (document_id_1, tfidf_1),
//          (document_id_2, tfidf_2),
//          ...)
// emit: (term, {document_id_1:tfidf_1, document_id_2:tfidf_2})
pipe = new Every(pipe,
    new Fields("document_id", "tfidf"),
    new JSONTupleAggregator(new Fields("scores"), "JSONObject"),
    new Fields("term", "scores"));

JSONTupleAggregator is an operation that can be found in the cascading.json project. It takes a group of tuples and emits them as either a JSONArray (nested list) or JSONObject (hash).

Using the Index

To use the index, simply perform a CoGroup on the term in your right-hand-side to the term in our index. See Cascading’s documentation on CoGroup for more information.

Full Code Listing

package com.xcombinator.cascading.pipes;
import com.xcombinator.cascading.operations.*;
import com.xcombinator.cascading.operations.buffers.*;
import cascading.tuple.Fields;
import cascading.pipe.*;
import cascading.operation.regex.RegexSplitter;
import cascading.operation.Identity;
import cascading.operation.text.DateParser;
import cascading.operation.Insert;
import cascading.flow.FlowProcess;
import cascading.tuple.TupleEntry;
import cascading.operation.BaseOperation;
import cascading.operation.Function;
import cascading.operation.FunctionCall;
import cascading.pipe.Each;
import cascading.pipe.Every;
import cascading.pipe.GroupBy;
import cascading.tuple.Tuple;
import cascading.operation.Debug;
import cascading.json.operation.aggregator.*;
import java.math.BigDecimal;
/**
 *
 * The goal of this SubAssembly is to create an index that can be used to find
 * the most relevant document given a term.
 * Required: you need to input the number of total documents.
 * Input: list of (document_id, term, count). e.g. :
 *
 *   (document_id,  term,  2)
 *   (document_id,  term2, 1)
 *   (document_id2, term2, 3)
 *   # etc
 *
 * where `count` is the number of times that term appears in that document
 *
 *
 * Emits: list of (term: {document_id_1:score_1,document_id_2,score_2...})
 *
 * Note that the tuple emitted is all String representation of the decimal
 * numbers. This is to allow easy (and correct) parsing if you write to a file
 * after this.
 *
 * It is assumed you've already done any normalization of the terms such as stemming etc.
 */
public class TfIdfIndexSubAssmbly extends SubAssembly {
  private static class TfIdfOperation extends BaseOperation implements Function {
    public TfIdfOperation() { super(new Fields("tf", "idf", "tfidf")); }
    public void operate(FlowProcess flowProcess, FunctionCall functionCall)
    {
      TupleEntry arguments = functionCall.getArguments();
      // tf
      Double termCount  = arguments.getDouble("term_count_in_document");
      Double totalTerms = arguments.getDouble("total_terms_in_document");
      BigDecimal tf = new BigDecimal(termCount / totalTerms);
      // idf
      Double totalDocuments  = arguments.getDouble("total_documents");
      Double td = arguments.getDouble("number_of_documents_containing_term");
      BigDecimal idf = new BigDecimal(Math.log(totalDocuments / (1 + td)));
      // tfidf
      BigDecimal tfidf = tf.multiply(idf);
      functionCall.getOutputCollector().add(
          new Tuple( tf.toPlainString(), idf.toPlainString(), tfidf.toPlainString() ));
    }
  }
  private Integer totalNumberOfDocuments;
  public TfIdfIndexSubAssmbly(Pipe pipe, Integer totalNumberOfDocuments) {
    this.totalNumberOfDocuments = totalNumberOfDocuments;
    // calculate the total terms in each document. note that the input set is
    // smaller because we've already counted the occurrence of each term in
    // each document
    // input: (document_id, term, term_count_in_document)
    // emits: (document_id, term, term_count_in_document, total_terms_in_document)
    // pipe = new Each(pipe, new Debug(true));
    pipe = new GroupBy(pipe, new Fields("document_id"));
    pipe = new Every(pipe,
        new BufferedSum(new Fields("total_terms_in_document"),
                       new Fields("term_count_in_document"),
                       new Fields("document_id", "term", "term_count_in_document")),
        Fields.SWAP);
    // calculate the number of documents containing each term
    // input: (document_id, term, term_count_in_document, total_terms_in_document)
    // emit:  (document_id, term, term_count_in_document, total_terms_in_document, number_of_documents_containing_term)
    pipe = new Each(pipe, new Insert(new Fields("term_in_doc"), 1), Fields.ALL); // we're going to sum these, easier than creating BufferedCount
    pipe = new GroupBy(pipe, new Fields("term"));
    pipe = new Every(pipe,
        new BufferedSum(new Fields("number_of_documents_containing_term"),
                       new Fields("term_in_doc"),
                       new Fields("document_id", "term", "term_count_in_document", "total_terms_in_document", "term_in_doc")),
        Fields.SWAP);
    pipe = new Each(pipe, //  reorder and rm some fields
        new Fields("document_id", "term", "term_count_in_document", "total_terms_in_document", "number_of_documents_containing_term"),
        new Identity());
    // here we do a hard-insert of the number of documents. again, you
    // could/should calculate this with MR, but this is not the place. It
    // would be better to calculate the number of documents when you are
    // generating the input documents file.
    int D = this.totalNumberOfDocuments;
    pipe = new Each(pipe, new Insert(new Fields("total_documents"), D), Fields.ALL);
    // now calculate tf * idf
    // input:  (document_id, term, term_count_in_document, total_terms_in_document, number_of_documents_containing_term, total_documents)
    // emit:   (document_id, term, term_count_in_document, total_terms_in_document, number_of_documents_containing_term, total_documents, tf, idf, tfidf)
    pipe = new Each(pipe, new TfIdfOperation(), Fields.ALL);
    pipe = new Each(pipe, // reorder and rm some fields
        new Fields("term", "document_id", "tfidf"),
        new Identity());
    pipe = new GroupBy(pipe, new Fields("term"));
    // "stripe" our group e.g.:
    // input: (term,
    //          (document_id_1, tfidf_1),
    //          (document_id_2, tfidf_2),
    //          ...)
    // emit: (term, {document_id_1:tfidf_1, document_id_2:tfidf_2})
    pipe = new Every(pipe,
        new Fields("document_id", "tfidf"),
        new JSONTupleAggregator(new Fields("scores"), "JSONObject"),
        new Fields("term", "scores"));
    // must register all assembly tails
    setTails(pipe);
  }
}

Share:

A common technique in MapReduce is to input a group of records, calculate a value from that group, and emit each record with the new value attached. While this is easy to do in raw MR jobs, the solution in Cascading is not very obvious. This tutorial introduces a new operation to Cascading called BufferedSum. BufferedSum allows us to calculate values from a group of tuples and emit the group value to individual tuples in a scalable way.

Describing the operation of BufferedSum is clearer when discussed in concrete terms, so let’s work with an example.

Example

When dealing with large amounts of documents in Hadoop, its common to have each input file to contain many documents. Our input file in this case will contain two documents:

a.txt\thello world world
b.txt\tgoodbye goodbye world

Lets say we want to calculate tf-idf for these documents. One of the first values we need is the count of the occurrence particular term within each document.

First, we will split each line into (document_id, body) pairs:

  pipe = new Each(pipe, 
      new Fields("line"), 
      new RegexSplitter(new Fields("document_id", "body"), "\t"));

From there we “tokenize” the document and extract each term:

  pipe = new Each(pipe, // tokenize words by space
      new Fields("body"),
      new RegexSplitGenerator(new Fields("term"), "\\s+"), 
      new Fields("document_id", "term"));

Now our tuple stream is the following:

a.txt hello
a.txt world
a.txt world
b.txt goodbye
b.txt goodbye
b.txt world

Count of term in document_id

We now have (document_id, term) and we want to calculate (document_id, term, term_count_in_document). With Cascading, this is easy, simply group by document_id and term and use the Count() function:

  // count how many times `term` appears in `document_id`
  pipe = new GroupBy(pipe, new Fields("document_id", "term"));
  pipe = new Every(pipe, 
      new Fields("term"), 
      new Count(new Fields("term_count_in_document")), 
      new Fields("document_id", "term", "term_count_in_document"));

Calculating total_terms_in_document

So far, so good. Up to this point Cascading has provided everything we need. However, next we want to get the total terms within each document and keep the tuples we have calculated thus far. Put another way, we have an input of (document_id, term, term_count_in_document) and we want to emit (document_id, term, term_count_in_document, total_terms_in_document)

Our first instinct might be to use GroupBy() and Count() like before. But there is a catch: Every operations emit the operator result with the group tuple (see the Each and Every Pipes in the Cascading User Guide).

This means if we group by document_id and Sum() the total_terms_in_document we will emit (document_id, total_terms_in_document). The number in total_terms_in_document will be accurate, but we lose our term and term_count_in_document.

If we try to save our other fields by grouping on all three of them (document_id, term, term_count_in_document) then we’ve “over-grouped” and every “group” is a single tuple (the input tuple) and we won’t get the count of terms in the document as a whole. BufferedSum was created to solve this problem.

BufferedSum

BufferedSum takes as its input three things:

• The name of the Field to output
• The name of the Field to sum
• The other Fields to “pull through” the operation

Here is how we can use BufferedSum to achieve the desired effect:

// input: (document_id, term, term_count_in_document)
// emits: (document_id, term, term_count_in_document, total_terms_in_document) 
pipe = new GroupBy(pipe, new Fields("document_id"));
pipe = new Every(pipe, 
    new BufferedSum(new Fields("total_terms_in_document"), 
                    new Fields("term_count_in_document"),
                    new Fields("document_id", "term", "term_count_in_document")), 
    Fields.SWAP);

Note: the output selector Fields.SWAP is critical due to Cascading tuple selection.

Memory considerations

One thing to be careful of when using BufferedSum is to try and keep your groups small enough to fit in memory. However, this is not a requirement. BufferedSum uses Cascading’s SpillableTupleList which will spill to the HDFS if it grows too large. That said, spilling is an expensive operation and should be avoided if possible.

Summary

BufferedSum is a widely useful operation when dealing with sums in Cascading. In Part 2 we will use BufferedSum and Cascading to finish calculating tf-idf.

The Code

package com.xcombinator.cascading.operations.buffers;
import cascading.flow.FlowProcess;
import cascading.operation.BaseOperation;
import cascading.operation.Buffer;
import cascading.operation.BufferCall;
import cascading.tuple.Fields;
import cascading.tuple.Tuple;
import cascading.tuple.TupleEntry;
import cascading.tuple.SpillableTupleList;
 
import java.util.Iterator;
 
/**
 * BufferedSum sums a value for every Tuple in a Group and emits every input
 * Tuple with the sum appended.
 * <p/>
 * 
 * 
 * EXAMPLE:
 *
 * {@code 
 *
 * // input: (document_id, term, term_count_in_document)
 * // emits: (document_id, term, term_count_in_document, total_terms_in_document) 
 *
 *     pipe = new GroupBy(pipe, new Fields("document_id"));
 *     pipe = new Every(pipe, 
 *         new BufferedSum(new Fields("total_terms_in_document"), 
 *                        new Fields("term_count_in_document"),
 *                        new Fields("document_id", "term", "term_count_in_document")), 
 *         Fields.SWAP);
 * }
 *
 * @see BufferedSum
 * 
 */
 
public class BufferedSum extends BaseOperation implements Buffer
  {
  private Double sum;
  private SpillableTupleList list;
  private Fields extrasSelector;
  private Fields fieldToSum;
 
  /**
   * Returns a BufferedSum Buffer Operation. 
   *
   * @param emittedSumFieldName a {@link Fields} naming the field to emit the sum value
   * @param fieldToSum          a {@link Fields} naming the field to sum
   * @param extrasSelector      a {@link Fields} naming the other fields to "pull through". These fields *must* be of the same order and size as the input Tuple
   */
  public BufferedSum( Fields emittedSumFieldName, Fields fieldToSum, Fields extrasSelector )
    {
    super( extrasSelector.append( emittedSumFieldName ) );
    this.extrasSelector = extrasSelector;
    this.fieldToSum = fieldToSum;
    }
 
  public void operate( FlowProcess flowProcess, BufferCall bufferCall )
    {
    Iterator<TupleEntry> iterator = bufferCall.getArgumentsIterator();
    sum = 0.0D;
    list = new SpillableTupleList( 10000 );
 
    while( iterator.hasNext() )
      {
      TupleEntry arguments = iterator.next(); // must be called
      sum += arguments.getDouble( this.fieldToSum.get(0) );
      list.add( arguments.getTuple() );
      }
 
    for( Tuple tuple : list )
      {
      bufferCall.getOutputCollector().add( tuple.append( new Tuple( sum ) ) );
      }
 
    }
  }

Share:

Its pretty easy once you know how to do it:

• Create a JobConf that defines the parameters for the streaming job
• Send up the hadoop-*-streaming.jar with your cascading job by putting it in your jar
• Send up the scripts (python, in this case) by using the -file option
• Send up any other dependencies, corpora, etc. by using the -file, -cacheFile, or -cacheArchive options (See the Hadoop Streaming page for more details)

Resources

NLTK

To generate the nltkandyaml.mod zip file do the following:

# download nltk and unzip
cd nltk
zip -r nltkandyaml.zip nltk yaml
mv nltkandyaml.zip nltkandyaml.mod

Note that this technique is taken from Cloudera

WordNet

The WordNet zip file needs to be flat. e.g. don’t zip up the files with a subdirectory. You could create this file like so:

# download and unzip the wordnet corpus
cd wordnet
zip -r ../wordnet-flat.zip *

Streaming Script

In python, we’ll be using zipimport.zipimporter to import the nltk libraries from a zip file. In Hadoop 0.20.0, Hadoop didn’t decompress our wordnet-flat.zip file automatically (but we’ve heard reports that it will, but I’m not sure which versions). For us the .zip file was placed in lib relative to the pwd of the script. This allowed us to keep the WordNet corpus as a zip and read it in that format.

wn = WordNetCorpusReader(nltk.data.find('lib/wordnet-flat.zip'))

(In this code we’re not using the python reducer.)

#!/usr/bin/env python 
import os
import re
import sys
import zipimport
 
importer = zipimport.zipimporter('nltkandyaml.mod')
yaml = importer.load_module('yaml')
nltk = importer.load_module('nltk')
punct = re.compile('[^\w\s]+')
 
from nltk.corpus.reader import wordnet
from nltk.corpus.reader import WordNetCorpusReader
 
nltk.data.path += ["."]
wn = WordNetCorpusReader(nltk.data.find('lib/wordnet-flat.zip'))
 
def mapper(args):
  line = sys.stdin.readline();
  try:
    while line:
      line = line.strip()
      word = line
      all_synonyms = []
 
      string_synsets = wn.synsets(word)
 
      for synset in string_synsets:
        synonyms = [lemma.name for lemma in wn.synset(synset.name).lemmas]
        synonyms.pop(0)
        for synonym in synonyms:
          synonym = re.sub("_", " ", synonym)
          all_synonyms.append(synonym) 
 
      print "\t".join([word, ','.join(all_synonyms)])
      line = sys.stdin.readline()
  except "end of file":
    return None
 
# we're not using this, but we could
def reducer(args):
  for line in sys.stdin:
    line = line.strip()
    print line
 
if __name__ == "__main__":
  if sys.argv[1] == "mapper":
    mapper(sys.argv[2:])
  elif sys.argv[1] == "reducer":
    reducer(sys.argv[2:])

Cascading Code

Here’s the bulk of the code that will achieve the effect we want. Like last time, we’re using two intermediate taps as the input and output of the streaming job. Also, we’re just using TextLine files for simplicity. If you don’t want the intermediate files hanging around, look at the comments towards the bottom for some example code on how to remove the files when the job is finished running.

package com.xcombinator.hadoopjobs.cascadingstreamingtest;
 
import cascading.cascade.*;
import cascading.flow.Flow;
import cascading.flow.FlowConnector;
import cascading.flow.MapReduceFlow;
import cascading.operation.aggregator.Count;
import cascading.operation.regex.*;
import cascading.pipe.*;
import cascading.scheme.*;
import cascading.tap.*;
import cascading.tuple.Fields;
import cascading.operation.Identity;
import java.util.Properties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.IdentityMapper;
import org.apache.hadoop.mapred.lib.IdentityReducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;
import cascading.operation.Debug;
import org.apache.hadoop.streaming.StreamJob;
import java.io.IOException;
 
/**
 * An example file to use a Hadoop Streaming job in cascading
 */
public class Main extends Configured implements Tool
  {
  private static final Logger LOG = Logger.getLogger( Main.class );
 
  public int run(String[] args)
  {
    JobConf conf = new JobConf(getConf(), this.getClass());
    Properties properties = new Properties();
    FlowConnector.setApplicationJarClass(properties, this.getClass());
 
    CascadeConnector cascadeConnector = new CascadeConnector();
    FlowConnector flowConnector = new FlowConnector(properties);
 
    String inputPath  = args[0];
    String outputPath = args[1];
    String intermediatePath1 = args[1] + "-mr-input";
    String intermediatePath2 = args[1] + "-mr-output";
 
    Scheme textLineScheme = new TextLine();
 
    Tap sourceTap = new Hfs(textLineScheme, inputPath);
    Tap intermediateTap1 = new Hfs(textLineScheme, intermediatePath1);
    Tap intermediateTap2 = new Hfs(textLineScheme, intermediatePath2);
    Tap sinkTap   = new Hfs(textLineScheme, outputPath);
 
    // create our first flow, sink to the intermediateTap
    Pipe wsPipe = new Each("wordsplit", 
        new Fields("line"), 
        new RegexSplitGenerator(new Fields("word"), "\\s+"), 
        new Fields("word"));
 
    Flow parsedLogFlow = flowConnector.connect(sourceTap, intermediateTap1, wsPipe);
 
    // Create a pipe and set our mr job for it 
    Pipe importPipe = new Pipe("mr pipe");
    Flow mrFlow;
 
    try {
      JobConf streamConf = StreamJob.createJob( new String[]{
          "-input", intermediateTap1.getPath().toString(), 
          "-output", intermediateTap2.getPath().toString(),
 
          // straight unix
          // "-mapper", "/bin/cat",
          // "-reducer", "/usr/bin/wc"
 
          // ruby
          // "-mapper", "src/main/ruby/word_count_mapper.rb",
          // "-reducer", "src/main/ruby/word_count_reducer.rb",
          // "-file", "src/main/ruby/word_count_mapper.rb",
          // "-file", "src/main/ruby/word_count_reducer.rb"
 
          // python
          "-mapper", "python synsets.py mapper",
          "-reducer", "org.apache.hadoop.mapred.lib.IdentityReducer",
          "-file", "src/main/python/synsets.py",
          "-file", "resources/nltkandyaml.mod",
          "-file", "resources/lib/wordnet-flat.zip",
          });
      mrFlow = new MapReduceFlow("streaming flow", streamConf, intermediateTap1,
        intermediateTap2, false, true);
    } catch(IOException ioe) {
       ioe.printStackTrace();
       System.exit(1);
       return 1;
    }
 
    // create our third "regular" cascading pipe. this is a bit contrived, but
    // the idea is substitute all 'e's with 'x's. it's just here to show how to
    // take the input of a streaming job back into cascading
    Pipe subPipe = new Pipe("subber");
    subPipe = new Each(subPipe,
        new Fields("line"),
        new RegexReplace(new Fields("linx"), "e", "x", true),
        new Fields("linx"));
    Flow subFlow = flowConnector.connect(intermediateTap2, sinkTap, subPipe);
 
    Cascade cascade = cascadeConnector.connect(parsedLogFlow, mrFlow, subFlow);
    cascade.complete();
 
    // to get rid of the intermediate files you could do this:
    // Path tmp = tap.getPath();
    // FileSystem fs = tmp.getFileSystem(conf);
    // fs.delete(tmp, true);
 
    return 0;
  }
 
  public static void main(String[] args) throws Exception 
  {
    int res = ToolRunner.run(new Configuration(), new Main(), args);
    System.exit(res);
  }
 
  }

Share:

interval is a tiny library that provides simple musical note pitch and interval arithmetic. It is intended to do one thing: given a pitch add (or subtract) an interval and give the resulting pitch.

Observe:

p = Interval::Pitch.from_string("c")
i = Interval::Interval.from_string("M3")
p2 = p + i
p2.to_short_name # => "e"
 
i.to_s # => "Major Third"
 
i2 = Interval::Interval.from_string("p5")
i2.to_s # => "Perfect Fifth"
 
(p2 - i2).to_s # => "a"

Interval Quiz

interval was written primarily for learning intervals. interval-quiz is a gem that depends on interval that provides a command-line quiz. Here’s the output of an interval-quiz session:

$ interval-quiz
Here are the intervals:
unison  p1        a1
second  m2 M2  d2 a2
third   m3 M3  d3 a3
fourth  p4     d4 a4
fifth   d5 p5  d5 a5
sixth   m6 M6  d6 a6
seventh m7 M7  d7 a7
octave  p8     d8
enter the intervals you want (or a blank line to quit):
M3
p5

["M3", "p5"]
1. above
2. below
3. both
do you want to be quizzed on intervals above, below, or both?  3
what is a major third below f# ? d
correct!
what is a major third above g# 1/1 (100%)? b#
correct!
what is a major third below b 2/2 (100%)? g
correct!
what is a perfect fifth below eb 3/3 (100%)? a
wrong. the answer is ab
what is a perfect fifth below c# 3/4 (75%)? d
wrong. the answer is f#

Installing

gem install interval interval-quiz

Source

http://github.com/jashmenn/interval
http://github.com/jashmenn/interval-quiz

Share:

Share:

However, sometimes you may have code for an existing MapReduce job or want to drop directly to Hadoop for efficiency. Even if you’re using raw MapReduce jobs, Cascading can still be useful in planning the overall data pipeline.

The code below is an example of how to use a raw MapReduce job in a Cascade. The main thing to take away is that we are creating intermediate sinks and sources and relying on Cascading to schedule the flows in the correct order.

NOTE: this code below depends on commit f0dd84cd which is a patch to MapReduceFlow.java that allows you to specifically set the Taps for a MapReduceFlow. I’ve contacted Chris about integrating this into the trunk.

Also note this patch applies to the branch wip-1.1 and later.

package com.xcombinator.hadoopjobs.mapreducetest;
 
import cascading.cascade.*;
import cascading.flow.Flow;
import cascading.flow.FlowConnector;
import cascading.flow.MapReduceFlow;
import cascading.operation.aggregator.Count;
import cascading.operation.regex.*;
import cascading.pipe.*;
import cascading.scheme.*;
import cascading.tap.*;
import cascading.tuple.Fields;
import cascading.operation.Identity;
import java.util.Properties;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.lib.IdentityMapper;
import org.apache.hadoop.mapred.lib.IdentityReducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;
import cascading.operation.Debug;
 
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
 
/**
 * An example file to use a raw MapReduce job in cascading
 */
public class Main extends Configured implements Tool
  {
  private static final Logger LOG = Logger.getLogger( Main.class );
 
  public int run(String[] args)
  {
    JobConf conf = new JobConf(getConf(), this.getClass());
    Properties properties = new Properties();
    FlowConnector.setApplicationJarClass(properties, this.getClass());
 
    CascadeConnector cascadeConnector = new CascadeConnector();
    FlowConnector flowConnector = new FlowConnector(properties);
 
    String inputPath  = args[0];
    String outputPath = args[1];
    String intermediatePath1 = args[1] + "-mr-input";
    String intermediatePath2 = args[1] + "-mr-output";
 
    Scheme textLineScheme = new TextLine();
 
    Tap sourceTap = new Hfs(textLineScheme, inputPath);
    Tap intermediateTap1 = new Hfs(new TextLine(new Fields("line")), intermediatePath1);
    Tap intermediateTap2 = new Hfs(textLineScheme, intermediatePath2);
    Tap sinkTap   = new Hfs(textLineScheme, outputPath);
 
    // create our first flow, sink to the intermediateTap
    Pipe wsPipe = new Each("wordsplit", 
        new Fields("line"), 
        new RegexSplitGenerator(new Fields("word"), "\\s+"), 
        new Fields("word"));
 
    Flow parsedLogFlow = flowConnector.connect(sourceTap, intermediateTap1, wsPipe);
 
    // Create a pipe and set our mr job for it 
    Pipe importPipe = new Pipe("mr pipe");
 
    JobConf mrconf = new JobConf();
    mrconf.setJobName("custom mr");
 
    mrconf.setOutputKeyClass(LongWritable.class);
    mrconf.setOutputValueClass(Text.class);
 
    // the IdentityMapper, in this case, will actually output the key, which is
    // a long of offset in bytes. Not what we'd usually want, but we'll leave
    // it in for now.
    mrconf.setMapperClass(IdentityMapper.class);
    mrconf.setReducerClass(IdentityReducer.class);
 
    // note that your input is straight text-lines. This means in a real mr job
    // you'd most likely need to split the line by some convention
    TextInputFormat format = new TextInputFormat();
    format.configure(mrconf);
 
    // NOTE: this is both here and in the MapReduceFlow below
    FileInputFormat.setInputPaths(mrconf, intermediateTap1.getPath());  
    FileOutputFormat.setOutputPath(mrconf, intermediateTap2.getPath()); // likewise
 
    // create our second flow, this one is for the mrjob. Notice source and sink taps
    Flow mrFlow = new MapReduceFlow("mrflow", 
      mrconf, intermediateTap1, intermediateTap2, false, true);
 
    // create our third "regular" cascading pipe
    Pipe countPipe = new Pipe("count");
 
    // b/c our IdentityMapper is emitting long of offset in the line, just
    // strip that out. You wouldn't have to do this if you had a smarter Mapper
    // class.
    countPipe = new Each(countPipe, 
        new Fields("line"), 
        new RegexParser(new Fields("word"), ".*?\\t(.*)"), 
        new Fields("word"));
 
    countPipe = new GroupBy(countPipe, new Fields("word"));
    countPipe = new Every(countPipe, new Count(), new Fields("count", "word"));
 
    // create the flow for the last count pipe
    Flow countFlow = flowConnector.connect(intermediateTap2, sinkTap, countPipe);
 
    Cascade cascade = cascadeConnector.connect(parsedLogFlow, mrFlow, countFlow);
    cascade.complete();
 
    // if you want to get rid of the intermediate files you 
    // could do something like the following here:
    // Path tmp = tap.getPath();
    // FileSystem fs = tmp.getFileSystem(conf);
    // fs.delete(tmp, true);
 
    return 0;
  }
 
  public static void main(String[] args) throws Exception 
  {
    int res = ToolRunner.run(new Configuration(), new Main(), args);
    System.exit(res);
  }
 
  }

Share:


sudo port install enchant # you'll need MacPorts 1.8.* for this


Then download the dictionaries from the OpenOffice wiki.

Next, place the files in ~/.enchant like so:

$ tree .enchant/
.enchant/
`-- myspell
    |-- en_US.aff
    `-- en_US.dic

UPDATE
You can also set the path to where you want to keep the dictionaries when you compile enchant by passing the --with-myspell-dir option to configure (or aspell, hspell etc).

Share:

Testing multi-process erlang gen_servers can be tricky. Typically one relies simply on pattern matching to verify that the response matches what you would expect.

{expected, Response} = gen_server:call(Pid, hi).

As long as the gen_server call hi returns expected as the first element of the tuple, then the tests pass.

The technique is also the same when building client-server code where both client and server are gen_servers. The common case is to simply test one side at a time; test the response of all client calls and then (independently) test the responses of the server calls.

What about asynchronous cast?

gen_server:call is convenient because it is synchronous and returns a value.
gen_server:cast, on the other hand, is asynchronous and always returns the atom ok. This can make casts difficult to test.

gen_server_mock is a library to mock gen_server processes that expect specific, ordered sets of messages. It allows you to unit test gen_servers by verifying they are receiving the expected set of messages.

Example 1

     {ok, Mock} = gen_server_mock:new(),
     gen_server_mock:expect(Mock, call, fun({foo, hi}, _From, _State) -> ok end),
     gen_server_mock:expect_call(Mock, fun({bar, bye}, _From, _State) -> ok end),
 
     ok = gen_server:call(Mock, {foo, hi}),  
     ok = gen_server:call(Mock, {bar, bye}),  
 
     ok = gen_server_mock:assert_expectations(Mock)

This Mock expects two calls: {foo, hi} and {bar, bye}. Since Mock receives both of these messages, assert_expectations does not raise any errors.

What is verified

gen_server_mock:assert_expectations(Mock) verifies that:

1. all expected messages were received
2. no messages were received that were not expected

You can catch the exit by using the following:

     Result = try gen_server_mock:assert_expectations(Mock)
     catch
         exit:Exception -> Exception
     end,
     % etc...

Example 2

     {ok, Mock} = gen_server_mock:new(),
 
     gen_server_mock:expect_call(Mock, fun(one,  _From, _State)            -> ok end),
     gen_server_mock:expect_call(Mock, fun(two,  _From,  State)            -> {ok, State} end),
     gen_server_mock:expect_call(Mock, fun(three, _From,  State)           -> {ok, good, State} end),
     gen_server_mock:expect_call(Mock, fun({echo, Response}, _From, State) -> {ok, Response, State} end),
     gen_server_mock:expect_cast(Mock, fun(fish, State) -> {ok, State} end),
     gen_server_mock:expect_info(Mock, fun(cat,  State) -> {ok, State} end),
 
     ok = gen_server:call(Mock, one),
     ok = gen_server:call(Mock, two),
     good = gen_server:call(Mock, three),
     tree = gen_server:call(Mock, {echo, tree}),
     ok = gen_server:cast(Mock, fish),
     Mock ! cat,
 
     gen_server_mock:assert_expectations(Mock)

Currently three types of messages are supported: call, cast, and info.

The signature of the fun of each expectation is the same as the corresponding
gen_server:handle_*. So the fun for expect_call has the same signature as handle_call: fun(Request, From, State). See man gen_server for more information.

However, the return value of the fun must be one of:

    ok |                  
    {ok, NewState} |
    {ok, ResponseValue, NewState} |

Anything else will be an error. Note that you can change the state of your Mock by returning NewState.

Arbitrary, non-gen_server messages are handled with expect_info, e.g. Mock ! cat fulfills the expect_info in the example above.

References

• Github Repo (Patches readily accepted)
• Work inspired by this post
• Mocks Aren’t Stubs

Share:

Share:

What you need to do is open up your vmx file and comment out (disable) the following line:

scsi0:0.mode = "independent-persistent"

Others have also reported finding the following line in their vmx file:

snapshot.disabled = "TRUE"

Share:

# Usage: expect install-tripwire.tcl pass1 pass2 
set PASS1 [lindex $argv 0]
set PASS2 [lindex $argv 1]
 
spawn make install
 
expect "Press ENTER to view the License Agreement."
send "\r"
send "q"
expect "license agreement. \[do not accept\]"
send "accept\r"
expect "Continue with installation? \[y/n\]"
send "y\r"
 
expect "Enter the site keyfile passphrase:"
send "$PASS1\r"
expect "Verify the site keyfile passphrase:"
send "$PASS1\r"
 
expect "Enter the local keyfile passphrase:"
send "$PASS2\r"
expect "Verify the local keyfile passphrase:"
send "$PASS2\r"
 
expect "Please enter your site passphrase:"
send "$PASS1\r"
 
expect "Please enter your site passphrase:"
send "$PASS1\r"

Share:

Summary

Setting up a scalable Hadoop cluster isn’t easy, but PoolParty makes it easier
and manageable.

By the time we’re done with this tutorial you’ll have a Hadoop cluster consisting of one master node and two slaves. The slaves are formatted with HDFS and process MapReduce jobs that are delegated to them from the master.

The whole cluster is monitored by Ganglia.

Benefits of PoolParty

The nodes are very interdependent. By that I mean that each node needs to have 2 or 3 configuration files that are based on the other currently running nodes in the cluster. As nodes are joining and leaving the cluster each of these files on every node needs to be updated. PoolParty handles this process for you more-or-less automatically. The benefit is that you don’t have roll your own methods to do this every time you want to setup a cluster.

In PoolParty plugins are first-class citizens. This means you can write your own plugins and they are every bit as powerful as the resources that make up PoolParty core itself. This makes it easy to break up server functionality into modules of code . PoolParty, in a sense, gives you object-oriented server configurations. You can, for instance, take a Ganglia object, call a few methods and PoolParty takes care of executing the required commands to deploy a configured Ganglia cluster.

Architecture

PoolParty is built around the notion of pools and clouds . A pool is simply a collection of clouds. A cloud is a homogeneous set of nodes. i.e. every node in a cloud is configured the same way . Obviously nodes in a cloud will have different sets of working data as they run, but the idea is any node in a cloud could be substituted for any other node in that same cloud.

PoolParty itself is designed to be fully distributed and masterless. There is no required concept of “master” and “slave” in PoolParty itself. That said, many pieces of software, such as Hadoop, do have this concept and PoolParty can be configured to take advantage of that.

We’ll be setting up our pool as two clouds hadoop_master and hadoop_slave. Obviously, hadoop_slave will be a cloud (cluster) of nodes configured to be Hadoop slaves. hadoop_master will also be a cloud of masters. In our example we’re only going to use 1 node as the master. But you could relatively easily configure everything to have more than one master.

Software involved

• Hadoop
• Hive
• Ganglia
• PoolParty

Prerequisites

This tutorial assumes that:

1. You have Amazon EC2 java tools installed. See EC2: Getting Started with the Command Line Tools
2. You have the proper EC2 environment variables setup. See Setting up EC2 on the PoolParty website. For instance, a typical PoolParty install would have these variables in $HOME/.ec2/keys_and_secrets.sh.
3. You have PoolParty installed from source. In theory, you should be able to install the gem. However, today you should probably install from source. Make sure you have git://github.com/auser/poolparty.git checked out and then follow the “Installing” directions on the PoolParty wiki. You only need to complete the two sections Dependencies required to build gem locally and Instructions . This will install all the development dependency gems and then make sure you have all of the submodules. NOTE PoolParty deploys ruby gem versions based on the versions on your local machine. So make sure you have the most recent versions of the required gems installed locally.
4. You have the jashmenn/poolparty-examples repository. git clone git://github.com/jashmenn/poolparty-examples.git /path/to/poolparty-examples
5. You have the jashmenn/poolparty-extensions repository. Note that this directory must be a sibling directory to the poolparty-examples directory. git clone git://github.com/jashmenn/poolparty-extensions.git /path/to/poolparty-extensions

EC2 Security

Now that we have the code issue complete, we now need to deal with Amazon’s security. (See here if you are unclear on how EC2 security works.)

Setup Keypairs

Every cloud in PoolParty must have its own unique keypair. Thats important enough it’s worth repeating: every cloud in PoolParty must have its own unique keypair .

So run the following commands:

ec2-add-keypair cloud_hadoop_slave > ~/.ssh/cloud_hadoop_slave
ec2-add-keypair cloud_hadoop_master > ~/.ssh/cloud_hadoop_master
chmod 600 ~/.ssh/cloud_hadoop_*

Security Groups

You’ll also want to create a security group for our pool .

ec2-add-group hadoop_pool -d "the pool of hadoop masters and slaves"

NOTICE: Hadoop has a crazy number of ports that it requires. The ports below will work but may not be the most secure configuration. If you understand this better than I please recommend better settings. Otherwise proceed knowing that these ports are probably a little too open.

We also need to open a number of ports for this security group:

<code>ec2-authorize -p 22 hadoop_pool               # ssh
ec2-authorize -p 8642 hadoop_pool             # poolparty internal daemons
ec2-authorize -P icmp -t -1:-1 hadoop_pool    # if you want to ping (optional, i guess)
ec2-authorize -p 80 hadoop_pool               # apache
 
ec2-authorize -p 8649 -P udp hadoop_pool      # ganglia UDP
ec2-authorize hadoop_pool -o hadoop_pool -u xxxxxxxxxxxx # xxxxxxxxxxxx is your amazon account id. ugly but true
</code>

Start your cloud

NOTE : There are a number of configurations that rely on the whole cloud being booted. This means that the first time you run cloud-start you may see a few shell errors. This is okay as long as it goes away after subsequent configures. The idea is that all nodes need to be started before the whole configuration will work properly.

cd /path/to/poolparty-examples/hadoop
cloud-list # sanity check, no instances should show up, no exceptions should be raised
cloud-start -vd

Tons of information will fly by. Be patient, this could take upwards of 15 minutes.

Everything done? Good. Now you’re going to need to configure a second time. Now that all the nodes are booted they can be configured to talk to each other properly.

cloud-configure -vd

Again, tons of output should fly by. Wait for it to finish.

Now what we want to do is actually run our hadoop sample job. Open up the hadoop/clouds.rb and find the lines that look like this:

hadoop do
  configure_master
  prep_example_job
  # run_example_job
end

Uncomment the run_example_job line and configure, but this time we only need to configure master.

cloud-configure -vd -c hadoop_master

This should work, but there is a chance the hdfs wont be started in time to load the sample job. If that happens, just configure one more time.
You know it worked if you see output like the following (it wont be at the bottom):

[Fri, 26 Jun 2009 20:09:50 +0000] DEBUG: STDERR: 09/06/26 20:09:11 INFO input.FileInputFormat: Total input paths to process : 3
09/06/26 20:09:12 INFO mapred.JobClient: Running job: job_200906262006_0001
09/06/26 20:09:13 INFO mapred.JobClient:  map 0% reduce 0%
09/06/26 20:09:32 INFO mapred.JobClient:  map 66% reduce 0%
09/06/26 20:09:38 INFO mapred.JobClient:  map 100% reduce 0%
09/06/26 20:09:47 INFO mapred.JobClient:  map 100% reduce 100%
09/06/26 20:09:49 INFO mapred.JobClient: Job complete: job_200906262006_0001
09/06/26 20:09:49 INFO mapred.JobClient: Counters: 17

Congradulations! You now have a scalable Hadoop cluster at your disposal!

What to do when something goes wrong

• Checkout the PoolParty IRC channel, we’re always around and ready to help #poolpartyrb.

This plugin was based on a number of helpful sites on the web. Checkout the following links:

Hadoop

• Michael Noll’s Haddop Tutorial)

Hive

• Apache’s Hive website

Ganglia

• IBM’s Ganglia Tutorial

References

• PoolParty Documentation

Share:

Checkout the screencast.

Share:

multiruby is a great way to make sure your ruby code runs on the multitude of ruby versions (it’s part of ZenTest). It doesn’t install macruby by default. Here are instructions on how to set it up.

Since I installed macruby from the package installer my macruby files are in /Library/Frameworks/MacRuby.framework/Versions/0.4/. If your macruby files are somewhere else, adjust accordingly.

First make sure you already have multiruby setup for other versions (don’t proceed if this doesn’t work):

multiruby_setup the_usual


One problem I had with the above command was that I had RUBYOPT set in my .profile. This was calling each of these ruby versions with RUBYOPT=rubygems when trying to install. This won’t work because rubygems is one of the things you’re trying to install. Make sure you unset this variable etc. before trying to setup multiruby.

ln -s /Library/Frameworks/MacRuby.framework/Versions/0.4/usr ~/.multiruby/install/macruby-0.0.4
sudo ln -s ~/.multiruby/install/macruby-0.0.4/bin/macruby ~/.multiruby/install/macruby-0.0.4/bin/ruby
sudo ln -s ~/.multiruby/install/macruby-0.0.4/bin/macgem ~/.multiruby/install/macruby-0.0.4/bin/gem
touch ~/.multiruby/versions/macruby-0.0.4.tar.gz # fake-out

Now try:

multiruby -e "p 1+1"


Share:

Since I use the command-line twitter gem (by John Nunemaker) I miss out on this feature and find myself using echo and wc to make sure I’m not over the limit.

To fix this problem, I’ve written a small, immediate-feedback utility called tweet. It shows you the number of characters typed so far and colorizes them accordingly.

Screenshots:

warning:

over:

Just download this script and save it as tweet somewhere in your $PATH

Share:

My goal was to temporarily revert one or many commits that I had pushed to the remote server.

http://cheat.errtheblog.com/s/git — specifically the “Fix mistakes / Undo” section was helpful.

What I found:

git revert -n <sha>

#run this for each commit you would like to “undo”

(the -n makes it so that you are not actually creating a commit, but staging the reverse of the changes made by your <sha> commit in your index. git status will show you this)

git ci -a # apply your revision
git push #origin to master (these steps effectively created one commit that was the product of reversing all the commits you picked in the git revert -n step)

now your index looks like:

now, lets say, the time has come to reapply your commits. Because you didn’t just do

git reset --hard <commit3>

or something like that, all you have to do is git reset –hard <sha1> which will “undo your undo”

then

git push #origin master

again and you are back to where you were.

Share:

I want desperately to know how to detach a process from one terminal, and re-attach it to another, without using screen from the get-go. The more I read about it, the more I figure that the answer (you can’t) probably has more to do with my lack of understanding of how processes and terminals work. I read a great post here that introduced me to disown -h (careful) and nohup, some really great bash builtins. I thought, ok, lets try it. This is where I got stuck.

I tried

while true; do sleep 10; done &
disown -h %1
exit

The disown builtin handles a problem with background processes: From the bash man page (and also the above blog)

The shell exits by default upon receipt of a SIGHUP

therefore killing your process you ‘thought’ you put in the background, and then logged out to go home for the night. Disown tells the jobspec not to accept a SIGHUP, and the -h switch tells it to remain in the jobs table. I thought, cool, maybe if it stays in the the jobs table, I could also transfer it to another jobs table of another tty. (no, you can’t…)
but now I had a process on my hands that wasn’t attached to a terminal, and would just run forever unless I rebooted.

The while loop itself didn’t have a process ID, which is interesting, and because of the nature of while, the sleep commands PID kept changing, so a normal ps aux | grep slee[p] | awk '{print $2}' | xargs kill -9 wasn’t working. (This post is loosing topic fast, but the slee[p] in the above command was a cool trick I learned so that I didn’t need a grep -v grep in there).

I *did* find that I could use ps to figure out the ppid (parent process ID) and just kill -9 that, but I was also interested in knowing for sure that it wasn’t in charge of doing something else important. A little digging around, and I came across the UNIX utility pstree which of course didn’t come on my mac, but I quickly figured out that it could be installed with

sudo port install pstree

Yesterday, I had done a similar thing with the UNIX command watch, which also nicely installed using port
And, for those who don’t know, the UNIX command watch is a great poller utility, that will display the first screen’s worth of output of any command, and update it on a regular basis.

I used ps | grep to find the ppid of the sleep process, then ran this command:

watch "pstree $PPID"

This was way cool, as every ten seconds, I watched as the PID of sleep (the child process of this bash process I had just found) changed.

Take away:

sudo port install watch
sudo port install pstree

Share:

shell -bash

Everywhere I read said to put that in your .screenrc file. On my Mac, OS X 10.5, however it seemed to have the (extremely) unwanted circumstance of changing my cwd whenever I started screen to my $HOME folder.

I was able to do some testing my remoting into a *fairly* vanilla fedora box and running the command:

cd /; screen -s -bash

This started bash as a login shell

shopt login_shell #=> login_shell    	on

but did not change my cwd.

I tried moving my .profile file to see if it was the culprit.

I finally came up with this hackish solution.

create file

screen_shell

and place in it the line

$SHELL -l

I knew from other testing that

shell 'bash -l'

didn’t work, course, now, having remembered that I had a similar experience with vi… I wonder if

shell bash\ -l

would work… (nope…)

so, I then changed my .screenrc file to read

shell /Users/bhenderson/.screen_shell

made sure that I did a

chmod u+x ~/.screen_shell

and off I went.

If anyone would like to add to my understanding of what was going on, or a better solution, please feel free to comment. thanks.

Share:

rsync -a /path/to/bar/ /path/to/foo/  # => /path/to/foo/[contents of bar]
rsync -a /path/to/bar /path/to/foo/   # => /path/to/foo/bar/[contents of bar]
 
cp -a /path/to/bar/ /path/to/foo/     # => /path/to/foo/bar
cp -a /path/to/bar /path/to/foo/      # => /path/to/foo/bar

why can’t cp have the same source behavior as rsync?

btw, I found the solution on this blog:

mkdir target_directory
cd source_directory
cp -a . target_directory

(on the blog, the author does cp -ap, but that looks redundant according to the man page)

Share:

def x
  5
end
 
if false
  x = 3
end
 
puts x # => ?

Share:

Erlang has a group of highly optimized list operations in the lists module (Try erl -man lists to see them all). We’re going to be using those to build our Erlang version of #each_with_index.

There are two methods from list we’re going to use: lists:seq and lists:zip. lists:seq simply returns a list of integers between a given range. For instance:

1> Seq = lists:seq(0, 9).
[0,1,2,3,4,5,6,7,8,9]

lists:zip takes two lists (of equal length) and returns a list of tuples. Like so:

2> lists:zip([a,b,c], [1,2,3]).
[{a,1},{b,2},{c,3}]

The last thing we need is a list comprehension, which is similar to each with a block in Ruby. For instance, in Ruby:

>> [:a, :b, :c].each { |element| puts element }
a
b
c
=> [:a, :b, :c]

In Erlang we would do the following:

3> [io:format("~p~n", [Element]) || Element <- [a,b,c]].
a
b
c
[ok,ok,ok]

There is one difference here: the Erlang list comprehension returns a new list which is the value of each call to io:format. In this way an Erlang list comprehension is closer to Ruby’s #collect rather than #each.

We now have all the pieces to build our Erlang version of #each_with_index.

4> EachWithIndex = fun(L, X) -> [
  X(Element, Index) || 
  {Element, Index} <- lists:zip(L, lists:seq(1, length(L)))
   ] end.

Whats going on here?

lists:zip(L, lists:seq(1, length(L)))

creates a list of tuples with each tuple containing an element of L and a companion integer.

[X(Element, Index) || 
  {Element, Index} <- lists:zip(L, lists:seq(1, length(L)))]

tells us to take that list of tuples and do a list comprehension on each of the elements, calling fun X in turn.

EachWithIndex = fun(L, X) -> ... end.

creates a fun (think lambda) and assigns it to EachWithIndex. You could just as easily create this as a regular function in a .erl file somewhere.

Now we can call our new EachWithIndex function by passing in a list and a fun:

5> EachWithIndex([a,b,c,d,e], fun(Element, Index) -> io:format("~p at ~p~n", [Element, Index]) end).
a at 1
b at 2
c at 3
d at 4
e at 5
[ok,ok,ok,ok,ok]

Share:

A new plugin install and mount s3fs volume on an instance using PoolParty.

Getting it

git clone git://github.com/jashmenn/poolparty-s3fs-plugin.git

Usage

NOTICE You must have already created your S3 bucket. This plugin will not
create it for you.

cloud(:app)
  ...
  s3fs(:bucket => "my-fun-bucket")
  ...
end

This will be mounted at /mnt/my-fun-bucket
You can mount multiple buckets and/or change the mount point:

s3fs do
  bucket "my-fun-bucket"
  bucket "my-other-bucket", "/mnt/a-disk"
end

References:

• http://code.google.com/p/s3fs/wiki/FuseOverAmazon
• http://groups.google.com/group/ec2ubuntu/browse_thread/thread/9093236bc07d220b?pli=1

Share:

or git clone git://github.com/jashmenn/poolparty-mrtg-plugin.git

Usage:

Declare mrtg in your cloud block

mrtg do
   monitor :cpu, :uptime, :memory, :open_files, :processes, :apache, :network_interfaces
end

Share: