Apr 28, 2016

Faster Lucene numeric field retrieval

I wrote previously about using a custom scoring loop to retrieve Lucene document values. It turns out there is another simple optimization that allows much faster field value retrieval. That feature is known as DocValues field type. The idea is that in addition to LongField/DoubleField types that can be indexed (and so filtered on when running a Lucene query) there is another numeric field type. 

The difference is that the DocValues fields cannot be indexed. So the pattern here is to have to Lucene document field subsets. The fields from the first one are indexed but not stored. They are used to match documents. The second subset consists of DocValues fields only. For a matching document the values of the DocFields can be retrieved from the index reader without fetching the entire document. The DocValues fields are implemented using memory-efficient techniques including reasonable compression.

In the example:

  • there are two numeric fields queries collected for each matching document by Processor 
  • Searcher can execute a custom scoring loop query
  • DocValuesQuery obtains entry points to DocValues storage and create a new Provider
  • Provider retrieves values of the required DocValues fields and notifies the Processor 
  • Processor represent the logic responsible for acting on the collected values for every matching Lucene document



Mar 21, 2016

AWS SQS for reactive services

Last year I left the cozy world of self-managed systems for the now predominant AWS platform. It's been a mixed blessing so far. On the positive side, devops are relieved of so many daily troubles of running infrastructure components such as message brokers. On the negative side, developers are deprived of many conveniences provided by any JMS/AMQP-style product such as topics and no need for polling. 

It's very common to use a message broker to implement asynchronous API for your backend services. Especially in a world where major RPC frameworks (I am looking at you, PBs/Avro/Thrift) have only partial support for fully asynchronous clients and servers. Not everyone is running AKKA or Finagle in production. All you need is message broadcasting (i.e. topics) and an instance identifier to ignore responses sent to someone else.

By contrast, SQS is designed for essentially one topology: a number of worker nodes processing a shared queue. There is a gap between SQS and Kinesis which I still find to be rather painful. But for services with one-way requests only SQS is indeed simple to use.

For people coming from traditional message broker world there are a few noteworthy differences:
  • you need to poll a queue to recieve messages, there is no broker to push messages to you
  • you can receive up to ten messages in one request
  • there is a visibility timeout associated with messages; messages received by one consumer but not deleted from the queue within the visibility timeout are returned to the queue (and become available to other consumers)
  • an absolute queue name and a queue URL are two different strings
  • you can list queues by queue name prefix
I have a handy example illustrating SQS message processing with AWS SDK. It implements a typical message processing sequence as shown below:

  • SqsQueuePoller is supposed to be called periodically to poll a queue
  • AsyncSqsClient is a half-sync/half-async-style wrapper around AWS SDK client
  • Handler represents a service capable of processing multiple requests concurrently
  • MessageRepository keeps track of the messages being processed
  • VisibilityTimeoutTracker makes sure the visibility timeout never expires for the messages being processed

Feb 19, 2016

Reading Parquet file

Now that we can write a Parquet file, it could be useful to be able to read it back. We still assume a simple record type as the data format. This example can be adjusted for more complicated data schemas with nested structures though in such a case you should probably consider using PB/Avro messages instead.

It is not a surprise that reading a file is sufficiently similar to writing it. One would need to:
  • extend the ReadSupport class to parse a file schema and create readers for individual fields
  • create a new ParquetReader instance making possible to read a file one row at a time
  • when called by Parquet library, copy each row field value to your current row buffer 

Your read support class is responsible for returning a concrete RecordMaterializer. This class represents the ability to read a single row. Internally, it holds a reference to a GroupConverter that knows which PrimitiveConverter instance to use for the row field with a given index. 

Individual PrimitiveConverters are responsible for writing a field value to the corresponding slot in the data structure representing individual file rows in your application. The row reading sequence is mostly self-explanatory, please have a look at the example source code to see it in action. 


Jan 12, 2016

Asynchronous RPC server with Thrift

A. Introduction


So many systems use Protocol Buffers to define APIs and DTOs that it’s easy to forget there is also Thrift. In contrast to the PBs, Thrift has always lacked proper documentation. It is enough to say that one of the most popular guides for it was written by an outsider. 

On the other hand, Thrift comes with an actual RPC library. The PBs world will get one only once GRPC is released. For those of us who do not base our backend services on something like Finagle, libthrift could be a reasonable choice at least initially. 

B. Documentation and examples


When one considers a new RPC framework one of the first questions to ask is how well its support for asynchronous communications is. Strangely enough, in case of Thrift the official documentation is pretty silent. As a matter of fact, I was not able to google an example of an asynchronous Thrift-based service or any guidelines for using Thrift-generated classes with “Async” in their name. 

What  I found was all about synchronous calls. Strangely enough, even one libthrift alternative I found  had nothing to say about asynchronous calls to my utter surprise. 

So I had to dig a little deeper to understand how to do it. While exploring it I created a small project that I will use as a complete example. The idea was to prototype a service with two different service interfaces. Each one takes one request object and returns a response object.

C. Thrift-generated code walk-through


If you look at the file generated by the Thrift compiler for your service you’ll see a few classes with the prefix “Async”:
  • AsyncIface: the asynchronous version of your service interface with additional AsyncMethodCallback argument
  • AsyncClient: client-side view of your RPC service interface; it requires some configuration to actually make calls 
  • AsyncProcessor: server-side intermediary between your actual request handler and the TServer you will use to wrap everything into a running process

D. Wiring up libthrift infrastructure


The generated classes need to be plugged into the libthrift infrastructure. On the server side, you will:
  • create an instance of your request handler
  • wrap it into an instance of Thrift-generated AsyncProcessor
  • in case of multiple service interfaces, create a multiplexed processor and register each async processor with it using a unique name
  • create a server configuration with typical parameters such as TCP port and a j.u.c executor to accept requests. There are a few TServer implementations to choose from
  • start the server




On the client side, you need to: 
  • create two reusable instances: a client manager and a protocol factory
  • with those two instantiate (for every service interface) a client factory 
  • make the factory create a client class instance
  • call the client

Each client takes a TCP host/port pair so in real life it would take a discovery service of some kind  to find those. Notice that AWS ELB supports TCP traffic load balancing and so there would be the only URL for any number of servers in that case.



E. The curious case of multiplexed asynchronous processor


When I mentioned a multiplexed processor I lied to you. It turns out there is no such thing currently in libthrift-0.9.3.jar . So if you use a multiplexed protocol factory on the client side there will be no peer on the server to demultiplex it. As an immediate workaround, I implemented one. As a real solution we have got a pending pull request for THRIFT-2427.

At first glance a glaring hole of this size in a mature library from a big company makes me think that Thrift is an obsolete dead-end. That would explain why they never bothered to produce reasonable documentation. I have some hope for GRPC in this respect.

F. Service client and request handler ideas


Aside from wiring up the auto-generated classes with the infrastructure from the libthrift library you will also need to implement an actual request processor and to call the client class somewhere. In my project you can find both pieces in the unit test. As a working end-to-end example it mostly follows this description even though it makes some effort to represent and configure RPC service definitions in a more generic way.

On the server side it is convenient to process request by 
  • supplying the corresponding job to some executor (different from the one used for networking)
  • call Thrift RPC callback from a CompletableFuture listener



In a similar way, on the client side a CompletableFuture allows to receive a response either synchronously with get or asynchronously in a CompletableFuture listener. Notice how an actual response instance is wrapped into a Thrift-generated class representing an RPC method call.


Sep 3, 2015

Customizing Lucene scoring loop

By default, one thinks about Lucene as a full-text search engine. You give it many text files and it will be able to find the most relevant ones for a query. The API is quite straightforward and there is no need to know anything about what happens underneath it.

There is another, arguably less orthodox, way to think about it. Lucene is widely known for its search speed. So instead of looking for the most relevant text documents one could use it as a more generic data storage with excellent indices. Some people used it this way for storing time series.

Once you start this kind of development you quickly recognize the need for calling Lucene efficiently. It's trivial to have millions of matching documents for your query. So if you want to extract data from all of them and not just a few top-scoring ones you will be wise to pay attention to new object allocation. It start from simple things such as caching and re-using Lucene document and filed instances. Then you stop calling search methods returning arrays and switch to callback-based ones.

And then there is the next stage when you find out that it's possible to go a level down and abuse Lucene a little. This approach is usually referred to as "custom scoring loop". Oddly enough, by far the best explanation of it I have seen is a blog post. I can't recommend enough that description, the author knows his stuff and goes quite deep into Lucene internals. 

The idea is to:

  • create a special kind of Query that can register a listener for matching documents
  • when the listener is called, instead of scoring a document in a different way, read the fields required for processing; if possible, process them right there
  • ignore the results returned by IndexSearcher

It's actually surprisingly simple to implement this idea. It's enough to extend three classes. The sequence diagram below shows a representative control flow:


  • extend StoredFieldVisitor to have a Visitor that knows the Lucene document fields you want to read
  • create a Processor that owns the Visitor and can be notified about availability of document field values
  • extend CustomScoreProvider to have a matching document listener; when called, apply the Visitor to read field values and make the Processor use them
  • extend CustomScoreQuery to register the listener with Lucene searcher
  • call IndexSearcher with an instance of the new query type; once finished, enjoy the data gathered by the Processor

Aug 11, 2015

Writing Parquet file

Parquet is a column-oriented binary file format very popular in big data analytics circles. Nowadays it's probably impossible to find a sql-on-hadoop engine that does not support this format. The Parquet library makes it trivial to write Avro and Protocol Buffers records to a file. Most of the time you can't beat the simplicity of a "df.write().parquet("some-file.par")"-like Spark one-liner.

But it's also possible to find yourself in a situation when you want to export data from an existing system which does not use Avro-like records. It's actually quite easy to do. One would need to:
  • extend the WriteSupport class responsible for writing a record instance to Parquet output
  • extend the ParquetWriter class to instantiate the new WriteSupport subclass
  • create a schema representing your record fields
  • for each record instance, write the field values to RecordConsumer

For simplicity sake let's assume a record type without nested structures. The high-level writing sequence looks roughy like that:

Here Record Writer represents a collection of Field Writers. Each Field Writer implementation knows how to write a field of a particular primitive type. The sequence is straightforward, please look at the example source code for details. Notice that to write a null value for a field it's enough to just skip the field for that record.

One unexpected complication is that for no particular reason Parquet library uses java.util.logging. This is the first time in my life I see anybody using it. You are not likely to have a logging configuration for it in your code base. You will definitely want to separate Parquet logs from the rest because they could be quite verbose. I actually had to use a rather unpleasant way to configure logging properly. 

Jul 2, 2015

Running SparkSQL in standalone mode for development experiments

Spark has very decent documentation in general. Cloudera and Databricks blogs cover the rest of the finer points concerning configuration optimizations. But it took me a few passes through the documentation to learn how to configure my Spark development cluster. It'll probably take you longer to copy and unpack Spark archive to your boxes than to configure all you need for experimenting in standalone mode. But to save you a few minutes, I created this gist which summarizes my experience. 

To install Spark, on each server:
  • unpack Spark archive
  • "cp conf/spark-env.sh.template conf/spark-env.sh"
  • "vi conf/spark-env.sh"

To run Spark:
  • on master, "./sbin/start-master.sh"
  • on worker(s), "./bin/spark-class org.apache.spark.deploy.worker.Worker spark://127.0.0.1:8091"

The gist shows how to configure a few most useful things:
  • Spark Executor memory, in case you want to quickly experiment with it without changing spark-env.sh on all the workers
  • recommended non-default serializer for more realistic performance
  • remote JMX access to Spark Executor (NOT secure)
  • data frame cache batch size

Jun 30, 2015

Insufficient support for sparse data in sparksql

I was experimenting with sparse data in sparksql and I learned at least one surprising thing. I'm really interested in OLAP-style analytics use cases. If you squint enough SparkSQL looks like an in-memory analytics database. So even though SparkSQL is not a data storage product its DataFrame cache can be abused for doing in-memory filtering and aggregation of multidimensional data. 

In my experiment I started from an 820,000 rows by 900 columns Parquet file that simulated a very sparse denormalized OLAP dataset. I had a dozen or so dense columns of the long type representing dimension member ids. The rest were very sparse columns of the double type pretending to be metric values. I was impressed to see that Parquet compressed it into a 22 MB binary file.

The next thing I tried was actually loading that file as a DataFrame and caching it in memory to accelerate future processing. The good news is that reading such a file is a one-liner in Spark. The bad news is that it took 3 GB of SparkSQL cache storage. Initially I though that I had misconfigured something. After all to me 3 GB very much looks like 800K * 900 * 8 bytes with a modicum of compression. After looking at my executor's debug-level logs I could see that indeed what happened. Double-type columns were "compressed" with the PassThrough compression scheme which is a no-op.

I dug just a little bit deeper to see that no compression scheme is actually supported for floating point types. Well, who knows, even assertEquals treats floating types differently. Maybe data science is all about integer-type feature vectors. But while looking at the NullableColumnBuilder I also noticed that it is biased towards dense data and is rather unhelpful if what you have is sparse. It encodes a null value by remembering its int (i.e. 4-byte) position. So if you have an empty long/double (i.e. 8-byte) column, the best you can expect is to spend half of the memory a fully populated column would take.

I actually couldn't make the suggested workaround work in Java but I was not really trying. So I am still convinced that currently SparkSQL is rather sparse data-unfriendly. The corresponding compression functionality is not pluggable. I guess one would need to implement a new NullableColumnBuilder trait, extend that trait by new DoubleColumnBuilder-style classes, and instantiate those classes in ColumnBuilder::apply(). I feel like trying it to see if I can make something simple work.

May 3, 2015

Data storage for SparkSQL-based analytics

I have been mulling over the question of Spark-friendly data storage recently. My favorite use-case here is still the one where:
  • in a multi-tenant environment 
  • you can provision enough memory for all/most of the data to reside in RAM
  • your target system is expected to have response time of a few seconds, preferably less
When I imagine an ideal data storage in this context I see things such as:
  • In a steady state, all data is in  memory, backed by some disk-based cold storage.
  • Datasets are split into multiple partitions which are held on different servers. It makes possible to execute a query in parallel on multiple servers. In case of a crash, only a smallish data subset is lost and has to be recovered from the cold storage.
  • A catalog service shared between the storage component and the Spark scheduler. It would allow the scheduler to honor data locality for expensive computations.
  • Obvious features such as efficient column compression and support for predicate push-down
  • It's too much to ask, but using the same storage (or at least transfer) format as the internal Spark cache storage would be nice for server recovery time. 
Curiously enough, I don't see storage options discussed much in Spark documentation and books. Probably because the default Spark positioning is against MapReduce and so one is expected to have ORC/Parquet files on HDFS.

My impression is that there are three reasonable alternatives. All of them are work in progress to such an extent that only your own experiments can show how far they are from production-quality. It is not entirely clear how stable they are but there are companies running them in production despite of the official experimental/beta status. So I can imagine some startups betting on those technologies with expectation of growing together.
  • Spark cache itself. On the positive side, it's easy to use, it's integrated into the engine and it's believed to be memory-efficient because of column-orientation and compression. The problem is that it's not a real storage (try to imagine updates), it's called a cache for a reason. What is much worse is that if you lose you driver then you lose every single cached DataFrame it owned. It's a really big deal and there are only tepid answers right now.
  • Tachyon. You would think that for a project originated at the same lab the integration (and PR around it) would be top-notch. Apparently there are missing pieces and Tachyon is not explicitly mentioned in what passes for a roadmap.
  • Cassandra. With an official connector on github, the story here seems to be more clear. For people who already run it I would expect it to be not a bad idea. You already have an in-memory columnar storage that can process predicates and serve column subsets. Co-locate Spark workers and Cassandra nodes and data transfer might be not that expensive.
  • GridGain Ignite. Frankly, I guess I am not being serious here. I have never looked at their in-memory file system anyway. But according to some not-so-fringe voices..
When I just started thinking about it, serving Parquet files via Tachyon as your SparkSQL data storage sounded like a buzzword-compliant joke. I am not so sure anymore even though it still looks strange to me. I would say that the Cassandra option looks much more traditional and so likely to be production-worthy sooner than anything else. But I admit to having little certainty as to which one is solid enough to be used this year.

Apr 19, 2015

SparkSQL as a foundation for analytics server

Imagine that you want to build an analytics server for the enterprise-scale data. In contrast to web-scale analytics with sql-on-hadoop, you probably have datasets which are of moderate size. Let's say high double-digit GBs. With 244 GB RAM instances available even on AWS (e.g. r3.8xlarge, or i2.8xlarge) you can fully expect to fit the largest dataset in memory. It's hardly a surprise that people have been talking about it for years [1]. Now a Java heap space of that size is arguably not a good idea because of garbage collection. But then again, with off-heap storage or rarely allocated primitive type arrays it could be manageable.

Query engine techniques we know and love

When you mention analytics I think of OLAP. When I think of OLAP I see columns and column-oriented storage. Column-oriented always rhymes with efficient compression in my book. After some initial Hadoop-inspired confusion even the sql-on-hadoop world is converging on essentially MPP relational query engine architecture. By now we got used to expecting from an analytics query engine typical features such as:
  • relational operators as the core computational model
  • optimizations biased towards in-memory processing
  • column oriented storage and data transfer; lots of compression [2]
  • once in memory, data is held in primitive type arrays; no Java auto-boxing ever
  • operators implemented to operate on batches of rows; hot loops compiled into byte code [3][4]
  • support for predicate pushdown and retrieval of a subset of columns only
  • query engine aware of dataset partitions and capable of locality-sensitive scheduling (aka "send computation to where the data is and not the other way around")

Why Spark?

So let's see how well Spark meets these expectations. By Spark here we definitely mean the new shining Spark SQL component [5] and its DataFrames API in particular. From our OLAP perspective we can think about Spark SQL as a relational MPP engine wrapped into the DateFrames API on the top and the Data Sources SPI on the bottom.

The DataFrames API accepts both SQL expressions and normal Scala method calls. All the usual relational algebra fun, nothing to see here. Except for what is known as data frame caching. In which case the data frame in question is put into in-memory column-oriented storage with appropriate compression applied.

Did you know Scala has quasiquotes? Neither did I. It turns out Scala macros can be useful in real life :) So the Spark optimizer will generate byte-code for your query. Something an average sql-on-hadoop engine can do only if it's called Impala :) And quasiquotes look so much cleaner than ASM.

The Data Source SPI is another important enabler. It's a means of plugging real data storage into Spark SQL engine. Among its few simple interfaces there is one which supports both predicate push down and retrieval of only a subset of all columns.

In general, it's safe to say that Spark is making real progress. It's already a very promising query engine and they are clearly paying attention to efficiency of in-memory execution. If they lack certain common capabilities now it is very likely the gap will be closed soon enough.

Why not?

The question I am the least comfortable with currently is data storage and the corresponding data transfer. When you have an essentially in-memory query engine it's only natural to expect an in-memory data storage closely integrated with it. 

Ideally, when you read from the storage you want to use the same binary format as the DataFrame cache. Whether you marshal data between the JVMs on the same node or run against out of heap storage you want to avoid/minimize any kind of transcoding.

For scalability and fault tolerance you want to split your dataset into multiple partitions. You want the query engine to be aware of those partitions so that it can send queries to the corresponding servers. Each partition could be replicated to reduce server crash recovery latency.

As of now, I don't quite understand the whole story about most promising Spark data storage. Clearly, Tachyon is on its way. I'm not entirely convinced that "a very fast HDFS" is how I imagine my data storage API though. GOOG will tell you that people also use Cassandra as a column storage with a good Spark connector.

It really feels like those are important considerations but not show-stoppers. Personally, I would expect more emphasis on data storage from the Spark team this year. It's just not clear to me right now how I would architect storage for a real analytics system on SparkSQL.

References

[1] SanssouciDB: An In-Memory Database for Processing Enterprise Workloads
[2] Integrating Compression and Execution in Column-Oriented Database Systems
[3] Vectorization vs. Compilation in Query Execution
[4] Hive Vectorized Query Execution Design
[5] Spark SQL: Relational Data Processing in Spark

Mar 11, 2015

Scala is hot in analytics companies

It is hard to believe it's been six years since I started worrying that Java is obsolete :) We lived long enough to see Java8 in production. A couple of years ago I thought that Scala flat-lined. I remember keynotes on never getting to mainstream and compiler folks leaving. But this year I can see a very different picture. Most of the companies dealing with analytics are writing Scala. A great many are either already running Spark in production or prototyping it. At this point it feels like a very good time to join such a company. There are too few people around with real Scala experience to make it a strict requirement.

It is interesting to observe that some arguments against Scala sound so familiar from good old C++ days. Some companies ban use of a certain feature subset (e.g. implicits). Some folks have trouble with a multi-paradigm language where there is so much choice. Few people are excited about function signatures in the standard library (remember STL? partially specialized template anyone?). I know good engineers are supposed to value simplicity. I must be a bad one, I am attracted to complexity. Scala reminds me C++ in this respect. There is always some language tidbit to pick up every time you read a book. In Java only the j.u.c package could fill one with so much joy for years :)

Mar 10, 2015

Research projects inch towards mainstream big data analytics

Just recently I wrote about a previously small research prototype that grew up into a much larger system. It took me some time to recognize that another university project from a few years ago got even closer to the world of real software. After rebranding it is known as Flink and resides at Apache Incubator.

It raises a few interesting questions about evolution of what could be liberally called sql-on-hadoop query engines. A growing number of open source systems used for analytics look architecturally very similar. If you squint enough, each and every of Presto/Impala/Hive/Drill/Spark/AsterixDB/Flink and their ilk is fundamentally an relational MPP database. Or at least its query engine half.

Some of them started with a poor computational model. To be more precise, the mapreduce revolution of Hadoop 1.0 allowed people access to the kind of computational infrastructure previously absent from the open source world. Without alternatives, people were forced to abuse a simple model invented for log batch processing. Others didn't limit themselves with two kinds of jobs and went for real relational operators. 

Impala was arguably the first wake up call from the database universe. Its technological stack alone was a clear indication of the kind of gun real database folks bring to a Hadoop knife fight. Nowadays it's Spark that is so bright and shiny that every other day people suggest it could replace MapReduce in Hadoop stack.

I reckon it'll take years to see how it all plays out. If Spark, with its academic pedigree, can challenge Hadoop/MR could it be the case that AsterixDB or Flink mature enough in a couple of years to do something similar? They also started at respected universities, roughly at the same time, had the right relational MPP core from day one, have some Scala in the code base. Are they too late to the game and will never be able to acquire enough traction? Do they need a company to provide support (and aggressive marketing) as a precondition? Will they be able to run on YARN/Mesos well enough to compete in a common Hadoop environment?

Feb 18, 2015

GridGain is now an Apache project

It's not exactly news but still quite a recent stage in its evolution. Years ago, say in 2009, it was quite a unique product. There were a few popular data grids/distributed caches to choose from. There was no alternative open source computational grid though. Moreover, GG was easy to use and well documented. Its code base was well structured as a consequence of a highly pluggable architecture. For a small team with roots in a peculiar location it was a very impressive product. 

I am not sure why GG is not more popular. It's not exactly obscure but it certainly feels like a niche system. From the core API it's clear that the original MapReduce had some influence. Taking into account how generic most of GG core is it probably became a limiting factor later. One really important piece missing from their architecture was an explicit scheduler. Without it GG could not be re-purposed as a really generic basis for something akin to "sql-on-hadoop".

Which also raises a question. Did they fail to pivot in that direction? They had a lot of experience with middleware. They were smart enough to appreciate Scala extremely early. Is it conceivable that they might have come up with something like Spark? Did they lack database internals expertise? Did not have enough resources to compete with Internet companies? There is very little reuse in this space. Netty and ANTLR are probably the only libraries shared by the Prestos/Hives/Drills of this world. So apparently nobody needs common middleware. The Impalas of this world try to skip even most of Hadoop.

I am curious about GG future. They went back and forth on their open source/community edition strategy and at at times it looked like desperation. I believe they initially released community edition simultaneously with enterprise one. Then separated them by almost a year. Their github repository was just a backup for development done somewhere else. And now it's all an Apache project. Could it be a new beginning? 

It would be interesting to know how people use GG in 2015 when the market is so over-saturated with "grid/cluster" frameworks. For example, Analytics and Big Data space is big but owned by Hadoop&Co and being seriously attacked by Spark. Event-oriented/stream processing is a contested space now but I never heard GG mentioned in that context. If you squint enough, distributed actors in Akka look similar to GG jobs. I hardly ever see GG in job ads or people's resumes. Is there a market segment large enough left for GG and what is it?

Dec 10, 2014

Aeron


I expect Aeron will follow a trajectory similar to the one Disruptor had. Namely, fewer people will actually use it in production than learn lots of valuable lessons from its design principles. With the exception of a market feed processing system I have actually seen the Disruptor used only in a low latency logger. But I still remember how surprising the emphasis on a single writer and all the nifty non-blocking tricks looked a few years ago. 

So I went through the original Strange Loop presentation for a first taste of extreme messaging wisdom. Well, for better or worse actually learning any new low-level details will take some source code digging. But from the first pass one technique in particular somehow reminded me both Disruptor and Kafka thinking. 

On the sender side, there are three buffers. A clean one (empty for the time being), a dirty one (completely filled, could be used to go a little back in sent message history) and the current one (partially filled, to be used for the next send request). There is a current offset pointer shared by all the sender threads. When a message is being sent, the sender thread first tries to increment that counter with a CAS, retrying if necessary. That CAS allows multiple senders to do most work, including actually writing a message to the output buffer, without synchronization.

On the receiver side, the setup is similar but there are two offset pointers. One for the high water mark (the highest message offset seen so far) and the other for the last message received in sequential order (i.e. without missing messages with lower offsets). The two pointers allow to asynchronously restore the original sequence for messages received out of order. In addition, another thread can periodically check the difference between the two offsets to detect a loss of messages in transit and so the need for re-transmission. 

Dec 5, 2014

AsterixDB


The other day I noticed that the Hyracks research prototype had a continuation. A few years ago it was quite interesting. Back in the day when the Dryad paper was in vogue but its source code was unavailable. The sql-on-hadoop segment was not as overcrowded. Heck, the term itself did not exist then if I remember correctly. I perused their recent papers and I am not quite sure what to think about it. 

I liked the one on data storage because it illustrates nicely the usage of LSM tree indexes. Their streaming support looks somewhat unnatural in comparison with the way event-oriented systems are built and used Storm-style. The critique of mainstream Java memory wastefulness is nice but too basic.They don't even consider vectorization which is such a hot topic in analytics. I'd rather see what the researchers think about using the off-heap storage with the latest JDK. 

I understand that the academics avoid making real contributions to production systems. I am not convinced it's always justified though. Consider something like a DB cost-based optimizer. Judging from the literature it's a legit academic topic. There is only one open source optimizer project I know, Apache Calcite. It's actually a very mature framework with a long history and major products using it. It is also very meta which makes it extremely difficult to really grasp despite of impressive javadocs. I have not found any design papers on it and I regret that deficiency very much. Anyway, the AsterixDB guys don't even mention Calcite/Optiq even though they admit that their own optimizer is rule-based. The way they describe their optimizer makes me think Algebricks wants to be Calcite when it grows up. 

On a lighter note, we are running out of previously unused words. When I hear that AsterixDB is a BDMS I conjure up images of a venerable video genre. When I read about Algebricks I remember Algerbird. 

Nov 30, 2014

Kafka design ideas


A new company behind Kafka development and a fascinating new high-throughput messaging framework making waves inspired me to go look at major ideas in Kafka design. Its strategy in general is to maximize throughput by having non-blocking, sequential access logic accompanied by decentralized decision making. 
  • A partition corresponds to a logical file which is implemented as a sequence approximately same-size physical files. 
  • New messages are appended to the current physical file. 
  • Each message is addressed by its global offset in the logical file. 
  • A consumer reads from a partition file sequentially (pre-fetching chunks in the background). 
  • No application level cache. Kafka relies on the file system page cache. 
  • A highly efficient Unix API (available in Java via FileChannel::transferTo()) is used to minimize data copying between application/kernel buffers. 
  • A consumers is responsible for remembering how much it has consumed to avoid broker-side book keeping. 
  • A message is deleted by the broker after a configured timeout (typically a few days) for the same reason. 
  • A partition is consumed by a single consumer (from each consumer group) to avoid synchronization. 
  • Brokers, consumers and partition ownership are registered in ZK. 
  • Broker and consumer membership changes trigger a rebalancing process. 
  • The rebalancing algorithm is decentralized and relies on typical ZK idioms. 
  • Consumers periodically update ZK registry with the last consumed offset. 
  • Kafka provides at least-once delivery semantics. When a consumer crashes the corresponding last read offset in ZK could lag behind. So the consumer that takes over will re-deliver the events from that window. 
  • Event-consuming application itself is supposed to apply deduplication logic if necessary.

Feb 26, 2012

Basic document and event analytics

Imagine that you want to add to your system the ability to find documents and events related to the "current context". Such a context could include important words, periods of time or topics from a preconfigured catalogue. The documents such as PDF and Word files could be found automatically by a web crawler or entered by placing them to a periodically scanned location. The events could be twitter and blog postings or GOOG alerts.

Ultimately, and especially if you expect to find most of input data by crawling at scale, you will likely want to apply data mining techniques for classification, clustering or recommending. Those are vast areas and so in the first version you will probably limit yourself to a simplified approach where some useful functionality can be provided without a Hadoop server farm and a room of data scientists.

Strictly speaking, document and event analytics are two distinctly different capabilities with their own data flows and available open source technologies. There are also certain architectural similarities between the two.

It is common in Big Data systems to have two independent workflows sharing the same data storage. One, typically batch-oriented, is responsible for data collecting and preprocessing. The other, basically real-time, is responsible for responding to user requests with the most relevant subset of preprocessed data. The batch-oriented workflow is often made scalable by running it on MapReduce-style middleware. The request-time workflow is frequently based on a NoSQL in-memory database to support scalability and high availability.

Document analytics
Preprocessing workflow:
  • A new documents is found by a crawler (web or file-system for non-public documents) 
  • The document is parsed into plain text and some accompanying metadata, the text is analyzed 
  • The filtered text is indexed and saved to persistent storage

Request processing workflow: 
  • A user sends a request with a context descriptor 
  • The request is converted to a query against the index shared with the preprocessing workflow 
  • The returned results are ranked using information from the context descriptor

Event analytics
Preprocessing workflow:
  • Feed funnel receives new events such as Twitter notifications or detects newly posted RSS updates
  • Relevant fields are extracted from the original raw data and packaged into event format used by the system
  • The events are filtered by relevance criteria such as subject or period of time 
  • The filtered events are aggregated by correlating with previously received events, known consumers or according to some preconfigured rules. In process, the original events are often converted into new events and fed back into the system.
  • The final leaf-level events are placed into persistent storage. The storage is likely to behave like a cache to allow expiration of obsolete events. 

Request processing workflow: 
  • In a pull-style design, a user sends a request with a context descriptor. In a push-style design, a long running session registers its interest in the events satisfying some predicates.
  • Query parser uses the context of the filters to retrieve matching events from the storage.

Available open source components

Assuming no need for classification, document analytics can be considered an exercise in full text search. Pretty much all open source products rely on the Lucene indexing engine for this purpose. Tika is another almost universally used library for parsing complex formats such as PDF. 

Even those the two libraries meet core functional requirements, they do not address questions such as scalability, pre-configured text processing pipelines or remote API. Those are answered by what is known as text search servers. They mostly provide support for distributed indexing and replace coding to Lucene API with XML file-style configuration of prepackaged functionality. The most popular alternatives:
  • Lucene core - a low-level approach which could make sense for a proof of concept project but would require some coding to productize the result.
  • SOLR - the flagship text search engine that requires a lot of XML configuration but has exhaustive documentation and a big community.
  • Elastic search - advertised as a more scalable SOLR (and until recently, with better near-real-time support) with significantly less documentation but no need for XML. It is likely to appeal to non-Java people.
  • Zoie - a niche player originally designed for for near-real-time search
Event analytics can be treated as a classical ESP problem. The world of open-source ESP/CEP is more obscure, probably because ESP originated in rich domains such as Finance and Telecoms. Until very recently Esper had been the only option. Unfortunately Esper does not support distributed processing in its community edition. Now there are two promising alternatives:
  • Esper - the golden standard of Java ESP equipped with a nice SQL-like DSL. The lack of scalability makes it appropriate only for proof of concept projects.
  • Yahoo S4 - still quite immature but with a good pedigree from a large enough company
  • Twitter Storm - probably even less mature, uses rather exotic technologies (Clojure, ZeroMQ), was written by what looks like a team of one. Functionality-wise, it is advertised to have delivery guarantees (which S4 does not have). In addition, it currently has a limited choice of integration plugins.
Functionality required for persistent storage of events is general enough to allow using any in-memory  product from Infinispan (simple and well-known) to Cassandra (popular at scale).

Jan 29, 2012

Scheduler design in YARN / MapReduce2 / Hadoop 0.23

YARN

The Hadoop branch started with version 0.23 introduced a few significant changes to Hadoop implementation. From the middleware perspective, the most salient one is a brand new approach to scheduling. Previously, Map and Reduce slots used to be explicitly distinguished for scheduling purposes which limited potential for using Hadoop framework in non-MapReduce environments. In YARN, one of the key goals was to make the entire framework more generic, Mesos-style, so that internal distributed machinery can be used for other computational models such as MPI.

I am particularly curious about the possibility to heavily customize YARN for running distributed computations based on a non-standard computational model. Now that even MSFT dropped their, arguably more powerful alternative, Hadoop is destined to become the most advanced open source distributed infrastructure. So many in-house computational platforms could potentially benefit from reusing its core components.

There is still interesting tension between fundamental Hadoop trade-offs and low latency requirements. To guarantee control over resource consumption and allow termination of misbehaving tasks Hadoop starts a new JVM instance for each allocated container. JVM startup and loading of task classes take a few seconds which would be spared if JVMs were reused. This approach is taken by, for example, by GridGain for the price of inability to restrict resource consumption of a particular task on a given node. We'll see how Hadoop developers extend resource allocation to CPUs which Mesos achieves with Linux Containers.

Scheduler API

The way the Scheduler API is defined has significant influence on how general and so reusable YARN will be. Currently, YARN is shipped with the same schedulers as before but if it gets used outside of default MapReduce world custom schedulers will likely be one of the most popular extension points.

If you look at the diagram below, you can see key abstractions of YARN Scheduler design.

  • Scheduler is notified about topology changes via event-based mechanism
  • Scheduler operates on one or more hierarchical task queues. A queue can be chosen in Job Configuration using the mapred.job.queue.name property.
  • Scheduler allocates resources on the basis of resource requests that currently are limited to memory only. A request includes optional preference for a particular host or rack.
  • Scheduler supports fault tolerance with the ability to recover itself from a state snapshot provided by Resource Manager

Scheduler events include notifications about changes to the topology of available nodes and the set of running applications:
Side notes

It remains to be seen how quickly YARN takes over classical Hadoop. Even though the new code base is more interesting, it seems to be rather immature despite three years since the previous major version. Some things which surprised me:

  • There are important pieces of code such as Scheduler recovery from failures which are currently commented out. I also saw a TODO item stating something like "synchronization approach is broken here" in another place. Probably it's me, but before I merged a couple files with trunk I had not been able to build original v0.23.
  • The good news is that they finally migrated to Maven. The bad news is that for a project of this magnitude they have only a handful of Maven modules and those are strangely nested and interleaved with multiple non-maven directories.
  • It's quite odd not to have any dependency injection framework used in a large server-side Java system.
  • Not only they have their own tiny web framework but it is mixed with pure server-side code
  • Even though Guava is declared as a dependency it is used only sporadically. As an example, they even have their own Service hierarchy

Proliferation of Hadoop branches does not make it any easier I guess.

Oct 31, 2011

Parsing queries with ANTLR using embedment helper

ANTLR is the most popular Java-based parser generator used by many products from Hibernate to Hive. There are two extremely well written and highly practical books written by the ANTLR developer. The most recent Martin Fowler's book also relies on ANTLR for external DSL examples.

So much high quality information could be confusing because not only each language can be described by a few different grammars but there are different options for parser designs. The rule of thumb seems to be that building and then walking the AST is appropriate when "compilation" is expected to have multiple passes (e.g. for tree rewriting for optimization purposes). In case of a one-pass "compilation" it could be more beneficial to use what Martin calls Embedment Helper.

The idea is to have a Builder-like object called by parser on detection of every interesting script element. The object is implemented in target language and embedded into the generated parser as Foreign Code. Thus, neither resources are spent on AST processing nor ANTLR grammar becomes unintelligible because of a lot of injected code.

A typical use case implies building for future use of a semantic model as opposed to executing the script while parsing. The model is populated by embedment helper. This post is intended to highlight a few simple idioms for developing a query parser with this approach.

Grammar headers
  • Remember that if you name your grammar TestQuery then ANTLR will generate classes TestQueryParser and TestQueryLexer
  • You will want to put them into some package using @header and @lexer::header sections
  • Embedment helper-related code will go to the @member section
grammar TestQuery;
tokens {
 SELECT = 'SELECT' ;
 FROM = 'FROM' ;
 LONG_TYPE = 'LONG';
 DOUBLE_TYPE = 'DOUBLE';
}

@header {
package net.ndolgov.antlrtest;

import org.antlr.runtime.*;
import java.io.IOException;
}

@members {
private EmbedmentHelper helper;
}

@lexer::header {
package net.ndolgov.antlrtest;
}

Embedment helper interface

The helper is a reference to an interface that defines a callback method for each interesting grammar element.

/**
 * Embedment Helper object (see http://martinfowler.com/dslCatalog/embedmentHelper.html) called by ANTLR-generated query parser.
 */
interface EmbedmentHelper {
    /**
     * Set storage id
     * @param id storage id
     */
    void onStorage(String id);

    /**
     * Add variable definition
     * @param type variable type
     * @param name variable name
     */
    void onVariable(Type type, String name);
}

Production rules

Now let's look at the rest of the grammar file to see how the helper is invoked. Note that the top-level production rule name "query" will result in a method called "query()" generated by ANTLR in the parser.

query : SELECT variable (',' variable )*
   FROM ID {helper.onStorage($ID.text);};
  
variable : type=varType id=ID {helper.onVariable(type, $id.text);};
     
varType returns [Type type]
 : LONG_TYPE {$type = Type.LONG;}
 | DOUBLE_TYPE {$type = Type.DOUBLE;};
  • Notice how Java code is embedded into the grammar using curly brackets e.g. "{helper.onStorage($ID.text);}"
  • Notice how to refer to a parsed piece of the query e.g. "$ID.text" in case of storage id
  • Notice how an alias can be used for the same purpose e.g. "type=varType" and later "{helper.onVariable(type ..."
  • Instead of returning just a substring of the original query a rule can return a Java object type e.g. "varType returns [Type type]" where Type is enumeration type. To create its instance we assign "{$type = Type.LONG;}". This also shows how to use Java enumerations in ANTLR grammars.

Embedment helper injection
Let's look again at the member section, this time updated to include helper injection.
@members {
    private EmbedmentHelper helper;

    public  EHT parseWithHelper(EHT helper) throws RecognitionException {
        this.helper = helper;
        query();
        return helper;
    }
}

The parseWithHelper method shows three convenient points:
  • the method returns exactly the helper type it is given
  • it assigns given helper to private variable so that it can be called from the code embedded into the grammar
  • it hides the call to the top-level query method of the generated parser

Parser facade
It could be also convenient to hide parser instantiation from the client code with a facade class:

/**
 * Parse a given query and return extracted execution-time representation of the parsed query
 */
public final class QueryParser {
    /**
     * Parse a query expression and return the extracted request configuration
     * @param expr query expression
     * @return extracted request configuration
     */
    public static QueryDescriptor parse(String expr) {
        try {
            final TestQueryParser parser = new TestQueryParser(new CommonTokenStream(new TestQueryLexer(new ANTLRStringStream(expr))));
            final EmbedmentHelperImpl helper = parser.parseWithHelper(new EmbedmentHelperImpl());

            return helper.queryDescriptor();
        } catch (RecognitionException e) {
            throw new RuntimeException("Could not parse query: " + expr, e);
        }
    }
}

Parsing error processing
One last thing to remember is error processing. In our case we just throw a runtime exception.

@members {

    public void emitErrorMessage(String msg) {
        throw new IllegalArgumentException("Query parser error: " + msg);
    }
}

Complete source code is available as a maven project on GitHub

May 30, 2011

Zookeeper-based implementation of GridGain Discovery SPI

The GridGain team is not particularly fond of Zookeeper. Their argument is based on the need for a rather static configuration which is somewhat at odds with the original vision for the highly dynamic GridGain framework. Nevertheless, nowadays a growing number of products use ZK as their discovery service to enable SOA-style architecture. Once you have ZK installed it's quite natural to try to reuse it to keep track of all kinds of servers in your system including GridGain nodes.

GridGain is a pretty interesting niche player. To the best of my knowledge they were the only open source Java product started as a computational grid framework when everyone was implementing K/V-store-style distributed caches. Curiously enough, nowadays everyone is bolting on something similar while GridGain is working on a distributed cache of their own.

When I first learned about GridGain a couple of years ago the two most striking features for me were their SPI-based architecture and the quality of their javadocs. Except for the kernal, pretty much every piece of functionality in GridGain belongs to one of a dozen well-defined SPI interfaces. So theoretically if you feel like you can provide your own implementation for a slice of the system. On the flip side, the entire framework is still shipped as one jar file and so Maven-based configuration management is complicated by the need to explicitly exclude at least half of transitive dependencies pulled in by default.

The Discovery SPI has a few obvious implementations out of the box. The recommended one is a recently rebranded and updated TCP-based implementation. Its protocol seems to have gossip characteristics but there is also the notion of coordinator. Under the hood, this implementation relies on an auxiliary notion of IP finder to determine the initial set of peers to try to connect to. If a connection succeeds, the peer server will forward the connection request to the coordinator. The coordinator is responsible for keeping a versioned view of the current topology.

The easy way

The TCP Discovery implementation internals are pretty complicated because the connection state machine, the corresponding message processing logic and socket IO all reside in the same Java class. Nevertheless it looks like the easiest way to integrate with ZK is to implement a new IP Finder. At a high level:
  • Choose a persistent znode with a sufficiently unique name/path (especially if you share a set of servers among multiple Gridgain clusters)
  • When a new instance of your Finder is being created: connect to ZK, check whether the base znode exists and create it if necessary, set a watcher for children of the base znode
  • When your ChildrenCallback is notified extract urls from the returned list of strings
  • When the local GridTcpDiscoverySpi instance registers an address with its Finder, create an ephemeral child node with a name based on the address (e.g. following the "hostname:port" convention) under the base znode
  • There is no need to remove the children created by a GridGain instance when the instance is stopped. ZK will detect a closed session and delete the corresponding nodes automatically.
I had a positive experience with an implementation along these lines with the previous version (3.0.5c). My understanding though is that this is more of a hack because the TCP SPI implementation will have all the socket-based machinery running redundantly. The DRY principle is clearly broken here. So this approach is probably more of a prototype to see how much you like ZK-based Discovery.

The right way

As they say, no pain no GridGain ;) And apparently the only correct way is to create a brand new Discovery SPI implementation. The good new is that there are at least two detailed examples: the JMS-based implementation and the Coherence-based one. The latter seems to be more relevant because Coherence is certainly more similar to GridGain than JMS is.

The bad news is that the Discovery SPI itself is not as trivial as one might think. You will need to keep in mind a few design details pertaining to the way the SPI is used by the rest of the system. Even when you think that some of them might be optional you will likely discover (ha-ha) that in reality other parts of the system actually use them.

The notion of GridGain Node includes the following elements:
  • a UUID used to uniquely identify a node (so you'll need to map them on whatever you put into ZK)
  • one or more IP addresses associated with the node (certainly to be held in ZK)
  • a set of node attributes created when the GridGain instance starts and given to the SPI by the kernal (another thing to put into ZK). Attributes are frequently used by SPI implementations to exchange configuration information via a generic mechanism
  • an implicit set of periodically refreshed node metrics. Here comes the pain - they change and so a chunk of data will need to be written to ZK periodically (and trigger a bunch of updates in ZK server and other GridGain instances).
  • a discovery listener to be notified about nodes that joined or left for internal event-based bookkeeping.
In comparison to the previous approach, the node representation in ZK is not only larger but also more dynamic. So a more promising ZK node structure could be borrowed from Norbert:

  • Create two persistent children of the base znode called "members" and "available"
  • The "members" node will have an ephemeral child node for each GridGain instance with the node state (attributes, metrics, probably UUID) kept as data 
  • The "available" node will have an ephemeral node for each IP address of all currently available instances (e.g. assuming the same "hostname:port"  name format). Each "available" node would refer to the corresponding "members" node (to simplify support for multiple IP adresses for the same instance).
  • The SPI would listen for changes to "available" nodes to detect joined and left nodes and to "members" nodes to detect updated metrics
The middle way

I am wary of frequent updates written to ZK. So another option would be to keep only URLs in ZK as in the first approach and come up with a metric exchange protocol of your own (with manually created sockets or even Netty). My inner child thinks it could be fun. My inner engineer thinks that enough already of reimplementing the same wheel over and over again. So this middle way would trade fewer ZK updates for a lot of coding and very real potential for subtle bugs.