Hive 0.9 with HBase 0.90

Hive 0.9.0 was released at the beginning of this month and it contains a lot of very nice improvements. Thanks to all involved!

Unfortunately it drops compatibility with HBase 0.90.x due to two issues which introduced a dependency on HBase 0.92:

• https://issues.apache.org/jira/browse/HIVE-2748
• https://issues.apache.org/jira/browse/HIVE-2764

Fortunately these were relatively easy to revert so that's what we did because we wanted to all the 0.9.0 goodness on our HBase 0.90.4 cluster (CDH3u3).

I've forked Hive on Github and reverted the parts of those two issues (HIVE-2748, HIVE-2764) that were causing problems.

For all those "stuck" with HBase 0.90 (e.g. CDH3 users) we've also deployed this custom Hive HBase Handler to our own Maven repository and will maintain that for the foreseeable future. You can just download the jar file and use it in your projects or use our Maven repository:

  gbif-thirdparty
  http://repository.gbif.org/content/repositories/thirdparty

And then declare a dependency on this custom HBase Handler:

  org.apache.hive
  hive-hbase-handler
  0.9.0
  hbase-0.90-compat

Note for Maven experts: I'm not sure if this is a valid classifier. I couldn't find any naming rules.

That's it! We might maintain this for future versions of Hive but we'd love to hear about any problems in any case.

HBase Performance Evaluation continued - The Smoking Gun

Update: See also part 1 and part 3.

In my last post I described my initial foray into testing our HBase cluster performance using the PerformanceEvaluation class.  I wasn't happy with our conclusions, which could largely be summed up as "we're not sure what's wrong, but it seems slow".  So in the grand tradition of people with itches that wouldn't go away, I kept scratching.  Everything that follows is based on testing with PerformanceEvaluation (the jar patched as in the last post) using a 300M row table built with PerformanceEvaluation sequentialWrite 300, and tested with PerformanceEvaluation scan 300.  I ran the scan test 3 times, so you should see 3 distinct bursts of activity in the charts.  And to recap our hardware setup - we have 3 regionservers and a separate master.

The first unsettling ganglia metric that kept me digging was of ethernet bytes_in and bytes_out.  I'll recreate those here:

Figure 1 - bytes_in (MB/s) with single gigabit ethernet

Figure 2 - bytes_out (MB/s) with single gigabit ethernet

This is a single gigabit ethernet link, and that is borne out by the Max values, which are near the theoretical max of 120 megabytes per second (MB/s), and as you can see they certainly aren't sitting at 120 constantly in the way one would expect if ethernet was our bottleneck. But that bytes_in graph sure looks like it's hitting some kind of limit - steep ramp up, relatively constant transfer rate, then steep ramp down.  So I decided to try channel bonding to effectively use 2 gigabit links on each server "bonded" together.  It turns out that's not quite as trivial as it may seem, but in the end I got it working following the Dell/Cloudera configuration suggestion of mode 6 / balance-alb, which is described more in the Linux kernel bonding documentation. The results are shown in the following graphs:

Figure 3 - bytes_in (MB/s) with dual (bonded) gigabit ethernet

Figure 4 - bytes_out (MB/s) with dual (bonded) gigabit ethernet

Unfortunately the scale has changed (ganglia ain't perfect) but you can see we're now hovering a bit higher than the single ethernet case, and at some points swinging significantly above the single ethernet limit (a few gusts up to 150).  So it would seem that even though the single gigabit graphs didn't look like they were hitting theoretical maximum, they were definitely limited.  Not by much though - maybe 10-15%?  As they say, the proof is in the pudding, so what was the final throughput for the two tests?

Ethernet	Throughput (rows/s)
Single	340k
Double	357k

Looks like 5%. Certainly not the doubling I was hoping for! Something else must be the limiting factor then, but what?  Well, what does CPU say?

Figure 5 - cpu_idle with dual (bonded) gigabit ethernet

That CPU is saying, "Feed me Seymour!".  If the CPU is hungry, what about the load average?

Figure 6 - one minute load average with dual (bonded) gigabit ethernet

These machines have 8 hyper-threaded cores, so effectively 16 cpus. The load average looks like it's keeping the CPUs fed - hovering around 20. But hang on - the load average isn't all about CPU - it's about work that the whole machine needs to do - including all the io subsystems. So if it ain't network, and it ain't CPU then it's gotta be disks. Let's look at how much time the CPU spent waiting on io (which in this case basically means "waiting on disks"):

Figure 7 - percentage cpu spent in wait_io state, with dual (bonded) gigabit ethernet

From a pure gut perspective, that seems kind of high.  But on closer inspection, that graph also looks familiar - it's basically the same as Figure 6 - the load average. So what can we deduce?  When ethernet is removed as a limiting factor the run queue is filled with processes that all cause an increase in wait_io - which means our processes would all finish faster if they didn't have to wait for io.  The limiting factor must, then, be the disks.

Recap (TL;DR)

Ethernet looked suspiciously capped, but performance (and ethernet usage) only increased slightly when the cap was lifted. Closer inspection of the resulting load average and CPU usage showed that the limiting factor was in fact the disks.  

Solution?  Replacing the disks might help a little - they're 3 year old 2.5" SATA, but they are 7.2k rpm, and a more modern 10k 2.5" would be marginally faster.  The real benefit would probably be more disks - the current wisdom on the mailing lists and in the documentation is to maximize the ratio of disk spindles to cpu cores and I think that's borne out by these results.  In the coming weeks we're going to build a new cluster of 6 machines with 12 disks each and I very much look forward to testing my theories once they're up and running.  Stay tuned!

Performance Evaluation of HBase

Update: See also followup posts: part 2 and part 3.

In the last post Lars talked about setting up Ganglia for monitoring our Hadoop and HBase installations.  That was in preparation for giving HBase a solid testing run to assess its suitability for hosting our index of occurrence records.  One of the important features in our new Data Portal will be the "Download" function that lets people download occurrences matching some search criteria and currently that process is a very manual and labour intensive one, so automating it will be a big help to us.  Using HBase it would be implemented as a full table scan, and that's why I've spent some time testing our scan performance. Anyone who has been down this road will probably have encountered the myriad opinions on what will improve performance (some of them conflicting) along with the seemingly endless parameters that can be tuned in a given cluster.  The overall result of that kind of research is: "You gotta do it yourself".  So here we go.

Because we hope to get some feedback from the HBase community on our results (ideally comparisons to other running clusters) we decided to use the PerformanceEvaluation class that ships with HBase.  Unfortunately it's not completely bug free so I've had to patch it to work in the way we would like.  From a stock cdh3u3 HBase I patched HBASE-5401 and HBASE-4688, and had a good, hard think about HBASE-5402.  There are a bunch of things I learned about the PerformanceEvaluation (PE) class in the last few weeks, and the implications of 5402 are some of them.  If anyone's interested I'll blog about those another time, but for now suffice it to say that in my final suite of tests I went with writing tables using "sequentialWrite" (which obviates the difficulties in 5402) and doing scans with the "scan" test.

There are many variables that can be changed in an HBase/Hadoop cluster setup but you have to start somewhere so here's our basic config that remained unchanged throughout the test (you can see more at its ganglia page):

Master (c1n2): HDFS NameNode, Hadoop JobTracker, HBase Master, and Zookeeper
Slaves (c2n1, c2n2, c2n3): HDFS DataNode, Hadoop TaskTracker, HBase RegionServer (6 GB heap)

Hardware:
c1n2: 1x Intel(R) Xeon(R) X3363 @ 2.83GHz (quad), 8GB RAM, 2x500G SATA 5.4K
c2n*: 2x Intel(R) Xeon(R) E5630 @ 2.53GHz (quad), 24GB RAM, 6x250G SATA 7.2K

The parameters I varied are:

• number of rows in the test table (50M, 100M, 200M)
• number of mappers per server (8, 10, 12, 14, 20 - more than 20 and we ran out of RAM)
• snappy compression (on/off)
• scanner caching (size)
• block caching (on/off)

Test Methodology

The test methodology was to start with an empty HDFS (replication factor of 2), an empty HBase and no other tasks running on the machines.  Then build a table using a command line like 

hbase org.apache.hadoop.hbase.PerformanceEvaluation sequentialWrite 100

and then wait for the split/compaction cycle to complete, and then run a major compaction to encourage data locality.  In tests using compression I would construct an empty table by hand, specifying compression=>'SNAPPY' for the info column family before running the sequentialWrite. I'd also wait for the region balancer to run, ensuring that all regionservers had an equal number of regions (+/- 1). Once that was complete I would run a scan test with a command like:

hbase org.apache.hadoop.hbase.PerformanceEvaluation scan 100

and note the real time taken as reported by the JobTracker as well as the total time spent in mappers as reported by PE itself.  I ran each scan test 3 times unless the first 2 tests were very close, in which case I sometimes skipped the 3rd run. Note also that PE uses 10 byte keys and 1000 kilobyte values by default. 

Two things were immediately obvious in my caching tests, which I'll talk about briefly before getting into the results proper.

Scanner Caching

The HBase API Scan class takes a setting called setCaching(int caching) which tells the scanner how many rows to fetch at a time from the server.  This defaults to 1, which is optimal for single gets, but is decidedly suboptimal for big scans like ours. Unfortunately PE doesn't allow for explicitly setting this value on the command line and so I took the advice of the API Javadoc and passed in a value of 1000 for the property hbase.client.scanner.caching in the hbase-site.xml file that forms the configuration for the MapReduce job that is PE.  It would seem, though, that this property is not being honoured by HBase and so my initial tests all ran with a cache size of 1, producing poor results.  I subsequently hard-coded the setCaching(1000) on the scans themselves in the PE and saw significant improvements. Those differences are visible in the results, below.

Block Caching

When I say block caching I mean HBase blocks - not HDFS blocks.  See the HBase guide or Lars George's excellent HBase: The Definitive Guide for many more details. You'll see many recommendations from different sources to turn block caching off during scans but it's not immediately obvious why that should be.  It turns out that it has no performance implications for the scan itself, but will significantly impact any random reads (gets) that may be relying on recently loaded blocks.  If the scan is using the block cache then it will load the cache with a block, read the block, and then load the next one, ignoring the first one for the rest of the scan.  The earlier loaded blocks will be evicted quickly (as new blocks are loaded) leading to "cache churn".  So I tested the block cache on and off and concluded that there was no difference in a dedicated scan test (no other activity in the cluster) and so for the rest of my tests I left the block cache on.  In order to test this behaviour I again had to hard-code PE with setCacheBlocks(false) (it defaults to true).

The Results

The full spreadsheet is available for download from our google code site.  I've just included the summary chart here, because it pretty much tells the story.  Of note: the y-axis is Records / second which I calculate as the total records scanned divided by the total time for the job as reported by the JobTracker.  This doesn't, therefore, take any of the PE generated per-mapper counts/times into account (which are still somewhat mysterious to me).

Conclusions (within our obviously limited test environment):

• with bigger datasets, more mappers are better; with smaller datasets, somewhat fewer are better
• compression slows things down.  This is probably the most surprising (and contentious) finding.  We would expect performance to improve with compression (at worst staying the same) because there is much less data to transfer when the files are compressed, and I/O is typically the limiting factor in HBase performance.  I kept an eye on the cpus with ganglia during the uncompressed tests and sure enough, they spent approximately double the amount of time in io_wait as in the compressed tests.
• ganglia revealed no obvious bottlenecks during the scan.  You can look at our ganglia charts from the past few weeks and apart from a few purposefully degenerate tests there was no obvious component that was limiting things - cpus never went above 60% (io_wait never above 30%), and RAM and disks "looked ok".  If anyone has insight here we'd really appreciate it.
• total variability across all these tests was only about 20%.  Either all tests are hitting the same, as yet unseen, hardware or configuration limit, or there really is no silver bullet to performance increases - tuning is an incremental thing.
• there are still strange things going on :)  I can't explain the fact that in the compressed test performance increases from 50 to 100M records, and then decreases dramatically with 200M records.

We'd love feedback on these results, both in terms of how we can improve our testing and in interpreting the results, as well as any numbers you may have from your clusters when running PE.  Is 300k records per second even in the ballpark?  It's crazy that these data don't exist out there already, so hopefully this post helps alleviate that somewhat.  

The good news is that we'll be getting new cluster hardware soon, and then I think the next step is to load our real occurrences data into a table and run similar scanning tests against it to optimize those scans.  Of course then we'll start writing new records at the same time, and then things will get.... interesting.  Should be fun :)

Monitoring Hadoop and HBase

We're getting serious in our Hadoop adoption. The first process (our so called "rollover") is now in production and it uses Hadoop, Hive, Oozie and various other parts of the Hadoop ecosystem.

Our next step is evaluating HBase and its performance on our (small and aging) cluster. To do that properly and to fix a rather embarrassing situation we first had to get proper monitoring up and running for our cluster. So far we've only had Cacti stats for OS level things (CPU, I/O, etc.) but we were missing actual Hadoop statistics.

So we've now set up Ganglia at GBIF and the best news is it's public and using the very latest Ganglia 3.3 which was released only a few days ago in February 2012. The setup was relatively painless. Ganglia was just nice to work with. To get monitoring of HBase working we had to apply HBASE-4854 because it's not included in our Hadoop distribution (CDH3u2). Thanks to Lars George for the hint.

So we can happily report that Ganglia 3.3 works perfect with the GangliaContext31 from Hadoop.

Now all that's left is learning what most of these stats mean and then trying to extract useful information from them. Any hints are more than welcome and the HBase community already offered to help. Thank you very much!

For anyone interested in a few notes and details about how we set Ganglia up keep on reading.

BioCASe now producing DarwinCore Archives

Guest post from Jörg Holetschek, Botanic Garden and Botanical Museum Berlin-Dahlem.

The traditional way of sharing occurrence data with GBIF has been web-service-based for years. Data publishers have used one of the existing provider software packages (DiGIR, BioCASe or TAPIR Link) to expose their data as a DiGIR-, BioCASe- or TAPIR-compliant web service. Biodiversity networks such as GBIF used harvesters to crawl and index the records published by these services, an approach that works fine for small and medium-sized datasets, but runs into difficulties when record numbers hit the millions: Harvesting can take days and puts a heavy load on both the publisher and the crawler.

To overcome this, GBIF recently introduced DarwinCore Archives for storing all information of a dataset to be published in a single file. GBIF directly ingesting this file eliminates the time-consuming back-and-forth communication between data provider and harvester, speeding up the process and reducing load for both sides. GBIF’s IPT allows easy creation of such DarwinCore Archives and is a good option for providers that have already used the DarwinCore standard in the past or that want to share rather slim observation data.

However, sixty-two of GBIF’s data publishers are currently using BioCASe. In contrast to DarwinCore, BioCASe and its associated data standard ABCD are targeted mainly at rich data originating from specimens of natural history collections (even though it can be used for any type of occurrence data, including observations). Many of the BioCASe data providers also share their data with special interest networks such as GeoCASe, the DNA-Bank Network, or the EDIT Specimen network, all of them relying on BioCASe web services. Switching to the IPT and the associated DarwinCore standard is not an option for them.

For this reason, we decided to extend the BioCASe Provider Software with a feature to create DarwinCore Archives. This allows providers to continue using the rich ABCD schema (or one of its extensions) for the specific networks they’re connected to while using DarwinCore Archives to share their data with GBIF. In order to combine the richness of ABCD with the efficiency of downloadable archives, we created a hybrid in-between, the so called ABCD Archives, which can be used instead of the BioCASe web service for harvesting purposes (see figure below).

The first step – creating the ABCD archive – is implemented natively in the Provider Software in Python. The second step – transforming the ABCD archive into one or several DarwinCore archives – is done by Pentaho Data Transformation, an open source library also known as Kettle. In the current version 3.0, the transformation step is a stand-alone command-line application that can be downloaded separately; ultimately, it will be bundled with the Provider Software and integrated into the user interface.

The latest version of the Provider Software and the DarwinCore Creator can be downloaded from the BioCASe website. A detailed documentation of the new archiving features can be found in the PyWrapper Wiki. The wiki also stores a sample ABCD archive and a sample DarwinCore archive created by BioCASe.

Updating a customized IPT

This post originally appeared on the Canadensys blog and is a follow-up of the post Customizing the IPT.

As mentioned at the very end of my post about customizing the IPT, I face a problem when I want to install a new version of the GBIF Integrated Publishing Toolkit: installing it will overwrite all my customized files! Luckily Tim Robertson gave me a hint on how to solve this: a shell script to reapply my customization.

Here's how it works (for Mac and Linux systems only):

Comparing the customized files with the default files

First of all, I need to compare my customized files with the files from the new IPT. They might have changed to include new functionalities or fix bugs. So, I installed the newest version of IPT on my localhost, opened the default files and compared them with my files. Although there are tools to compare files, I mostly did this manually. The biggest change in version 2.0.3 was the addition of localization, for which I'm using a different UI, so I had to tweak some things here and there. It took me about 3 hours until I was satisfied with the new customized IPT version on my localhost.

I also subscribed to the RSS of the IPT Google Code website, to be notified of any changes in the code of "my" files, but I was just using this as a heads-up for coming changes. It is more efficient to change everything at once, when a stable version of IPT is out.

• RSS subscription for any changes in /webapp/WEB-INF/pages/inc, which contains most of my customized files
• RSS subscription for any changes in /webapp/styles/main.css, where I'm commenting out a lot of stuff so my CSS can kick in.

Setting up a file structure

This is how we've organized the files on our server. I've created a folder called ipt-customization, which contains all my customized files. That way, they can never be overwritten by a new IPT installation, which gets deployed in webapps. The folder also contains a script to apply the customization and a folder to backup the default files currently used by IPT.

• ipt-data
• webapps
  
  • ipt
• ipt-customization
  
  • backup-default
  • apply-customization.sh
  • revert-customization.sh
  • header.ftl
  • header_setup.ftl
  • menu.ftl
  • footer.ftl
  • main.css
  • custom.js

Creating the shell script

The apply-customization.sh script works in two steps:

1. Backup the default files, by copying them from IPT to the folder backup-default. The script will ask if I want to overwrite any previously backed up files. The last part is important if I'm running the script several times. In that case I do not want to overwrite the backups with the already customized files.
2. Overwrite the files currently used by IPT with the customized files, by copying them from my ipt-customization folder to the correct folder in IPT

# backup files of new IPT installation
cp -i ../webapps/ipt/WEB-INF/pages/inc/footer.ftl ../ipt-customization/backup-default/
cp -i ../webapps/ipt/WEB-INF/pages/inc/header_setup.ftl ../ipt-customization/backup-default/
cp -i ../webapps/ipt/WEB-INF/pages/inc/header.ftl ../ipt-customization/backup-default/
cp -i ../webapps/ipt/WEB-INF/pages/inc/menu.ftl ../ipt-customization/backup-default/
cp -i ../webapps/ipt/styles/main.css ../ipt-customization/backup-default/

# apply customization
cp footer.ftl ../webapps/ipt/WEB-INF/pages/inc/
cp header_setup.ftl ../webapps/ipt/WEB-INF/pages/inc/
cp header.ftl ../webapps/ipt/WEB-INF/pages/inc/
cp menu.ftl ../webapps/ipt/WEB-INF/pages/inc/
cp main.css ../webapps/ipt/styles/
cp custom.js ../webapps/ipt/js/

I also created a script revert-customization.sh, to revert the customization to the default IPT, in case something is broken. It moves the backed up files back to IPT:

# revert customization
cp backup-default/footer.ftl ../webapps/ipt/WEB-INF/pages/inc/
cp backup-default/header_setup.ftl ../webapps/ipt/WEB-INF/pages/inc/
cp backup-default/header.ftl ../webapps/ipt/WEB-INF/pages/inc/
cp backup-default/menu.ftl ../webapps/ipt/WEB-INF/pages/inc/
cp backup-default/main.css ../webapps/ipt/styles/
rm ../webapps/ipt/js/custom.js

Running the script

From the command line, I login to my server, navigate to the folder ipt-customization and make my script executable:

chmod +x apply-customization.sh

I only have to do this the first time I want to use my script. From then on I can use:

sh ./apply-customization.sh

To execute the script and customize my new version of IPT!

Bug fixing in the GBIF Data Portal

Despite our current efforts to develop a new Portal v2, our current data portal at data.gbif.org has not been left unattended. Bug fixes are being done periodically from feedback sent to us from our user community. In order to keep our community informed, this post will summarize the most important fixes and enhancements done in the past months:

• The data portal's main page now shows the total number of occurrence records with coordinates, along with the total count of records (non-georeferenced and georeferenced).

• Decimal coordinate searches were not working properly. When a user wanted to refine their coordinate searches to use decimals, the data portal was returning an erroneous count of occurrence records. Issue was fixed. Details here.

• Any feedback e-mail message sent from an occurrence or a taxon page now includes the original sender's email address in the CC field. Previously the sender's email address was not included in the feedback email, which represented a problem when the receiver replied to the email, but the sender never knew about the reply.

• The Taxon Web Service's GET operation was returning errors when trying to request some specific taxons. The problem was detected and fixed.

• On an occurrence detail page, when retrieving the original record from a data publisher, and the source data came from a Darwin Core Archive, it was not possible to retrieve a single record due to the single-file nature of a DwC Archive. (As opposed to a DiGIR request, in which you could extract just a single record). A fix was introduced so that the user can decide if he/she wants to download the complete archive (see an example).

• When using the data portal's Web Services to produce KML output, there were some problems when the generated KML contained HTML elements and Google Earth tried to open the file for visualization (This is a standard problem of XML). A small fix was introduced to escape the conflicting HTML inside the KML output.

• Other small GUI enhancements where also done.

Updates to the data portal's codebase is now done seldomly, but our goal is to fix any major issues that our user community reports. If you ever encounter problems, please don't hesitate to contact us at portal@gbif.org.