Customizing Lucene scoring loop with DIY scorer

Since my last Lucene post I have been shown what seems to be the ultimate way to customize Lucene scoring loop. If you squint enough, you could probably see it discussed in blogs and official documentation. But it is rather hard to visualize it in detail unless you are already well knowledgeable of the internals. In a certain sense it's more like wholesale replacement of the standard Lucene query processing path with something more lean and specialized. 

As usual, I am mostly interested in using Lucene in the OLAP context as opposed to classical full-text search. It implies the need to retrieve values from all the documents matching some predicates. It also mostly requires numeric fields. There are a few systems out there that internally use an inverted index for fast analytics and you could imagine them using Lucene in early prototypes.

Before we start, I should probably remind you that internally Lucene creates multiple index segments. Every segment is represented by a leaf reader context in the search path. You can easily obtain them from the standard IndexReader. And they will be our entry point into the DIY scoring loop.

If you are interested in this area, you most likely have some proprietary query DSL. Under normal circumstances, an expression in that DSL format is converted to a Lucene query. Most likely it takes a top-level BooleanQuery with multiple clauses to represent all query predicates. Internally, Lucene uses such constructions as ConjunctionDISI to represent those clauses.

Fundamentally, the idea is to:

• use your DSL instead of standard Lucene Query/Weight abstractions
• implement your own scorer hierarchy, mostly TermScorer-like ones for numeric types and Conjunction/Disjunction ones to compose them into trees; yours will be simpler and you also will be able to support resetting them and save on memory allocation
• implement you own query compiler to transform your DSL query into a composite of the new scorers
• continue using document processing of the kind discussed in previous posts; this is an efficient way to accumulate results in a more iterative manner

To illustrate this approach I implemented one scorer and a couple of toy DSL classes to represent a query, a compiler for it, and a document processor. With the exception of the scorer, all of them would be different in your system and so cannot be implemented in a reusable way.

In my example I have got a few representative abstractions and their most trivial implementations ever. Any realistic implementation would venture too far from Lucene proper but you should be able to see the idea. A quick look at the test code should be enough to see how the parts fit each other.

• IndexSearcher - a perfectly standard Lucene searcher that we use to find leaf contexts
• QueryBuilder - represents a DIY Lucene query compiler that transforms a DSL query into a Scorer (in real life, most likely hiding a Scorer tree behind a composite class)
• Searcher - DIY IndexSearcher replacement that knows how to find index segments and apply the same scorer and processor to all of them
• Processor - a mean of collecting field values from matching documents
• QueryScorer -  DIY scorer that replaces Scorers and Weights and can be reused

I can imagine a less onerous approach where you would reuse standard scorers. Unfortunately it requires dealing with many low-level details and interfaces even though you presumably don't need their functionality. Well, if nothing else, this discussion illustrates how little it takes to retrieve data from a Lucene index when you don't need to actually score the results. 

Processing data without deserialization with flatbuffers

While reading about Arrow I was reminded about flatbuffers. I decided to try them for a simple case to see how they compare to hand-coded data structures. The promise was that given a schema described with an IDL similar to PBs/Thrift, the flatc compiler can generate data structures that do not require deserialization to perform operations. Which comes in handy when you have a massive input data stream.

I created a trivial schema that purports to represent a simplistic time series. It's basically a sequence of (time, value) pairs. I was curious how useful the generated code could be. For example, would it be smart enough to unwrap an array of fixed-size value pairs into two parallel arrays? 

The first nasty surprise was to learn that flatbuffers don't have an official maven artifact. They don't even have a maven plugin. This alone is a huge red flag even though some kind people came to the rescue. At least "brew install flatbuffers" actually installed the flatc compiler on MacOS. 

Flatbuffers have "struct" and "table" abstractions. The former is supposed to be bare-bones serializable data structure. The latter helps with schema changes but adds overhead to store some schema details. I was interested in the most compact format possible. My first attempt was to introduce a DataPoint struct and then have an array of them. No can do, flatc does not allow it. My second attempt was more successful. With two primitive type arrays I was actually able to compile my IDL.

To begin with, flatbuffers internally use ByteBuff instances and not byte arrays. That complicates life if you want something simple. I imagine it should complicate life even if your use case is more complex because there is no support for configurable buffer allocation. You can use a ByteBuff instance of your own but if you fail to size it appropriately, the framework will allocate a new large instance.

When I looked at the way an array is written to the ByteBuff I was disappointed to see that they copy it one element at a time. Simple omissions like that or absence of signatures such as "write(byte[] array, int from, int length)" are unexpected in a framework aimed at people who care about efficiency. It was also disappointing to see 50+ bytes of overhead when serializing a 10*16 bytes record. Those tables are not free.

So flatbuffers could still be an option if you deal with complicated, deeply-nested data structures. For simple record types they do not seem to provide any benefit but still require exorbitant effort in configuration management.

Single producer multiple consumer queue with Disruptor

1. Disruptor with WorkPool

While the first use case that comes to your mind when you think about the Disruptor is "multiple highly concurrent readers seeing every event" there is another, less frequently mentioned feature. The WorkerPool is an abstraction that allows to configure multiple concurrent readers with the Disruptor load balancing among them. In such Disruptor setup, every event written to the ring buffer is processed by a single reader.

Imagine that you want to process a stream of records on as many CPU cores as you have got. One easy way to set up this configuration is to use the Disruptor. If nothing else, you should expect reasonable concurrency out of the box.

2. Disruptor initialization and shutdown 

The disruptor was not designed to have a complicated lifecycle. Its original architecture pretty much assumes that you start it, and then it runs until you shut down the entire thing. So if your input is not really a continuos stream but something more like a large file iterator then some attention should be paid to stopping the Disruptor safely.

If you have workers stateful enough to require a new group for every input stream you'll need to stop the Disruptor processing the previous stream before setting up a new one with a different WorkerPool. Moreover, the Disruptor is very good at saturating CPUs so if you allow it to use all of them you will probably want to have a request queue in front of it to prevent creation of multiple Disruptors.

In case of WorkerPool, you need a producer thread and a consumer ThreadFactory to be called by the Disruptor. If you represent your producer as a Runnable instance, you could re-use the same single-threaded executor for multiple requests. As you have no direct control over the threads created by the Disruptor with the ThreadFactory you will rely on the Disruptor shutdown call to stop them.

The Disruptor shutdown procedure is quite straightforward. Once you know that all the input events were put into the ring buffer you are supposed to shut down the Disruptor with a timeout long enough to allow the remaining events to be processed by the workers. One way to do it would be to create and wait for a latch on the client thread and make the producer thread count down the latch when it's done with the input. 

3. Disruptor WorkPool example

While experimenting with a real-life prototype I came up with a complete flow illustrating Disruptor WorkPool usage in a backend service.

• Client - represents a backend service; owns the producer executor, catches exceptions
• ParallelRequestRunner - hides a few steps required to process one input stream; calls a ParallelProcessor to process input, waits for the producer to finish
• ParallelProcessor - creates a new Disruptor with a WorkerPool of a given size; submits a Runnable wrapper of producer for execution
• DataRowEventConsumer - processes a single event at a time
• DataRowEventProducer - reads input and puts one event at a time into the ring buffer
• ParallelProcessingContext - keeps track of the participants to allow waiting for the producer to finish, triggers Disruptor shutdown, collects statistics from the consumers

Druid and Pinot low latency OLAP design ideas

0. Introduction

Imagine a data schema that can be described as (M, D1, .. , Dn, {(t,v)} ) where every entry has

• M -  a measure (e.g. Site Visits)
• Di - a few dimensions (e.g. Country, Traffic Source)
• {(t,v)} - a time series with a measure value for each timestamp

Calling it a time series OLAP could be a misnomer. Time is just another dimension in a typical OLAP schema. Time series usually implies just a metric without dimensions (e.g. server CPU usage as a function of time). Nevertheless in real life systems dealing with such data is quite common. They also face similar challenges. Chief among them is the need for low latency/interactive queries. So traditional Hadoop scalability is not enough. There are a couple systems out there that are less well known than Spark or Flink but seem to be good at solving the time series OLAP puzzle.

1. Druid ideas 

• For event streams, pre-aggregate small batches (at some minimum granularity e.g. 1 min)
• Time is a dimension to be treated differently because all the queries use it
• Partition on time into batches (+ versions), each to become a file
• Each batch to have (IR-style): a dictionary of terms, a compressed bitmap for each term (and so query filters are compiled into efficient binary AND/OR operations on large bitmaps)
• Column-oriented format for partitions
• A separate service to decide which partitions to be stored /cached by which historic data server
• Metadata is stored as BLOBs in a relational DB
• ZK for service discovery and segment mappings
• Historical and real-time nodes, queried by coordinators separately to produce final results

2. Pinot ideas

• Kafka and Hadoop M/R based
• Historical and real-time nodes, queried by brokers
• Data segment : fixed-size blocks of records stored as a single file
• Data segments
• Time is special. Ex: brokers know how to query realtime and historic nodes for the same query by sending queries with different time filters. Realtime is slower because it aggregates at different granularity ("min/hour range instead of day")
• ZK for service discovery
• Realtime has segments in-memory, flushes to disk periodically, queries both.
• Multiple segments (with the same schema) constitute a table
• Column-oriented format for partition for segments
• A few indices for a segment (forward, single-value sorted, )

3. Common threads

• Column oriented storage formats
• Partition on time
• Information Retrieval/Search Engine techniques for data storage (inverted index)
• ZK is a popular choice of discovery service

4. In other news

There also smaller scale products using search engine technology for OLAP data storage.

Asynchronous AWS S3 file transfer

I am probably ten years too late in writing about the oldest AWS service but the power of AWS SDK asynchronous file transfer is irresistible. Moving a 300MB file takes just a few seconds. It is very easy to plug this API into reactive backend services. For this exercise we assume that one service uploads a file to S3 and another service periodically checks for new files in that location.

Fundamentally, asynchronous operations require an instance of the TransferManager class and a callback to process status notifications. I wrapped the whole process into a few classes representing abstractions for uploading to, downloading from, and detecting newly uploaded files in a pre-configured location in some S3 bucket location.

The TransferManager API typically takes a Request object and an asynchronous status listener. It returns a Transfer instance that can be used to retrieve error message in case of failure. Polling an S3 location for available files requires a loop because the results are returned in batches. Checking if a file exists at a given S3 path  is implemented as an attempt to fetch the corresponding file metadata and treating a thrown exception as "FileNotFound".

S3 is a simple (duh!) service so there are only two additional notes. First, it is a good idea to encode some metadata into file names on S3. Things such as tenant id or version or video resolution. It helps with deciding how to handle a downloaded file by parsing its name. Second, it's convenient to superimpose a "directory structure" onto the flat namespace of the S3 bucket abstraction. 

So a reasonable file naming convention might include a three-part prefix appended to all relative file paths: a backend service name, a file schema version, and a namespace representing either an environment (e.g. PROD) or a developer (in development deployments). The version part in particular makes upgrades much easier in production. For example,
"$BUCKET/prod/somesvc/v2/relative_path/file.ext".

The digram below shows a typical sequence of operations for uploading a file and then finding it with S3 polling from a different service.

In my example,

• FileDownloader / FileUploader - abstractions used by the client to start a file transfer operation
• TransferCallback - the callback interface called by file transfer operations to report final status to the client asynchronously
• S3Destination - a way to specify a common bucket "subdirectory" for multiple files
• S3TransferProgressListener - converts progress events to success/failure notifications; makes possible to extract an error message 
• S3FileTransferClient / S3Client - TransferManager-based implementation of file transfer operations
• S3Uploader / S3Downloader - S3Client-based implementation of file transfer API     
• S3ChangeDetector - a job to be run periodically (e.g. with ScheduledExecutorService::scheduleWithFixedDelay) on the receiver side to look for new files on S3
• FileHandler - the callback interface called by S3ChangeDetector for every found file not seen before (the way of keeping track of previously seen files is likely to be application-specific)

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

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

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. 

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.

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

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. 

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

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.

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.

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

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 :)

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?

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?

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. 

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.