Reproducing Spatial Joins using Hadoop and EC2

Finally I got some time to continue researching the use of Hadoop for the processing of the GBIF dataset with respect to the protected areas of the world.

The problem
There is a growing number of point data, and an increasing need to perform cross referencing of the GBIF data with external sources, in this case the protected areas of the world.

The last 3 years of working on reasonably large databases (>100G) has told me that this join was not going to be an easy one, despite the obvious partitioning strategy that can be invoked here (spatial partitioning).
Therefore Javi and I set about this task by

1. Using PostGIS and the traditional approach of joining 2 tables of 150M point records with 120,000 polygon records - finding will be posted shortly
   
2. Using Hadoop and the Amazon EC2 to process the join

Using Hadoop
This is ongoing but I thought I'd post the findings so far.

Strategy 1: 

Partition by cell
Start by splitting the 150M records from one file into a file per 1 degree cell.
Loop over each polygon:

• Determine 1 degree cells covered polygon
  
• Use the input files as a merge input to hadoop and pass in the Polygon for the Map to do the contains()
  

Result  

Hadoop will not handle so many file splits no matter what configuration I try so far. Mailing list responses suggest this is not the ideal use of Hadoop and it prefers single large files, rather than many smaller ones.

Amendment

Using 10 degree cells, it does work but takes a LONG time... so long I killed it. This was expected since it is really the equivalent of only a DB join with an index on an INTEGER teg_deg column.

Strategy 2: 

1-Degree in-memory index of polygons
So this approach goes through the join the opposite way. The polygons are loaded into memory and a

HashMap<Integer, Collection<Polygon>>

for each Map operation to use.  

The logic is effectively give me polygons intersecting the 1 degree square for this point, and then test them properly. This worked ok, but requires a LOT of memory for the polygons to be held.

Result

About 6 hours to process 60,000 polygons.

Strategy 3: 

RTree index of polygons
Very similar to strategy 2, this approach indexes all the polygons by RTree. Thus for each point, the index provides only candidate polygons whose bounding box contains the point, and the number of candidate polygons to test is far smaller than that of strategy 2.

Result

45 minutes for a 130M x 30,000 cross reference but requires the larger EC2 instance sizes. The RTree index means it will need processed in batches.
Hadoop requires tuning, just like a database does for large joins. Here is the final tuning for strategy 3:

• mapred.child.java.opts=-Xmx1G // for the R-Tree index
• mapred.output.compress=false // for easy reading of output
• mapred.tasktracker.map.tasks.maximum=4 // we have 7G per node, so 4G for 4 Map tasks is ok
• mapred.job.reuse.jvm.num.tasks=-1 // Reuse the Map as it has the index built (takes 90 seconds for startup)
• dfs.block.size=134217728 // use larger blocks - in retropsect this was a probable mistake and will be removed for next test

More to follow...

Thanks to the Hadoop-core Users mailing list for their insights, and Andrea of OpenGEO for pointing me at the JTS RTree implementation

Cascading Hadoop

When running MapReduce stuff, it very quickly becomes apparent that simple jobs need queued up then an operation run on the output of another, or perhaps an operation run on multiple outputs from another - almost like a cascaded effect one could say... Enter Cascading, a project that aims to build fault tolerant workflows on top of Hadoop.

I have only just started to play with it (similar to Hadoop it is not in the public maven repositories, so of course I wrote my own pom to do a local install). It is a new project and a small team (1 guy?) but it looks promising, although I think it misses a few more getting started examples for common operations - but I figure if people blog about it like here, the examples will pop up pretty quickly.

One of the nice features that immediately attracted me to it was the fact that it can do visualisations of the workflow for you like so (this is just an example workflow, not the code below):


And here is the example. I am using my standard subset of GBIF data, and grouping together the records by scientific name and then sorting them on resource and basisOfRecord (some databases can't mix group by and order by in SQL without temp table creation, so this seemed like a nice example).

 public void run(String input, String output) {
  // source is the input file - here I read from local file system
  // (e.g. not the distributed fs)
  Tap source = new Lfs( new TextLine(), input );
  
  // sink is the output file - here I write to local file system
  // (e.g. not the distributed fs)
  Tap sink = new Lfs( new TextLine(), output, true );
  
  // my tab file schema
  Fields dwcFields = new Fields( "resource", "kingdom", "phylum", "class", "order", "family",
    "genus", "scientificName", "basisOfRecord", "latitude", "longitude" );
  
  // parse the data
  Pipe pipe = new Each( "parser", new Fields( "line" ), new RegexSplitter(dwcFields));
  
  // define some group and sort fields
  Fields groupFields = new Fields("scientificName");
  Fields sortFields = new Fields("resource", "basisOfRecord");
  
  // a group by with a sort...
  // note that this takes the previous pipe
  pipe = new GroupBy(pipe, groupFields, sortFields);

  // connect the assembly to the SOURCE and SINK taps
  Flow parsedLogFlow = new FlowConnector().connect( source, sink, pipe );
  
  // start execution of the flow (either locally or on the cluster)
  parsedLogFlow.start();

  // block until the flow completes
  parsedLogFlow.complete();  
 }

So this was very simple, and it was only the first night playing. Note that this code does not have a mention of a MapReduce job, or anything more complex than a simple tap, pipe, sink workflow,,,

I will proceed by trying to do a much more complex workflow - I think splitting the world data into the 2.8 grids I proposed earlier (6 zoom levels), followed by doing some breakdowns for various analytics I anticipate producing. Then I will report back with metrics running from EC2.
What I would really like to do, is have some nice metadata that accompanies the datafiles at each process that gives the semantics of the file - e.g. something that describes the columns in the tab file, so I expect to use the TDWG vocabularies and do some RDF (perhaps RDF represented as JSON?) This way I can set up the Fields automatically, based on the content of the file, and accept different input formats.

Hadoop on Amazon EC2 to generate Species by Cell Index

I finally took the plunge onto the Amazon Elastic Cloud to do some proper distributed processing using Hadoop after spending the past few weeks playing with MapReduce and distributed processing.

This is a post about my experiences and lessons, but for those who just want the result:

Macbook Pro, 2G JVM, 1 node Hadoop, species per 1 degree cell for all GBIF occurrences: 2449 secs
EC2 20 small instances, species per 1 degree cell for all GBIF occurrences: 472 secs

And the details:

This experiment was to simply generate the Species per Cell (1 degree x 1 degree) index for the entire GBIF occurrence record store (135M record index used) and run some comparisons for local versus cloud execution.

So the code (About 20 lines of 'real' code and maybe it could be optimised further - the reduce for example):

/**
* Generates the species by cell index
* @author timrobertson
*/
public class SpeciesByCell extends MapReduceBase
implements Mapper,
Reducer {

  // assuming this is a large object as they reuse it in the tutorials...
  private final static IntWritable cellId = new IntWritable();
  private Text speciesMap = new Text();
  private Text speciesReduce = new Text();

  // reuse the pattern for performance
  private Pattern tabPattern = Pattern.compile("\t");

/**
 * Outputs Cell:Species
 */
public void map(LongWritable key, Text value,
    OutputCollector output, Reporter reporter) {
  String data = value.toString();
  if (data != null) {
  
    // sci name is column 2, lat 11, long 12
    // split is not the best way to split a string
    String parts[] = tabPattern.split(data);
    if (parts.length>= 12) {
        try {
          cellId.set(SpeciesByCell.toCellId(Float.parseFloat(parts[10]), Float.parseFloat(parts[11])));
          speciesMap.set(parts[1]);
          output.collect(cellId, speciesMap);
        } catch (NumberFormatException e) {
        } catch (UnableToGenerateCellIdException e) {
        } catch (IOException e) {
        }
    }
  }
}

/**
 * Distincts the species
 */
public void reduce(IntWritable key, Iterator values,
    OutputCollector output, Reporter reported) throws IOException {
  Set species = new HashSet();
  while (values.hasNext()) {
    species.add(values.next().toString());
  }
  for (String s : species) {
    speciesReduce.set(s);
    output.collect(key, speciesReduce);
  }

}

/**
 * Gives a cell id
 */
public static int toCellId(Float latitude, Float longitude) throws UnableToGenerateCellIdException {
  if (latitude== null
      || latitude < -90         || latitude > 90
      || longitude < -180         || longitude > 180) {
    throw new UnableToGenerateCellIdException("Latitude["+ latitude+"], Longitude["+longitude+"] cannot be " +
        "converted to a cell id");
  } else {
    int la = new Double(Math.floor(latitude + 90)).intValue();
    int lo = new Double(Math.floor(longitude + 180)).intValue();
    int cellId = (la * 360) + lo;
    return cellId;
  }
}
}

Hadoop says it supports GZip files - so I GZipped the input data, which is simply 13 columns of DwC. Unzipped, the data is 13G, gzipped 972M.
Following uploading and running however, I found some pretty unusual results. Falling back to my favourite invasive (Passer domesticus) to run a smaller test I found that on GZipped the map worked on 246,768 rows, not the full 900,000+. Therefore, I conclude that I am either not running it right, or it does not support GZipped input. I don't see how the distributed chunking could work if it is GZipped, and looking through the docs I think it is only on the output that you can use GZip. So, I then started again, and extracted the full 13G and got that on the HDFS for processing.

Now, since I am a maven user, I need to build up an executable jar; here is the pom section that sets the manifest:

<build>
  <defaultGoal>package</defaultGoal>
  <plugins>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-jar-plugin</artifactId>
      <configuration>
        <archive>
          <manifest>
          <mainClass>com.ibiodiversity.index.mapreduce.SpeciesByCell</mainClass>
          <packageName>com.ibiodiversity.index.mapreduce</packageName>
          </manifest>
        </archive>
      </configuration>
    </plugin>
  </plugins>
</build>

Now, we put the input file onto S3 for use in EC2:
bin/hadoop fs -put /path/to/source s3://<id>:<secret>@<bucket>/path/to/target

(Ok, do NOT do s3://<id>:<secret>@<bucket>/ - it blew up when I tried to move the file out of S3 onto a local HDFS since it was in the root. My second attempt (of 2 hours transfer) was to ibiodiversity-gbif-dwc/dwc/allData)

Of course, my has a / in it, so the Hadoop client blows up even when they are escaped as %2f... in the end I got the hadoop code into eclipse and hacked it directly.

Uploading to S3 using the hadoop FSShell take a LONG time from my house! 3 minutes per 32meg = 186kb/s
Now that it is on S3, I can run many index generations without this latency, but each time there is a version, it would need updated. Clearly it will be better to harvest straight to S3 (perhaps in pages?).

1.5 hours (of hoping the network stays alive) later... (make that 3.5 hours total waiting time, but it is there for next time)

Now I fire up a master instance (standard Hadoop 0.17.0 AMI), connect to it, and copy across the input data using:
cd /usr/local/hadoop-0.17.0
bin/hadoop fs -mkdir logs
bin/hadoop distcp s3://:@/path/to/logs logs

The dreaded / problem again!!!
This time no way to easily hack around it, so top tip: keep reproducing your secret key until it has no /_- characters in it!!!
Now the copy works fine.

Copy up my jar file to the master (run locally)

. bin/hadoop-ec2-env.sh
scp $SSH_OPTS /tmp/speciesByCell.jar root@$MASTER_HOST_DNS:

So, as always we need a benchmark from single node.
Macbook Pro 2.4G, 2G JVM Memory:

org.apache.hadoop.mapred.Counters Counters: 11
org.apache.hadoop.mapred.Counters   Map-Reduce Framework
org.apache.hadoop.mapred.Counters     Map input records=134475707
org.apache.hadoop.mapred.Counters     Map output records=106297454
org.apache.hadoop.mapred.Counters     Map input bytes=14209537679
org.apache.hadoop.mapred.Counters     Map output bytes=2406590553
org.apache.hadoop.mapred.Counters     Combine input records=106297454
org.apache.hadoop.mapred.Counters     Combine output records=1130567
org.apache.hadoop.mapred.Counters     Reduce input groups=41845
org.apache.hadoop.mapred.Counters     Reduce input records=9565669
org.apache.hadoop.mapred.Counters     Reduce output records=7259291
org.apache.hadoop.mapred.Counters   File Systems
org.apache.hadoop.mapred.Counters     Local bytes read=3038621493011
org.apache.hadoop.mapred.Counters     Local bytes written=73325771174

Finished in 2449 secs!

And 20 Node cluster:

08/06/29 17:31:04 INFO mapred.JobClient: Counters: 17
08/06/29 17:31:04 INFO mapred.JobClient:   File Systems
08/06/29 17:31:04 INFO mapred.JobClient:     Local bytes read=298092555
08/06/29 17:31:04 INFO mapred.JobClient:     Local bytes written=597267841
08/06/29 17:31:04 INFO mapred.JobClient:     HDFS bytes read=14210393954
08/06/29 17:31:04 INFO mapred.JobClient:     HDFS bytes written=53736532
08/06/29 17:31:04 INFO mapred.JobClient:   Job Counters
08/06/29 17:31:04 INFO mapred.JobClient:     Launched map tasks=231
08/06/29 17:31:04 INFO mapred.JobClient:     Launched reduce tasks=1
08/06/29 17:31:04 INFO mapred.JobClient:     Data-local map tasks=207
08/06/29 17:31:04 INFO mapred.JobClient:     Rack-local map tasks=5
08/06/29 17:31:04 INFO mapred.JobClient:   Map-Reduce Framework
08/06/29 17:31:04 INFO mapred.JobClient:     Map input records=134475707
08/06/29 17:31:04 INFO mapred.JobClient:     Map output records=106297454
08/06/29 17:31:04 INFO mapred.JobClient:     Map input bytes=14209537679
08/06/29 17:31:04 INFO mapred.JobClient:     Map output bytes=2406590553
08/06/29 17:31:04 INFO mapred.JobClient:     Combine input records=104849073
08/06/29 17:31:04 INFO mapred.JobClient:     Combine output records=780651
08/06/29 17:31:04 INFO mapred.JobClient:     Reduce input groups=41845
08/06/29 17:31:04 INFO mapred.JobClient:     Reduce input records=9416965
08/06/29 17:31:04 INFO mapred.JobClient:     Reduce output records=7259291

Finished in 472 secs!


None of these were tuned in anyway (Mappers, Reducers, Block Size etc)

Cost of running: well, less than 1 hour so ~$2 (20 x $0.1 + a little for transfers)

(Thanks to Tom White for publishing guidelines on Hadoop on EC2 here on which I based this test.)

Mapreduce for species in a Protected Area

The World Commission on Protected Areas (WCPA) publish shapefiles for protected areas, so I thought I'd run these against the GBIF index and get some species lists.
Since I was running on the full 135 million record GBIF index, and the IUCN Categories I-VI National shapefiles containing 60,753 polygons, I went back to Hadoop running in single node mode (i.e. the simplest of simple) as I am not sure the Mapreduce 'Lite' I wrote would cut it.

Input file: 135 Million records of 12 DwC fields (13G)
Stack: Hadoop for Mapreduce, Geotools for Polygon

I went for the brut force approach; the Map held a List<Polygon> and I looped each time over the polygons for each GBIF point record (60,000 x 135M = 8,100,000,000,000 comparisons ;o)

Of course this did not perform... but I wanted to post some benchmarks:

Reference: A Map that does nothing, and the GBIF index input: 500secs
Per Polygon: 1500secs to produce the species by polygon
(Note: This is running Mapreduce in a non clustered, single server environment)

I tried this using the polygon bounding boxes only, to remove Geotools from the equation in each Map operation and the results were the same.

It is clear that to do this kind of processing, the data needs to be sliced up to reduce the number of combinations.
I am pondering using the same tiling algorithm that the map guys seem to have settled upon (e.g. GE superoverlay style). By doing this, a preselect based on the intersect of the protected area bounding box with the tiled data would result in massively reduced processing. I am thinking 7 zoom levels resulting in 32768 distinct tiles so therefore working only on 2.8x2.8 degree cells. Currently GBIF models to 1x1 degree and 0.1x0.1 degree cell but this does not easily port to Mapreduce for partitioning purposes of the input data.
(As a side effect, the tiles could be processed in parallel for specific mapping views ready for immediate serving through Geoserver...)

MapReduce 'Lite'

I have been playing recently with the Hadoop distributed filesystem which ships with a MapReduce framework for processing large volumes of 'point' style biodiversity data.  The results were reasonably promising, but Hadoop is designed to process LARGE quantities of data (terrabytes) and not the mediocre few 100G that I am processing.  Futhermore, Hadoop is designed to be run on multiple machines and not the 'dual node on one machine' that I was running.  The MapReduce idea however really does fit the bill nicely for what I want to process, and therefore I went looking for a MapReduce 'lite' in Java but alas, it seems only geared for enterprise development.  So I started coding...

Inside iBiodiversity you will find a simple MapReduce implementation in Java.  Working on 11 million point occurrence records, my implementation generates the species per 1x1 degree cell index.  This is necessary to efficiently allow UI offering a clickable Map that pulls up the distinct species the system has data for.  Tuning parameter wise, you can configure the size of the pages to be worked on (P) and the in memory sort size (M) before a temporary file is written.  I found P:M of 10:1 was producing the best results, with the processing 11M records taking 175s with only 256M of process memory.  

This looks fairly promising as a means of quickly writing code that can run in parallel manner to process data into a format for custom views of biodiversity data (Species by cell, cells by species, cells by month by species, species by cell by decade, species interaction at the same time / place, aggregate counts etc etc).  If it can work on my MapRecuce 'Lite' then we know as data grows it will port to Hadoop easily.  

What next?  It will go into some index (H2, SOLR, Compass looking likely candidates and all semi tested already in iBiodiversity) with a service on top for Javi to do some UI magic