Buzzword compliance

Over the last few years a number of new technologies have emerged (inspired largely by Google) to help wrangle Big Data.  Things like Hadoop, HBase, Hive, Lucene, Solr and a host of others are becoming the "buzzwords" for handling the type of data that we at the secretariat are working with. As a number of our previous posts here have shown, the GBIF dev team is wholeheartedly embracing these new technologies, and we recently went to the Berlin Buzzwords conference (as a group) to get a sense of how the broader community is using these tools.

My particular interest is in HBase, which is a style of database that can handle "millions of columns and billions of rows".  Since we're optimistic about the continued growth of the number of occurrence records indexed by GBIF, it's not unreasonable to think about 1 billion (10^9) indexed records within the medium-term, and while our current MySQL solution has held up reasonably well so far (now closing in on 300 million indexed records) it certainly won't handle an ever-growing future.

I'm now in the process of evaluating HBase's ability to respond to the kinds of queries we need to support, particularly downloads of large datasets corresponding to queries in the data portal.  As in most databases, schema design is quite important in HBase, as is the selection of a "primary key" format for each table.  A number of the talks at Berlin Buzzwords addressed these issues and I was very happy to hear from some of the core contributers to HBase and their conclusion that figuring out the right setup for any particular problem is far from trivial.  Notable among the presenters were Jean-Daniel Cryans from StumbleUpon (a fellow Canadian, woot!) and Jonathan Gray from Facebook (with luck their slides will be up at the Buzzwords slides page soon).  Jonathan's presentation especially gives a sense of what HBase is capable of given the truly huge amount of data Facebook drives through it (all of Facebook's messaging is held in HBase).

Apart from learning a number of new techniques and approaches to developing with HBase, more than anything I'm excited to dive into the details knowing such a strong and supportive community is out there to help me when I get stuck.  You can follow along my testing and deliberations on the wiki page for our occurrence record project.

Getting started with Avro RPC

Apache Avro is a data exchange format started by Doug Cutting of Lucene and Hadoop fame. A good introduction to Avro is on the cloudera blog so an introduction is not the intention of this post.

Avro is surprisingly difficult to get into, as it is lacking the most basic "getting started" documentation for a new-comer to the project. This post serves as a reminder to myself of what I did, and hopefully to help others get the hello world working quickly. If people find it useful, let's fill it out and submit it to the Avro wiki!

Prerequisites: knowledge of Apache Maven

Start by adding the Avro maven plugin to the pom. This is needed to compile the Avro schema definitions into the Java classes.

<plugin>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-maven-plugin</artifactId>
  <version>1.5.1</version>
  <executions>
    <execution>
      <id>schemas</id>
      <phase>generate-sources</phase>
      <goals>
        <goal>schema</goal>
        <goal>protocol</goal>
        <goal>idl-protocol</goal>
      </goals>
      <configuration>
        <excludes>
          <exclude>**/mapred/tether/**</exclude>
        </excludes>
        <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
        <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
        <testSourceDirectory>${project.basedir}/src/test/avro/</testSourceDirectory>
        <testOutputDirectory>${project.basedir}/src/test/java/</testOutputDirectory>
      </configuration>
    </execution>
  </executions>
</plugin>

Now add the dependency on Avro and the Avro IPC (Inter Process Calls)

<dependency>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro</artifactId>
  <version>1.5.1</version>
</dependency>
<dependency>
  <groupId>org.apache.avro</groupId>
  <artifactId>avro-ipc</artifactId>
  <version>1.5.1</version>
</dependency>

Now we create the Avro Protocol file, which defines the RPC exchange. This file is stored in /src/main/avro/nublookup.avpr and looks like so:

{"namespace": "org.gbif.ecat.ws",
 "protocol": "NubLookup",
 "types": [
     {"name": "Request", "type": "record",
      "fields": [
        {"name": "kingdom", "type": ["string", "null"]},
        {"name": "phylum", "type": ["string", "null"]},
        {"name": "class", "type": ["string", "null"]},
        {"name": "order", "type": ["string", "null"]},
        {"name": "family", "type": ["string", "null"]},
        {"name": "genus", "type": ["string", "null"]},
        {"name": "name", "type": ["string", "null"]}
      ]
     },
     {"name": "Response", "type": "record",
      "fields": [
        {"name": "kingdomId", "type": ["int", "null"]},
        {"name": "phylumId", "type": ["int", "null"]},
        {"name": "classId", "type": ["int", "null"]},
        {"name": "orderId", "type": ["int", "null"]},
        {"name": "familyId", "type": ["int", "null"]},
        {"name": "genusId", "type": ["int", "null"]},
        {"name": "nameId", "type": ["int", "null"]}
      ]
     }  
 ],
 "messages": {
     "send": {
         "request": [{"name": "request", "type": "Request"}],
         "response": "Response"
     }
 }
}

This protocol defines an interface called NubLookup, that takes a Request and returns a Response. Simple stuff.

From the command line issue a compile:

$mvn compile

This will generate into src/main/java and the package I declared in the .avpr file (org.gbif.ecat.ws in my case).

Now we can test it using a simple Netty server which is included in the Avro dependency:

public class Test {
  private static NettyServer server;
  
  // A mock implementation
  public static class NubLookupImpl implements NubLookup {
    public Response send(Request request) throws AvroRemoteException {
      Response r = new Response();
      r.kingdomId=100;
      return r;
    }
  }
  
  public static void main(String[] args) throws IOException {
    server = new NettyServer(new SpecificResponder(
        NubLookup.class, 
        new NubLookupImpl()), 
        new InetSocketAddress(7001)); 

      NettyTransceiver client = new NettyTransceiver(
          new InetSocketAddress(server.getPort()));
      
      NubLookup proxy = (NubLookup) SpecificRequestor.getClient(NubLookup.class, client);
      
      Request req = new Request();
      req.name = new Utf8("Puma");
      System.out.println("Result: " + proxy.send(req).kingdomId);

      client.close();
      server.close();
  }
}

I am evaluating Avro to provide the high performance RPC chatter for lookup services while we process the content for the portal. I'll blog later about the performance compared to the Jersey REST implementation currently running.

MySQL: A speed-up of over 9000 times using partitioning

I wanted to write about a MySQL performance optimization using partitioning as I recently applied it to the Harvesting and Indexing Toolkit’s (HIT) log table. The log table was already using a composite index (indexes on multiple columns), but as this table grew bigger and bigger (>50 million records) queries were being answered at a turtle’s pace. To set things up, imagine that in the HIT application there is a log page that allows the user to tail the latest log messages in almost real time. Behind the scenes, the application is querying the log table every few seconds for the most recent logs, and the effect is a running view of the logs. The tail query used looks like this:

mysql> select * from log_event where id >= ‘latest id’ and datasource_id = ‘datasource_id’ and level >= ‘log level’ order by id desc;

In effect this query asks: “give me the latest logs for datasource with id X having having at least a certain log level”. Partitioning basically divides a table into different portions that are stored and can be queried separately. The benefit is that if a query only has to hit a small portion instead of the whole table, it can be answered faster. There are different ways that you can partition tables in MySQL, and you can read about them all in the MySQL reference manual. I first experimented using Key partitioning using the table ID. Unfortunately, because different logs for a datasource could be spread across different partitions, the tail query would have to hit all partitions. To check how many partitions the query hits, I used the following query:

mysql> explain partitions select * from … ;

This resulted in an even slower response than without partitioning, so Tim thought about it from a different angle. He discovered a nice solution using Range partitioning by datasource ID instead. This way the table would get divided into ranges of datasources that are contiguous but not overlapping. A range size of 1000 was used, so the 1st partition would contain all logs for datasources with IDs between 0 – 999, the 2nd partition would contain all logs for datasources with IDs between 1000 – 1999 and so on. Part of the command used to apply Tim’s partitioning strategy (having 36 partitions) is displayed below:

ALTER TABLE log_event
ADD PRIMARY KEY(id,bio_datasource_fk)
PARTITION BY RANGE (bio_datasource_fk) (
    PARTITION p0 VALUES LESS THAN (1000),
    PARTITION p1 VALUES LESS THAN (2000),
    .
    .
    PARTITION p36 VALUES LESS THAN MAXVALUE
);

Checking how many partitions the tail query would hit, I confirmed that it only ever uses a single partition. The result was impressive, and initial tests resulted in a speed-up of over 9000 times! Important to note is that the primary key must include all fields in the partition. Therefore because we were partitioning using the datasource id, this field had to be included in the primary key before partitioning would work. Also, an index on the id was also added to further optimize the query - why not right? The speed-up might be dramatic now, but as more log messages get written to a partition and it starts to swell, I envisage having to either delete old logs or repartition the table again using smaller range sizes in order to sustain good performance. There is a trade-off between the number of partitions and performance, so some tweaking is needed in every case I guess. Lastly, I’ll reiterate that improper partitioning can actually make things worse. Perhaps it could work for you too, but please apply with caution.

Ordered updates with Postgres

When updating a postgres table you sometimes want the update to happen in a specific order. For example I found myself in a situation when I wanted to assign new sequential ids to records in the alphabetical order given by a text string column.

With postgres 8.4 the solution using an updateable, ordered view didn't work (anymore?). After experimenting a little I found that clustering a table according to the desired order is a simple solution that works exactly as hoped for. Clustering changes the actual order of the table data instead of only adding a new index. And apparently postgres uses this native order for updates.

CREATE TABLE idupd (id int, name varchar(128));

CREATE INDEX idupd_idx ON idupd (name);

CLUSTER idupd USING idupd_idx;

CREATE SEQUENCE idupd_seq;

SELECT setval('idupd_seq', 100);

UPDATE idupd set id=nextval('idupd_seq');

Decoupling components

Recent blog posts have introduced some of the registry and portal processing work under development at GBIF.  Here I'd like to introduce some of the research underway to  improve the overall processing workflows by identifying well defined components and decoupling unnecessary dependencies.  The target being to improve the robustness, reliability and throughput of the data indexing performed for the portal.

Key to the GBIF portal is the crawling, processing and indexing of the content shared through the GBIF network, which is currently performed by the Harvesting and Indexing Toolkit (HIT).  Today the HIT operates largely as follows:

1. Synchronise with the registry to discover the technical endpoints
2. Allow the administrator to schedule the harvest and process of an endpoint, as follows:
1. Initiate a metadata request to discover the datasets at the endpoint
2. For each resource initiate a request for the inventory of distinct scientific names
3. Process the names into ranges 
4. Harvest the records by name range
5. Process the harvested responses into tab delimited files
6. Synchronise the tab delimited files with the database "verbatim" tables
7. Process the "verbatim" tables into interpreted tables

Logically the HIT is depicted:

Some of the limitations in this model include:

1. The tight coupling between the HIT and the target DB mean we need to stop the harvesting when we are going to perform very expensive processing on the database
2. Changes to the user interface for the HIT require the harvester to be stopped
3. The user interface console is driven by the same machine that is crawling, meaning the UI becomes unresponsive periodically.
4. The tight coupling between the HIT and the target DB preclude the option of storing in multiple datastores (as is current desire as we investigate enriching the occurrence store)

The HIT can be separated into the following distinct concerns:

1. An administration console to allow the scheduling, oversight and diagnostics of crawlers
2. Crawlers that harvest the content 
3. Synchronisers that interpret and persist the content into the target datastores  

An event driven architecture would allow this to happen and overcome the current limitations.  In this model, components can be deployed independently, and message each other through a queue when significant events occur .  Subscribers to the queue determine what action if any to take on a per message basis.  The architecture under research is shown:

In this depiction, the following sequence of events would occur:

1. Through the Administration console, the administrator schedules the crawling of a resource.  
2. The scheduler broadcasts to the queue that the resource is to be crawled rather than spawning a crawler directly.  
3. When capacity allows, a crawler will act on this event and crawl the resource, storing to the filesystem as it goes.  On each response message, the crawler will broadcast that the response is to be handled.
4. Synchronizers will act on the new response messages and store them in the occurrence target stores.  In the above depiction, there are actually 2 target stores, each of which would act on the message indicating there is new data to synchronise.

This architecture would have significant improvements to the existing setup.  The crawlers would only ever stop when bug fixing in the crawlers themselves occurs.  Different target stores can be researched independently of the crawling codebase.  The user interface for the scheduling can be developed, and redeployed without interrupting the crawling.  

As an aside, during this exercise we are also investigating improvements in the following:

1. The HIT (today) performs the metadata request, but does NOT update the registry with the datasets that are discovered, only the data portal.  The GBIF registry is "dataset aware" for the datasets served through the Integrated Publishing Toolkit and ultimately we intend the registry to be able to reconcile the multiple identifiers associated with a dataset.  For example, it should be possible in the future to synchronise with the like of the Biodiversity Collections Index which is a dataset level registry.
2. The harvesting procedure is rather complex, with many points for failure; it involves inventories of scientific names, processing into ranges of names and a harvest based on the name ranges.  Early tests suggest a more simpler approach of discrete name ranges [Aaa-Aaz, Aba-Abz ... Zza Zzz] yield better results.

Watch this space for results of this investigation...

The Phantom Records Menace

For a data administrator, going to the web test interface of data publisher can be incredibly useful if one needs to compare the data that was collected using the Harvesting and Indexing Toolkit: HIT and what is available from the publisher. In a perfect world transfer of records would happen without a glitch but when we eventually get less (or more!) than we asked for the search/test interfaces can be a real help (for instance the PyWrapper quering utilities)

Sometimes GBIF will index a resource that for no apparent reason turns in fewer records than what is expected from the line count that the HIT performs automatically. In this particular case there appears to be several identical records on top of that – which we are made aware of by the HIT that warns us that there are multiple records with the same “holy triplet”: Institution code, collection code and catalogue number.

Now what happens when a request goes out for this name range: Abies alba Mill. - Achillea millefolium L. followed by a request for Achillea millefolium agg. - Acinos arvensis (Lam.) Dandy? Those of you with good eyesight will have spotted that the request asks for Achillea millefolium L. before Achillea millefolium agg. This is because this particular instance or configuration of pywrapper returns a name range that is sorted according to the character values you find in UTF-8 and ASCII/Latin-1 which orders all upper-case characters before the lower-case ones. Whether this is an artifact of the underlying database system or the pywrapper itself, or even a specific version of the wrapper is not yet known, but the scenario exists today and consumers should be aware of this. The HIT then builds requests based on this name range and if the requests by chance divide between “Achillea millefolium L. and Achillea millefolium agg.” you will be receiving overlapping responses - that is two responses that contain parts of each other’s records – because the response is not based on a BINARY select statement and therefore returns the records alphabetically sorted without giving precedence to upper-case letters. This behavior can be replicated by going to the pywrapper interface and searching these name ranges. Fortunately the HIT removes the redundant records during the synchronizing process. However, the record count is based on the line count at the point where the records are received from the access point. This is why the record count in the HIT is inflated and as you see this kind of error can be am bit difficult to spot.

2011 GBIF Registry Refactoring

For the past couple of months, I have been working closely with another GBIF developer (and also fellow blog writer) Federico Mendez, on development tasks on the GBIF's Registry application. This post provides an overview of the work being done on this matter.

First, I will like to explain the nuts and bolts of the current Registry application (the one online), and then the additions/modifications it has "suffered" during 2011 (modifications have not been deployed). As stated on The evolution of the GBIF Registry blog post, in 2010 the Registry entered a new stage on its development by moving to a single DB,  enhanced web service API, and a web user interface. On top of this, an admin-only web interface was created so that we could do internal curation of the data inside the Secretariat.

Hibernate's framework was chosen as the preferred persistence framework and the Data-Access-Object (DAO) classes were coded with the HQL necessary to provide an interface to the Hibernate persistence mechanism. The Business tier consisted of several Manager classes that relied on the DAOs to get the required data. These Managers also were the ones responsible for populating the Data-Transfer-Objects (DTOs) so that they could be passed to the Presentation tier. This last tier made use of plain Java Server Pages (JSPs), along with JQuery, Ajax, CSS among others. Then, at the start of this year 2011, a decision was made to improve the application's underlying implementation in some aspects:

1. Use of MyBatis data mapper framework. This involved walking away from Hibernate's Object-Relational Mapping (ORM) approach. Our use of Hibernate involved HQL, adding an extra latency component when converting HQL to SQL, but in MyBatis we use direct SQL mapped statements making it quicker to access the DB. (I will share some benchmarking on my next blog post, to justify this remark)


2. We found out that using a DTO pattern represented somewhat of an overkill for an application that didn't had such complexity at the model level. We could trim some code complexity by passing the model objects straight to the presentation tier. So we did, and all DTOFactories & DTO objects were gone. 


3. Several codebase improvements were introduced mainly by Federico, cutting down huge amounts of lines and making it easier to add new functionality with less effort (e.g. heavy use of Java's generics) 


4. At the web service level, the Struts2 Rest plugin was replaced by the Jersey library. I personally found the Struts2 Rest plugin lacking documentation (1 year ago) so the Registry's use of it was kind of ad hoc. My next blog post will include more reasoning about this decision.


5. We now make use of the Guice dependency injection framework. Beforehand, we were making use of Spring's ability for this. Also, these injections are made through annotations now; with Spring we were using XML based injection. 


6. The Registry project is now divided into different libraries. In particular: 

• registry-core: Business & persistence logic
• registry-web: All related to the web application (Struts2)
• registry-ws: All the web service stuff
• There are also some libraries Federico has created to manage the interaction between the Registry and all technical installations (DiGIR, Tapir, BioCase, etc) of those publishers sharing data with GBIF. These are extremely important libraries as they are the ones who keep the Registry up to date.

(2011 refactoring)

I must emphasize again that these changes are not yet deployed, this in an ongoing project but if you are really interested to see the progress being made, please feel free to visit the project's site. Also, these changes won't affect the current web services API, or the DB structure. Merely the changes are to improve the underlying codebase.