pero on anything

You should not try to use Hadoop as a “drop-in” replacement of your current (R)DBMS. That said it is still possible to utilize the power of cluster computing while circumventing its weaknesses when it comes to ad-hoc or real-time queries. We use Hadoop as an on-line system tightly integrated with our application and use it for both, long-running analytical queries and ad-hoc style queries.

In the mindset of a “traditional” database engineer one of the biggest concerns about Hadoop, or MapReduce in conjunction with a distributed file system in general, is the lack of indexes. Set aside that the debate “(R)DBMS vs MapReduce” is most of the time superfluous and sometimes almost leads to religious debates, the absence of a thing like an index is one the biggest hurdles you face when migrating data from a traditional DBMS.
Even though you will love the ability to view your data in any way you want without caring about its structure, at some point you feel that it is not right to always scan you 45TB of log files. (Even though it is soooo easy…).

Brute force is easy. Brute force is bad.

When we began migrating all those TBs of log-style data from our huge MySQL installations to Hadoop we did a lot of testing. We tested everything from Hadoop and MapReduce settings to different MapReduce abstractions like PIG, Cascading, Hive and others. There was this huge mass of data grinning at us and waited to be analysed in multiple ways, from “online” real-time access to “offline” decision making analysis. Due to our multiple views on the same data we came to this conclusion quite quickly: “Brute force is easy. Brute force is bad.” Yes, we can optimize our Hadoop installations and we can choose the really best query mechanism (actually we ended up writing our own), but it will not make things noticeably faster if you continue scanning all of our data all of the time.

Partitions are (sometimes) the better indexes

So, why are you using indexes (in the context of data retrieval)? I know why we did and do. It is all about primary key lookup and data clustering. Say you have the following table (MySQL):

Just a simple order log with an unique identifier (id) and a single associated product and customer. Since we want to view our data from different perspectives we added two additional indexes on product and customer. (In this example we need two indexes because MySQL can only use the leftmost prefix of an index.)
Dumping the whole table as a single CSV-file into your Hadoop cluster would mean that you always have to use what (R)DBMS call a “full table scan”. It would be pretty much the same like removing all indexes from your MySQL-table. Try to search for all products a customer ordered without the index idx_product_customer. (In fact Hadoop would perform this full table scan an order of magnitude faster.) But it would be ridiculous to remove all indexes from your table. But that is actually what you did when you exported the whole table into a flat-file!
What you should do, and what we did with great success, is to split up your flat-file CSV and arrange the data so that you can decide beforehand which part of the data needs to be accessed. So let’s split up the data and simulate all of the indexes (besides the primary key, more on that later on). A file-system-layout could look like this:

So when searching all orders customer_1 placed, we just use this file-pattern orders/*/customer_1.csv. Remember: HDFS and MapReduce’s inputs (like FileInputFormat) support globbing.

Now we actually simulated indexes by partitioning the data!

From here on you can go into more detail depending on your data structure. As an example you could add the date- and id-range to the file name like this:

This comes handy if you keep adding data to your cluster.
To make thinks even easier you could write your own InputFormat that encapsulates the building of the paths that match your query.

The small file problem

Since Hadoop has been designed to work on quite huge blocks of data it is not efficient when using a lot of small files. To prevent the creation of millions of very small files take a closer look at your data. Say your average customer places 50 orders. It would be a waste of resources to store multiple files for a single customer, each filling only a few KBs. A possible solution: group customers together.

You have to find the right balance between file-size and access pattern.

Some final words

To make it clear: Even though we have found a way to partition our data we have not gained the same flexibility as we have in any descent (R)DBMS (with enough of disk space, processing power and - most of all - RAM!). Querying for all orders made by a single customer may still take 0.01s in a (R)DBMS vs. 10s (or more) in Hadoop.

Never try to simply replace your (R)DBMS with Hadoop! Eventually you will end up writing a blog post saying that MapReduce and Hadoop are hopelessly worse than your favourite (R)DBMS. Hadoop is not a database!

Real-time lookups

You can still accomplish real-time lookup performance using Hadoop. One thing you could do is to take a look at HBase, a Google BigTable implementation.
Some times it is enough to use MapFiles which are simply a huge disk-based Hashtable.
In our application we implemented primary key and secondary keys directly on CSV files using MapFiles and distribute the lookup and local in-memory-caches over several machines. To speed things up even more we use a memcached cluster. (Eventually we will release all of this along with our MapReduce-abstraction as open-source once we feel it is mature and stable enough.)
One way or the other: Data redundancy will most likely become your best friend in these situations.

Regardless the techniques you are actually using you still have to think about your data in another way. You always have to when moving from a traditional (R)DBMS to any other kind of data storage and retrieval system!

Adding a lot of unit tests for our application that uses Hadoop and its Map-Reduce-Engine significantly increased integration build time. Hadoop comes with a LocalJobRunner which is used by default so you do not have to set up a complete cluster in order to run some Unit-Tests. This is great! But the problem is: it still produces a lot of overhead. Ramp-up and tear-down of a job might still take up to a few seconds. Having some hundreds of Map-Reduce-Jobs in your unit-test-base will definitely drag you away from the ideal “10 minute integration build feedback” you are always striving to get.

I cannot provide a complete solution to this “problem” (hey, it is still great to be able to run Map-Reduce-Jobs locally!), but the following configuration parameters cut the execution times of our tests in halves:

These settings are only feasible for small jobs with little input, of course.

I’m always glad to hear of better solutions to decrease the overhead even more!

Yesterday we faced a strange problem. A newly set up Hadoop cluster got unstable after a few minutes. Logs reported a lot of exceptions like:

java.io.IOException: Too many open files
at sun.nio.ch.EPollArrayWrapper.epollCreate(Native Method)
at sun.nio.ch.EPollArrayWrapper.(EPollArrayWrapper.java:68)
at sun.nio.ch.EPollSelectorImpl.(EPollSelectorImpl.java:52)
at sun.nio.ch.EPollSelectorProvider.openSelector(EPollSelectorProvider.java:18)
at sun.nio.ch.Util.getTemporarySelector(Util.java:123)
at sun.nio.ch.SocketAdaptor.connect(SocketAdaptor.java:92)
at org.apache.hadoop.hdfs.server.datanode.DataXceiver.writeBlock(DataXceiver.java:281)
at org.apache.hadoop.hdfs.server.datanode.DataXceiver.run(DataXceiver.java:102)
at java.lang.Thread.run(Thread.java:619)


or

DataXceiver
java.io.EOFException
at java.io.DataInputStream.readShort(DataInputStream.java:298)
at org.apache.hadoop.hdfs.server.datanode.DataXceiver.run(DataXceiver.java:78)
at java.lang.Thread.run(Thread.java:619)


and others. We double-checked ulimit -n and it reported 32768 on all datanodes as expected. lsof -u hadoop | wc -l was as low as 2000, so “Too many open files”-exceptions seemed strange.

A day and several installation routines later we figured out that the available epoll resources were not sufficient any more. Java JDK 1.6 uses epoll to implement non-blocking-IO. With kernel 2.6.27 resource limits have been introduced and the default on openSuSE is 128 - way too low.

Increasing the limit with echo 1024 > /proc/sys/fs/epoll/max_user_instances fixed the cluster immediately. To make this setting boot safe add the following line to /etc/sysctl.conf:

fs.epoll.max_user_instances = 1024


A lot of people keep asking me about the status of HSCALE so I thought it is best to just write about it.
Since I am waiting for the next GPL release of MySQL Proxy I concentrate on other things, both project-related and not project-related.

Continuous integration and test strategy enhancements

First of all there is a CI server up for almost 3 months now. Check out teamcity.hscale.org (Login as guest user). I also introduced a lint-process to discover bugs in my LUA code which arise mostly by mistyping variable names. By the way: You should definitely have a lint-like process in place if you are doing LUA programming or other languages of that type. It helps a lot.

What’s in svn trunk?

Multiple backends

The current status is this: Spreading across multiple backends is fully implemented and there are a lot of tests for that. Check out the latest code at http://svn.hscale.org/trunk. The tests running against multiple backends are “cheating” a little bit: Since the proxy still does not allow for ad-hoc allocation of backend connections I just create a lot of “dummy” connections to the proxy and use the approach of connection pooling the way the proxy does it. See an example. Note: This will not work in production!

Dictionary partition lookup and auto-partitioning

Besides the modulus partition lookup, which is only used in tests, there is now the dictionary partition lookup which allows for explicit partition definition. Detailed documentation can be found in the project wiki. It works as described here.
Another feature implemented is auto-partitioning. Depending on the partitioning function used it is possible to create new partitions on the fly. This way you can automatically spread the load across your MySQL servers. Another benefit is that you have a fine grained partition set-up right from the start which makes re-partitioning a lot easier afterwards.
Please take a look at the code and especially the tests to see how it works.

Parallel setup of HSCALE servers

In order to eliminate the SPOF (single point of failure) and possible bottleneck that directing all traffic through a single proxy would imply, HSCALE is designed to run in parallel. This means you can set up multiple proxies running HSCALE in parallel. This way you can easily implement fail-over scenarios using heartbeat or your favourite high availability solution and spread the load.
Why should running HSCALE in parallel be a problem in the first place? The easy answer: Because partition information is cached within each instance. As soon as the partition information changes your HSCALE instances would run out of sync using different partition information which would be disastrous to your data integrity. To avoid this HSCALE (more precise the dictionary partition lookup) works in two different modes: “NORMAL” and “FORCE”. If the mode is set to “NORMAL” then partition information is cached internally and only refreshed in a configurable time interval (configuration parameter “reloadInterval”). Whenever a change is made to the partition set up the mode is changed to “FORCE” which forces all HSCALE instances to re-fetch the partition information prior executing any query. After a configurable amount of time the system switches back to “NORMAL”. This is only the big picture. The implementation itself is a bit more complicated.
This approach is simple and robust because no other components are involved (like message queues) and it is guaranteed by design that no HSCALE instance is able to run with a wrong partition mapping. Changes made to the partition information involve a little overhead since all HSCALE instances will reload the partition mapping quite often. But taken into account that the partition information does not change that often (100 times a day would be huge!) it is an affordable price to pay for data integrity.

What’s next

Currently HSCALE is almost feature complete. The thing that’s missing - and making HSCALE production ready - is a MySQL Proxy version with different backend handling (Jan, please do not hate me!). Currently we (our company) do not have the resources digging into MySQL Proxy ourself and HSCALE is currently(!) not top-priority. So we will just wait and see.

Part of the problems we intended to solve with HSCALE are now moved to a Hadoop cluster since we have huge masses of log-style, read-only data which has to be analysed in multiple dimensions. HSCALE will gain focus right after that or as soon as a “suitable” GPL version of MySQL Proxy comes out.

If you are in need for a production ready, proxy-based sharding solution, please take a look at Spock Proxy. They use a different approach - they actually forked MySQL Proxy and implemented everything into it, thus no LUA is used - but the idea behind it is basically the same. They also offer some features HSCALE will not offer in the near future like handling of auto_increment columns across partitions. Some features are not there mostly because of the different design approach like arbitrary partitioning functions (they only offer range-based partitioning which is ok for many scenarios) or query hinting.

Speaking of Hadoop cluster - an idea that is ringing in my head for a while is to implement a MySQL Proxy LUA script that enables running (basic) queries against a cluster. It would be a little fun project.

First of all, I know this is a known problem, but it struck me so hard, I just had to write about it!

As Peter Zaitev points out calling SHOW STATUS might have a huge performance impact.

We recently replaced one of our servers with a DELL R900 with 96GB RAM. Having a disk-bound workload and > 1TB worth of data in InnoDB we expected a noticeable performance gain compared to the former server with 32GB. The new server even has better RAID and HDD.

But that was not the case. Things got even worse! A lot of queries “hang”, server load peaked at almost 7 and we saw a lot of cpu activity. Just before I started a deep analysis of what is going on inside I spotted a SHOW GLOBAL STATUS which ran every second. DOH!

Where did it came from? An administrator was running MySQL Administrator and used the Health chart to monitor some variables. So it periodically sent SHOW GLOBAL STATUS to the server. That resulted in a lot of queries waiting for the buffer pool (look at Peter’s post and the comments to understand why). And things get worse with bigger InnoDB buffer pool (this particular mysql instance uses 70GB!).

MySQL Administrator (the tool, not the person ) shut down - everything just looks great now!

While evaluating LuaSQL as backend connection replacement I came across this. I did a quick performance test using mysqlslap and it showed that just reading and copying the result can be significantly faster with LuaSQL.

Benchmark details

What I did was just sending the query to the backend and building up a new result-set in LUA.

This is the code for LuaSQL:

And this is the code for using MySQL Proxy only:

As you can see we do nothing but copy the result-set in LUA. This mimics the result-set aggregation HSCALE does if a full partition scan is necessary.

Results

Using LuaSQL:

Using MySQL Proxy only:

For comparison: Using empty read_query and read_query_result functions:

Versions used: MySQL Proxy 0.7.0 (svn-rev 511), LuaSQL 2.1.1, MySQL server 5.0.51a, mysqlslap 5.1.26rc

Of course I repeated the tests several times to verify the results.

Without digging too deep into the source of both MySQL Proxy and LuaSQL the biggest difference is that LuaSQL pushes the result-set row-by-row onto the LUA-stack whereas MySQL Proxy puts the whole result.

Update: As Jan points out below this is not true. MySQL Proxy puts the result row by row onto th LUA stack, too.

Update #2: The tests above ran on my Notebook (MacBookPro 2.4GHz, 4GB RAM running Ubuntu 8.04 64 bit). They are reproducible. Running the same tests on an 8-core-server putting the MySQL database on another server results in the MySQL Proxy version running slightly faster (about 2-5%) than the LuaSQL version.

Conclusions

Even though this tiny benchmark showed that the speed of LuaSQL seems to be feasible, there are still drawbacks. First of all: Depending on your workload only a fraction of your queries need result-set altering. Namely it’s only full partition scans that need this. Most of the time you just need to change the table name or the backend. And then LuaSQL is 100% slower than MySQL Proxy alone.

Another downside of LuaSQL is that it does not return the mysql field types but only the LUA types. This makes it impossible to build up a correct result-set that can be sent back to the client.

So still we need suitable (for HSCALE) backend connection handling in MySQL Proxy if we want higher performance.

Built-in result-set merging would be a big win, too. Then we could even have streaming combined result-sets taking the memory pressure from the proxy (since every result-set has to be fully loaded into memory).

That said I think about using LuaSQL for configuration handling since it is a lot easier than doing it via proxy.queries:append -> read_query_result -> proxy.queries:append -> ....

After working on build and test improvements (for example incorporating lualint and LuaCov) as well as other lua “side-projects” (i.e. Log4LUA) we are running towards HSCALE 0.3.

The focus of the forthcoming version 0.3 of HSCALE is Dictionary Based Partition Lookup. Using this partition lookup module lets you take full control over how your partitions are created and where they are actually located.

Update: Dictionary Based Partition Lookup is fully implemented. See this blog post and the wiki page about it.

Please note: Due to the problems with backend connection handling version 0.3 will still focus on single server backends. Even though support for multiple backends is already implemented in HSCALE. Please look at the proxyUnit tests for a glimpse on how multi-server backends will be handled.

Please check out the current development snapshot at svn.hscale.org to see what is already there. Currently the partition lookup is fully functional but the administrative commands and some further hashing functions are missing (see below).

How does dictionary partition lookup work?

As the name implies partitions are looked up in a dictionary. The dictionary itself is stored in the main database and cached internally. So now you can freely move partitions around and create new ones.

What is done internally is this:

1. Apply a hashing function (read further) to the value.
2. Lookup the partition based on the hashed value.
3. If no partition has been found use the default partition created for every partitioned table.
4. Return the partition (and assigned backend)

Hashing functions

To reduce the number of partitions to be created and the overall administration overhead, a hashing function may be applied to the partition value before the partition is looked up.
Currently there are 3 hashing functions available:

• MOD(X): A modulus function grouping X partition values together. Works only for numbers of course. If you have *lots* of different partition values with a smaller number of rows each, then it might be better to group them together instead of creating *lots* of partitions.
  Example: Using MOD(3) the values 1, 4 and 7 will end up in the same partition.
• PREFIX(length) Partition values are grouped together by the first length characters. Works on everything (everything is treated as string).
  Example: Using PREFIX(3) the values “foo“, “foobar” and “footaliciuos” will end up in the same partition.
• NONE(): This function does … nothing! Use it if you really want a 1:1 relationship between partition values and partitions.

Further hashing functions are planned (and might make it to version 0.3):

• DIV(X): As opposite to MOD(X) this function divides the partition value by X. So this like a fixed range function. While MOD(X) creates at most X partitions DIV(X) creates infinite number of partitions.
• DATE(pattern): Enables date-range based partitions.
  Example: Using DATE("%Y-%m") will group by (year-)month.

Administrative commands

Because handling of partitions is a delicate thing, creation and maintenance of partitions should not be left to some obscure SQL-statements. Therefor the dictionary partition lookup will provide administrative commands that try to avoid mis-configuration. It will be checked whether partitions overlap etc.

Because this is still work in progress the commands might change.

Table setup

First of all your table has to be set up:

The table has to exist before and will be renamed to the default table name ('users_default' in our example). SETUP_TABLE can only be called once per table.

Add partitions

The example above creates a partition with the name 'nick1' for table 'users' and partition value 'mar' (users with nickname 'marvel', 'martin' etc.). The partition name will directly reflect to the table name of the partition. So the table for this partition will be users_nick1. Usually you would use the partition value ('mar') as partition name but you don’t have to. You are able store multiple partitions in the same table. This sounds strange but it makes sense to define a finer partitioning scheme upfront and actually use a wider scheme until you really need to split up data. This makes it a lot easier to split things up afterwards.

Moving partition data

In version 0.3 partition data will not be moved to a newly created partition. You will have to do it by hand. This will be implemented as soon as multiple backends are fully supported because that implies a different approach since data has to be moved between different servers then. An administrative command to move partitions will also be available then (HSCALE MOVE_PARTITION(...)).

Multiple instances of HSCALE working on the same data

HSCALE is designed to support multiple instances of it running in parallel mostly to avoid to be the bottleneck and single point of failure. Every instance of HSCALE periodically refreshes the internal partition configuration to reflect changes made by another instance. In the case of creating and moving partitions, partitions will be locked inside HSCALE so all clients will wait until the operation finishes. This guarantees data integrity.

Version 0.3 will be released within the next few weeks but definitely after MySQL Proxy 0.7 has been released so we can thoroughly test it against the newest version.

Please feel free to discuss certain features and design decisions shown above. Any feedback is welcome!

Everything is available at the project page

Please report issues and feature request here.

Version 0.2 is a bug fix release but with 2 significant changes in syntax:

1. Changed all constructor methods from create(...) to new. Seems to be more common in the LUA world.
2. The logger class is now returned by the module. So it is
   local logger = require("optivo.common.log4lua.logger")
   local LOG = logger.new(…)

   instead of

   local logger = require("optivo.common.log4lua.logger")
   local LOG = logger.Logger.create(…)

Some potential bugs have been spotted using (a slightly adopted version of) lualint and LuaCov.

With release of version 0.2 the Log4LUA API is declared as stable. Future versions might change the syntax but are guaranteed to be backwards compatible.

UPDATE Log4LUA 0.2 released. Go to the project page.

After the dust about backend connection handling in MySQL Proxy had settled, I begun working on other parts of HSCALE again. (I’ll return to the backend handling later after talking to Jan Kneschke about their plans.)

One of the issues that bothered me the most was logging. Until now I just had used a simple debug function that could be enabled or disabled using an environment variable. The pretty straightforward approach a lot of people use. But there is more to logging than to print debug messages. So I searched the web and of course I found LuaLogging. It is more flexible, has different levels (DEBUG, INFO, WARN, ERROR, FATAL) and appenders. You can print your log messages to the screen, write them to a file or send an email.

After working with other logging frameworks in other languages (mostly Log4X like Log4J) I got used to a number of features missing in LuaLogging:

• There are no means of external configuration. The logger is configured inside the code. You cannot use different configurations for development and production out of the box. This is crucial in my opinion, since during development it is handy to enable all log levels and print everything on screen, while in production you want to log to a file and disable at least the DEBUG level.
• You get no information where the log event came from. Log messages are more meaningful if you know where the log message has been created, i.e. in which source file and line position. This makes it easier to track down bugs and errors.
• There is no log category concept. Having different log categories makes it easier to group log messages either by source or by meaning. As an example you could have one log category called “core” where all core messages go to and another “connection” where all connection related log messages go to.
• You cannot use multiple appenders. In production you might want to log all messages to a file and get an email for all errors.

Since adding all this functionality to LuaLogging would in fact mean to re-write it, I wrote yet another logging facility and called it: Log4LUA. I choose the name because this package behaves almost like Log4X.

Download Log4LUA

A detailed documentation on how to use it can be found here (luadoc).

Features

• External configuration. Configure your logging system via a config file.
• Logger categories Configure different categories for different logging tasks.
• Detailed information available: source file, line, function or the whole stack trace
• Console, file and smtp (email) appenders
• Multiple appenders per category Log to a file and get the worst errors by email
• Level threshold for smtp appender. Don’t get every log message by mail only the ones above a defined level.
• Log file rotations for file appender. Based on a date pattern

Basic usage

First write a configuration file (say log4lua.conf.lua):

In your lua source (say mycode.lua) use the following to get a logger:

Start your program with:

For more examples look at the HSCALE sources. Log4LUA is used everywhere.

Please feel free to comment on this. Or send usage or bug reports. I will setup a project page later on.

Recently I added the first code to support multiple MySQL backends in HSCALE (see svn.hscale.org). As a “side note” I have to thank Giuseppe Maxia for MySQL Sandbox which made multi server testing a bliss!

While coding this I started to feel that writing HSCALE on top of MySQL Proxy is no more as easy and clean as it started out to be. And finally I reached the point where I have to say:

HSCALE (and maybe other advanced multi-server applications) and (current) MySQL Proxy don’t fit very well.

Before I go into details please let me make clear that this is mostly due to the specific nature of MySQL Proxy being a connection and protocol based proxy. MySQL Proxy is great and there are a lot of cool applications that fit perfectly with it.

Aside from some other minor glitches (like missing tokens in SQL tokenizer) the biggest show stopper is:

Handling of multiple backends

The biggest problem I had to face is the handling of backend connections in MySQL Proxy. What I would need for HSCALE are dedicated connections to every backend for every proxy connection. So if a client connects to the proxy, a connection to each backend is opened and attached to this particular client connection. This way I easily could maintain the state of the connection by sending commands like SET variable or USE database to all backends.
Dedicated connections are vital for XA or transactions in general, too.

The way the proxy handles backend connections right now is somewhat cumbersome (See an example.). The way you have to maintain your own connection pool is hard to understand and uses too much “magic” in my opinion. And the pool only grows with the number of connections made to the proxy. People are having problems with this approach (read the MySQL Proxy forum, for instance this thread).

The current experimental multi-server code in HSCALE uses this connection pooling technique since it is the only way to establish connections to other backend servers.

So what can we do now to implement multi-server support in HSCALE?

1. Wait until MySQL Proxy supports dedicated backend connections This would be the easiest and most elegant solution. But it could be that Jan Kneschke and the other developers decide that this is not what they intended to do with the proxy. And this would be totally ok from their point of view! Even if they decide that this is a cool feature it could take a long while until it is implemented. (Hint: I would gladly help out here ).
2. Work around this using luasql While possible this solution would not be preferable since we are using two technologies for the same thing.
3. Switch to another proxy There are quite a few. This would be no fun and I did not take a look at them all to see if it is even possible. It would be complete rewrite though.
4. Fork MySQL Proxy or implement a plugin The new plugin technology could be used to implement the multi-backend connection stuff myself. As a last resort, I could fork the whole project (like Spock Proxy did, see below). This would still mean a lot of work though and loss of the MySQL efforts put into the proxy.
5. Use other sharding technologies and eventually abandon HSCALE As an example Spock Proxy, a fork of MySQL Proxy does a great deal of what we intend to do with HSCALE.
6. Re-implement HSCALE as a JDBC driver Since our applications are written in Java exclusively we could do that.

Of course I would love to continue HSCALE as a MySQL Proxy application! The main reasons are the efforts MySQL is putting into MySQL Proxy, the extensibility we gain combining multiple LUA scripts (there are more things we do with the proxy apart from HSCALE) and last but not least the community echo HSCALE already received.

As a next step I will take a deeper look into the proxy code and evaluate the efforts needed to add dedicated connections.