Adapting a gRPC service to REST with ScalaPB Gateway

Now that we have seen a Scala gRPC service and a Scala RESTful service it's only natural to wonder if there is any middle ground. You want to have a nice and performant backend but you could also  have a lot of RESTful legacy code or be really into curl. Your wish is my command.

The first piece of the puzzle is the fact that GOOG has a standard for mapping REST APIs onto protobuf declarations. With no hacks or any hard thinking you can augment your existing proto files to be REST-friendly.

The second piece is making use of such a mapping. Apparently there is a Go implementation of what is known as gRPC Gateway. It can work with any gRPC service and so could be used with the ones implemented in Scala. But there is another, arguably more appealing option. The ScalaPB library we met before can generate a Scala RESTful gateway too. I guess it's not as well tested or supported but then again there is not that much code behind it to prevent it from serious use.

So I went ahead and updated my two-service gRPC example to support ScalaPB REST Gateway generation. It intentionally borrows a lot from the pure gRPC example so that the difference can be easily seen by comparing files. Technically, you still need exactly the same gRPC service implementation as before. The proto files are almost the same.

There are two new objects. To begin with we need a new abstraction to wire together the generated Gateway code. It's essentially a gRPC client with hooks to plug the RESTful stuff into it. The other is just a typical HTTP client. I followed the path of least resistance and used Jackson for JSON marshaling and commons-http for sending requests. There is a little wiring but in general the code speaks for itself.

What I learned from this exercise is that making gRPC co-exist with REST is easy. I am not sure if running a Gateway locally for each service instance is a better idea than creating a single API Gateway service to front the rest of the backend.

As a parting note, I am surprised there is no other IDL for describing RESTful services for the purpose of code generation. With a standard AST and parser library it would be quite strait forward to generate REST endpoints for any popular service framework. It's equally surprising that the Akka team does not have a generator for akka-http endpoint from protobuf declarations. It looks like such a natural step for them and the ScalaPB project even has the foundation in place.

Asynchronous RPC server with Akka HTTP

At work my team maintains a number of RESTful services. The good news is that they are implemented with akka-http which makes things tolerable. The bad news is that it's all about REST and JSON. So we are having a lively debate about a possible migration to something more like gRPC.

Personally, I am not a fan of non-binary protocols and data formats. At least partially it is related to my experience in building analytics systems where any query routinely requires a lot of data. Yes, for the time being you have to use REST/JSON to communicate between the browser and your API Gateway. But using the same approach to make your backend services talk to each other never made sense to me. I am really looking forward to the day when WASM/HTTP2&Co finally make front end development great again.

In the mean time I would like to see if I can gather enough evidence in favor of my position. So in the spirit of my recent gRPC experiment I am going to implement basically the same API with REST. As a bonus, I want to mention a couple of observations I made using akka-http. As usual, the code is on github. I continue the tradition of implementing a service with two endpoints. I even mostly follow the same code structure as before with gRPC.

The picture above sketches high-level interactions. There are a few noteworthy Akka-related details inside but nobody will be surprised by another RESTful service implementation.

• TestServiceA - is where the actual logic would be. It also hosts a trait and case classes which would be generated from a protobuf file in case of gRPC
• HttpEndpointA - an actual HTTP endpoint that plugs service logic into HTTP network endpoint. A good example of Akka HTTP expressiveness.
• AkkaHttpServer - the service implementation entry point. In typical Scala fashion it directly implements the main method and performs basic Akka dependency wiring. It is also responsible for binding request handlers to the underlying HTTP server.
• Http - technically an object in a high-level Akka API it represents the usual functionality of request/response-style interactions over HTTP 
• AkkaHttpClient - not shown; wraps akka-http-based client logic. I spent some time getting it right because client-side examples are harder to find. In particular it illustrates how to use Spray JSON marshaller. But it could be any HTTP client capable of sending a POST request with a JSON body. 

If you are curious enough to look at the code I would like to highlight a few details useful in real life. Not all of them are prominent enough in usual examples. The HttpEndpointA class would be a good illustration of most of them.

• ExceptionHandler - you can configure a handler for exceptions uncaught otherwise. It is the last line of defense against unexpected problems. 
• RejectionHandler - such a handler is very useful if you want to explicitly deal with invalid requests (e.g. garbled JSON or more likely a JSON body representing a request to some other service)
• request timeout - even though Akka has a default global property ("akka.http.server.request-timeout") you could configure it explicitly for a service that is expected to take an unusually long time 
• implicit JSON (de)serialization with Spray - it's very convenient to use but rather puzzling initially to configure. In addition to the marshaller object holding implicit vals notice how the "import ServiceAJsonMarshaller" statement makes it usable by "as" and "complete" directives
• composing multiple Routes - different styles are possible here; if there were another URL to support in this endpoint we could (a) declare another "def process2: Route" and then compose the two with "process ~ process2" using another somewhat cryptic Akka directive

On the client side there are a few more details to mention. My first reaction was that Akka HTTP client code looks somehow less straitforward than what one would see with, say, commons-http. Any HTTP client would work for this example but I wanted to see one with akka-http to better grok the API.

• JSON marshalling with Spray was a puzzle again. It was not clear to me from the official documentation but googling helped.
• Somewhat unusually, the client requires the same ActorSystem and Materializer infrastructure as the server. 

My conclusion is that if circumstances force me to implement anything RESTful I will reach for Akka HTTP first. But in comparison with real RPC it's more work to produce less safe code. But that's a topic for another day.

gRPC in Scala with ScalaPB

Curiously enough my old post on gRPC still gets disproportionately many hits. Now that I work with Scala full time I was curious to see how my old Java example would compare with a Scala version. Spoiler alert: I don't have that much new to report. Other people covered it better before. This post probably makes more sense as a head-to-head comparison between Java and Scala versions.

From what I see Scala people tend to avoid Java libraries. I am skeptical it's a good idea especially around networking and concurrency where the Java ecosystem is at its best. But I am still learning and have no strong opinions developed yet. After all, last time I used Java j.u.c Futures, Guava Futures, and CompletableFutures were still all in the game.

In case of gRPC the unofficial Scala code generator is known as ScalaPB. Once you configure it in your SBT project it feels as natural as the official Java version. The good news is that the generated code operates in terms of Scala Futures and so feels idiomatic. But you still need to initialize the gRPC foundation and that part looks almost the same as in Java.

As before, the goal is to support two different service endpoints on the same server and to have fully asynchronous client and server implementations. As we concentrate on the request/response approach we completely ignore the question of streaming. gRPC bootstrap code is so simple that I could not find enough of it to extract anything truly generic. So even though I notionally have a server and a client they just wrap initialization logic and might be unnecessary.

There are a couple technicalities I would like to highlight:

• I had to google and experiment to find out how to configure ScalaPB for a multi-module project and a non-default proto file directory
• even though your actual logic is all about Futures and so requires ExecutionContexts, it's easier to start with an Executor to name threads and configure underlying thread pools and then wrap it into a context
• I can think of a few slightly different ways to propagate ExecutionContexts; implicit vals are idiomatic but seem to be less readable despite saving a little typing

All in all, using ScalaPB feels a little more idiomatic but not that different from how it's done in Java.

Spark Data Source API V2 - write query planning and execution

Before you can read anything you need to have it written in the first place. The previously considered read path is a little simpler but certainly shares a lot with the write path we'll delve into today. Initially I am going to concentrate on the basic flow and ignore predicate push-down and column pruning. As my running example I assume a trivial Spark expression that uses some existing Dataset such as "df.write.format("someDSV2").option("path", hdfsDir).save()".

A data source is represented in a logical plan by WriteToDataSourceV2 instance created by data frame writer. During planning DataSourceV2Strategy transforms the relation into a WriteToDataSourceV2Exec instance. The latter uses the Spark context to execute the plan associated with the Dataset. The resultant RDD determines the number of input partitions and so the number of required write tasks. In addition, WriteToDataSourceV2Exec helps with keeping track of WriterCommitMessages used to make writing data source partitions transactional.

Query planning

In this phase the goal is to create a WriteToDataSourceV2 from a Dataset and client-supplied options and then transform it into a WriteToDataSourceV2Exec. The process is initiated by saving a Dataset. It triggers both logical and physical planning stages.

• DataFrameWriter searches for an implementation of the DataSourceV2 marker interface. The implementation is expected to also implement the WriteSupport mixin interface that is the actual entry point into writing functionality
• WriteSupport is a factory of DataSourceV2Writer instances
• DataSourceV2Writer can create a DataWriterFactory and is ultimately responsible for committing or rolling back write partition tasks
• WriteToDataSourceV2 is the logical plan node that represents a data source
• A QueryExecution is associated with the WriteToDataSourceV2 to help with coordination of query planning
• SparkPlanner uses DataSourceV2Strategy to transform logical plan nodes into physical ones
• DataSourceV2Strategy recognizes data source logical nodes and transforms them into physical ones
• WriteToDataSourceV2 carries a DataSourceV2Writer and logical plan associated with the Dataset; both are given to WriteToDataSourceV2Exec
• WriteToDataSourceV2Exec is the physical plan node that represents a data source

Query execution - Driver side

In this phase the goal is to obtain data partitions from the RDD associated with the Dataset and trigger execution of a write task for each partition. In this section we look at the Driver-side half of the process.

• WriteToDataSourceV2Exec creates a DataWriterFactory and a write task for each partition
• The Spark context executes the tasks on worker nodes and collects WriteCommitMessages from successfully finished ones 
• WriteToDataSourceV2Exec uses DataSourceV2Writer to commit or rollback the write transaction using collected WriteCommitMessages

Query execution - Executor side

The second half of the execution phase takes place on the Executor side. 

• DataWriterFactory is serialized to worker nodes where DataWritingSparkTask are executed
• Executor provides a DataWritingSparkTask with a Row iterator and a partition index
• The task acquires a DataWriter from the DataWriterFactory and writes all partition Rows with the DataWriter
• The task calls the DataWriter to commit writing and propagates a WriteCommitMessage to the driver

Spark Data Source API V2 - read query planning and execution

Without much marketing noise the Spark team decided to dramatically overhaul the current data connector API in v2.3 The new version was quietly merged into master recently. In comparison with the original approach the new one definitely looks like a real API. Even though nobody will be surprised by further evolution of the version one can see in master it already looks irresistibly promising. So I decided to spend some time digging into it. 

You probably remember the high-level architecture of the Spark query compilation process. It follows the traditional DBMS approach of parsing a query into a logical plan, optimizing the logical plan into a physical plan, and executing the physical plan on worker nodes. I am going to structure my investigation along similar lines.

The following discussion does not shy away from relevant Spark query execution internals. Those are always fascinating even when not directly actionable. Moreover, the Data Source API implementation is a nice vertical slice of the generic Spark infrastructure. So this is a convenient opportunity to refresh my memory. I will omit fine details of the Spark internals in an attempt to highlight the big picture.

As my running example I assume a trivial Spark expression such as "SparkSession.read.format("someDSV2").load.count()".

Query plans and Data Sources in a 100 words or less

Logical and physical plan nodes are subtypes of the QueryPlan. Spark Planner is responsible for transforming a logical plan tree into a physical one. The planner relies on a configurable set of strategies for actual transformation rules.

A data source is represented in a logical plan by DataSourceV2Relation instance created by data frame reader. During planning DataSourceV2Strategy transforms the relation into a DataSourceV2ScanExec instance. The latter creates a DataSourceRDD instance that is used to actually compute the results in parallel from multiple partitions.

Logical query planning

In this phase the goal is to create a DataSourceV2Relation from client-supplied options. The process is initiated by loading a Dataset.

• DataFrameReader searches for an implementation of the DataSourceV2 marker interface. The implementation is expected to also implement one of ReadSupport mixin interfaces that allow the engine to know optional capabilities of the data source
• ReadSupport is a factory of DataSourceV2Reader instances
• DataSourceV2Reader knows the data source schema and can create ReadTasks
• DataSourceV2Relation is the logical plan node that represents a data source
• Dataset owns a QueryExecution and is associated with the DataSourceV2Relation

Physical query planning

In this phase the goal is to transform a DataSourceV2Relation into a DataSourceV2ScanExec. The process is initiated by executing an action on the Dataset created by the previous phase.

• Dataset owns a QueryExecution
• QueryExecution coordinates query planning
• SparkPlanner uses DataSourceV2Strategy to transform logical plan nodes into physical ones
• DataSourceV2Strategy recognizes data source logical nodes and transforms them into physical ones
• DataSourceV2Relation carries a DataSourceV2Reader that is given to DataSourceV2ScanExec
• DataSourceV2ScanExec is the physical plan node that represents a data source

Query execution

In this phase the goal is to obtain data partitions from the RDD created by  DataSourceV2ScanExec and collect data Rows from all the partitions.

• DataSourceV2ScanExec creates ReadTasks using its DataSourceV2Reader and converts the tasks into a DataSourceRDD
• DataSourceRDD wraps each task into a DataSourceRDDPartition that knows its id
• ReadTask represents a single data partition. A ReadTask is serialized and sent to a Spark worker.
• After DataSourceRDD execution is triggered, on executor side the DataSourceRDD obtains a DataReader from the ReadTask
• DataReader is an iterator of Rows from one data partition  

Parquet file query DSL with FastParse

A year after my last foray into DSL parsers I find myself working with Scala full time. My first impression is that Scala folks like re-implementing things from scratch despite easy access to the first-class Java library ecosystem. With the standard Scala library having surprisingly high for a mature language turnover (e.g. see original Actor library replaced with AKKA, very inefficient original parser combinators, Scala Collections overhaul, Monix tasks trying to catch up with CompletableFutures) I am not entirely convinced it's always justified.

Nevertheless it could be helpful to be able to see the same DSL grammar implemented using another modern parser library. FastParse seems to be the current favorite. While trying to wrap my head around it I went back to the original grammar and ported it to FastParse. There is not that much to see because the actual parser is a single class of fewer than 100 LoC. I implemented it mostly following the example of Parboiled-based parser but it also has family resemblance with ANTLR grammar.

Among my first impressions:

• FastParse version is the smallest in LoC (with a factor of 1.5-2)
• FastParse can easily express basic lexer rules that I had to experiment with harder in ANTLR 
• it takes effort (and copy-pasting from examples) to deal with FastParse "grammar" code. Overloaded Scala operators are reasonable and even convenient but their non-standard nature requires some getting used to
• there is no need for additional MVN/SBT plugins
• Scala itself is the largest transitive dependency, FastParse is a couple of JARs of under 400K
• FastParse generates reasonable traces for debugging 
• it's very easy to enable case-insensitivity though it adds noise to the parser
• it's very easy to automatically skip whitespaces though the syntax for doing it looks odd
• map-based syntax for building custom AST nodes is pleasant
• online documentation is quite extensive and includes a couple large-scale parsers
• you can use FastParse from Java; I did it to re-use my tests written for last year's parsers

In general I definitely like FastParse more than Parboiled. This comparison is somewhat unfair because in comparison with Java any Scala code is more terse. I would go as far as to say that for tiny DSLs FastParse should be the default choice. For a large grammar readability, visual tools, and likely performance of ANTLR are still as important as ever.

Tantalizing promise of GPU for analytics workload

For the last few years the trend in analytics systems has been to implement traditional MPP-style architecture as an open source, usually JVM-based product. Impala was the only exception. Surprisingly, its C++/LLVM foundation did not make it the leader of the "SQL on Hadoop" space which is probably a testament to modern Java performance. 

More recently, everyone began converging on the triad of columnar storage (off-heap when in-memory), vectorized operators, and operator byte code generation. Parquet / Arrow-style columnar format helps with compression and reducing disk-to-memory data transfer volume. Vectorization can take advantage of SIMD (e.g. see Lecture #21 - Vectorized Execution) or at the very least alleviate CPU cache line thrashing. 

By v2.0, Spark had covered most of the ground. It's interesting to note that even though cached dataframes are stored in  a compressed columnar format, physical operators are still not vectorized and so process a row at a time (e.g. see a rare glimpse of the internal machinery in the design doc attached to SPARK-14098 "Generate Java code to build CachedColumnarBatch"). 

Judging from SPARK-15687 "Columnar execution engine", the need for vectorized operators vis-a-vis whole stage code generation seems to be a topic to seariously debate this year. Nevertheless it would still be a well understood approach other features such as SPARK-19489 "Stable serialization format" could benefit from.

What is easy to miss behind all this progress is that the GPU revolution ignited by DL/ML quietly arrived to the analytics world too. My high-level understanding of the GPU is essentially "very fast operations on columnar data once you copy it from main memory to the GPU". Which almost looks like very advanced SIMD with high latency. So my initial expectation was that analytics queries would not be much faster because of the sheer data volume required by an average query.

I was clearly wrong. My favorite source of query engine benchmarking statistics shows that MapD and Brytlyt score very high. Both were designed from ground up for the GPU. One challenge with GPU-related technology is that it's way too low-level in comparison with any JVM-based development. 

So most people will probably end up just learning the big picture. For a reasonable overview of GPU usage in the OLAP context, I would recommend the Red Fox research project and their "Red Fox: An Execution Environment for Relational Query Processing on GPUs" paper in particular.

To the best of my knowledge, this topic is of limited interest for Spark project currently. At least partially because it would take columnar data formats and vectorized operators to align Spark internals with GPU coding idioms and data layouts. It probably makes sense to keep an eye SPARK-12620 "Proposal of GPU exploitation for Spark" and whatever else its author does even if that prototype did not succeed.

Rules of three in software development

Most people agree that software development is a craft. As such, it accumulated heuristics and rules of thumb that are hard to prove in any remotely scientific way. But see enough code bases over a few years and certain patterns will certainly look plausible. I would like to ponder a couple of representative examples that struck a chord with me.

Make it work, make it work right, make it work fast

I believe this saying originated in the heydays of the OO. You could probably interpret it as "building a vertical slice of the system" in RUP or "YAGNI and iterations" in Agile. If you squint enough even MVP-centered thinking common in startups could be another reincarnation of the same principle.

In practical terms it boils down to:

• find a small use-case, choose OSS libraries that can help implement it quickly, prepare project infrastructure (e.g. build system, code tree structure), write the code, produce something small but runnable and call it a prototype
• once you acquire basic experience with new technologies and have your prototype to share with other people you are ready to have a meaningful technical discussion with them; if all goes well, extend the prototype into an MVP that could be deployed to production
• once you have a basic version running in production, start working on a second major release that can improve scalability and stability

My last experience with this flow was with Spark and Parquet. In mid-2015 I prototyped a few basic ideas such as exporting data from our system as Parquet files and running Spark locally to work with Parquet files. I made a few observations about small details not discussed in the documentation. So when ideas for a new ETL pipeline were discussed I could show actual code examples closely related to our particular needs.

Once we could show some numbers from running the prototype in Spring 2016 other people were comfortable with making an architectural decision. By Fall 2016 we had an MVP that could do most of what the legacy code was capable of. While migration to the new pipeline started in production in late Fall, we could switch attention to the second major release which addressed stability issues and applied data storage optimizations.

OSS/3d party, low-level API, less generic in-house implementation 

Another evolution I see frequently goes like that:

• Come up with big ideas for a new service
• Quickly develop a prototype using an OSS product that implements some of the big ideas
• In case of encouraging performance test results, finish the MVP and go to production
• Once initial performance gets insufficient, start digging into the OSS code and documentation is search for obscure lower-level APIs; replace usage of simple and easy APIs with more efficient but cumbersome ones
• Once you understand your system trade-offs and data access patterns consider having your own, less generic but more optimized for your circumstances replacement for the initially used OSS component  

A recent example would be solving the challenge of data retrieval for interactive analytics. Postgres was not fast enough anymore. The big idea was to use search engine-style approaches such as fast bitmap indices. The obvious OSS candidate was Lucene. We used a less popular but still high-level API with additional optimizations added later. 

Once we had it in production the urgency subsided and we had time to recognize some context-specific assumptions about our data we could make. The second major release dived much deeper into Lucene internals and resulted in a few customizations built to take advantage of what we found. At some point in the future the next step could be going the same way as Druid did - replacing Lucene altogether with an in-house implementation of the inverted index.

Three kinds of developers

By the virtue of being a human issue and not just a technical question this example is harder to discuss. As a matter of fact, there is a beautiful post that you should read even if you ignore the rest of that highly recommended blog. 

There is no question that both individual psychology and level of experience play a huge role in this. Unfortunately all I can imagine to be actionable about it is to try to be self-aware enough to see those patterns in yourself and your teammates. 

It might be also the case that if you are a settler you will not like A-round companies because of all the chaos, poor engineering, and one-man components. Conversely, a pioneer might feel strange after a B round when a larger team starts building real technology and team communications and documentation needs grow in importance.

Probably because of the time I spent in B-round startups I believe I have seen more pioneers and settlers than town planners. The latter are probably those mythical "enterprise developers" frequently mentioned on HackerNews. 

Compressing time series with Gorilla in Java

You might know FB for their censorship efforts and spying on the users but did you know they also write software? :) In a two-year old paper they discuss a very insightful and practical approach to compressing time series. Better yet, they even have an OSS implementation. Though C++ is not hard to read, it makes the original code not that useful in the JVM land. 

First, a brief recap of the compression ideas:

• The data model is a sequence of (timestamp:int64, value:double) pairs. They split a sequence into blocks aligned with two-hour windows but we'll ignore such higher-level concerns.
• Timestamps and values are compressed separately. 
• Timestamps are stored as the first timestamp value followed by variable-length encoded delta of deltas. Four value ranges are supported ([-63:64] , [-255:256], [-2047:2048], "fits into 32 bits").
• Values are stored as a sequence of the first value followed by variable-length encoded values produced by XOR-ing next value with the previous one. The binary representation of an XOR-ed value is split into "leading zeros", "block value", and "trailing zeroes" parts. The numbers of leading and trailing zeroes are stored as bytes.

I wanted to see how well Gorilla compression can cope with less predictable data. The kind of time series I am interested in is not as periodic as IoT data. So I ended up porting a couple of classes to Java. The time series class can be used for writing data or reading it but not both simultaneously. 

The current implementation is intended to help with testing compression quality and does not necessarily support other uses. The unit test shows how to read a compressed time series. The only catch is that reading the first timestamp requires a dedicated method call while all the values can be read calling the same method.

Parquet file query DSL with ANTLR

Last time on Sheldon Cooper presents fun with flags we saw a lightweight alternative to ANTLR used to parse a simple query DSL. I was having too much trouble with another alternative so in the meantime I decided to finish a proper ANTLR version of the parser. As luck would have it, in ANTLR4 there is more than one way to write a parser. So for a low, low price of a single grammar file you get a parser written in the officially recommended way and the other following a more traditional approach.

Strictly speaking, there are two grammar files but if you squint a little you can see one was created by adding actions to a copy of the other. ANTLR4 makes an attempt to avoid actions in a grammar file to simplify the file and make possible parser generation in languages other than Java. 

Instead of actions one can implement a parse tree listener interface. The good news is that it indeed allows better isolation between grammars and Java code. The bad news is that there is even more choice if you want to do something that requires sharing state between parser rules. Like building an AST.

Of the two simplest choices, I decided to use a manually maintained stack. It could probably be done with a visitor implementation returning AST nodes but the listener with a stack approach makes the juxtaposition with Parboiled even better. A new trick in ANTLR4 is using "# Something" snippets in the grammar to make ANTLR generate a listener method for every rule alternative. 

The listener ended up looking more similar to the Parboiled parser than I had expected. That made me think about trying the traditional approach with actions embedded into grammar. At first glance, the listener approach looks cleaner indeed at no discernible cost.

ANTLR is still very good with skipping white spaces at the lexer level. No need to contaminate the parser rules with redundant checks everywhere. But to my surprise I actually found a couple other problems. To begin with, there is no easy way to make keywords case-insensitive. Secondly, I failed to write a parser rule that would return "unquoted" string values without breaking the ID lexer rule. I am probably not the only one confused with the latter. After some experimentation I simply settled on manually calling String::substring.

In addition to two spectacular books there is a lot of online documentation as well. The ANTLR maven plugin works seamlessly with no additional configuration so the gossip of "additional build process step" is widely exaggerated. 

Between an easier to read grammar file and very powerful IDE integration ANTLR remains the leader. But from what I see the code size and complexity are roughly the same as for Parboiled. There seems to be the only 300K ANTLR runtime library. Parboiled has half a dozen dependencies (mostly ASM jars) of comparable total size.

So as a provisional conclusion, I don't feel strongly anymore that ANTLR is the only right answer to parsing needs in Java. I should probably run a performance test at some point but I doubt +-20ms is of real concern to most small DSLs consumers.

Parquet file query DSL with parboiled

Now that we know how to push down Parquet predicates the next logical step is to come up with a DSL to make writing predicates easier. The DSL would be handy for debugging, especially if it can be implemented with few additional dependencies. Running Spark just to have access to Dataset query syntax is not always convenient.

What I have in mind is something very straightforward, after all there is little else to express than predicates. It could be as simple as "SELECT * FROM '/tmp/somefile.par' WHERE ('column1'=42) || ('column2'=24)" to filter a file by a couple of columns.

When I think about query parsers ANTLR is the default choice. Its functionality, stability, and documentation are outstanding. But the syntax I mentioned makes me think that a lighter approach could be possible. So to a significant degree this exercise will be an attempt to compare a few parser libraries in the context of very small DSLs.

I decided to start with a heterogeneous AST representation and the pseudo-SQL syntax I mentioned above. To evaluate the alternatives I am going to pay attention to:

• parser-related code size, complexity
• grammar readability
• transitive dependencies brought by the parser library
• build process changes required (e.g. maven plugins)
• tool chain, support for debugging of parser errors

Today's contender is Parboiled and its Java library in particular. The actual parser for my DSL is a reasonably small class. Syntactically, the PEG rules look quite isomorphic to how I imagine the corresponding CFG. The library provides a built-in stack that is very helpful for building AST nodes. When the stack is not enough, one can also use action variables that allow collecting results from other rules. 

Parboiled wiki provides a lot of useful information and in addition there are a few nice parser examples. Parboiled is a pure Java library so there is no need for additional maven plugins. It has just a few ASM dependencies. As for debugging, I found the tracing parser very convenient. 

Coding-wise, it took me some time to get used to calling "push/pop" as the last Rule in a Sequence. I also started with too many action variables but was able to covert most of them to operations on the stack. 

One minor problem with Parboiled is the lack of easy means of skipping white spaces in the tokenizer. I am pretty sure my current parser can be broken by adding space characters to a query here and there. Guarding every Sequence with a white space consuming rule seems to be the only viable approach.

Parquet file filtering with predicate pushdown

As I discussed before, Parquet file format is extremely well suited for storing data in analytics systems because of its columnar approach. But it gets better (even after you take into account column data compression). Parquet library supports predicate push-down which makes it very complimentary to Spark-like analytics query engines. Depending on your data model and data distribution it could be possible to skip entire blocks when reading a Parquet file.

The Filter API supports predicates that can use usual comparison operators on column field values. Individual predicates can be composed using logical AND and OR operators. In addition, user-defined predicates are supported. They can skip an entire file block if the column metadata indicates that the block has no matching values.

The Filter API is comprised of just a few abstractions:

• FilterCompat.Filter - the push-down predicate interface taken by ParquetReader builder as an optional parameter
• FilterApi - builder methods that create individual predicates and compose them together 
• UserDefinedPredicate - the interface to be implemented by user-defined filters; it supports block-level predicates in addition to usual record-level ones

I updated my Parquet example to demonstrate usage of a disjunctive filter and a user-defined one. At a high-level:

• FilterApi is repeatedly called to resolve a column name into a column reference, create individual predicates using the reference, and compose predicates
• Behind the scene, a tree of inner Operators class instances is created
• FilterCompat facade is called to wrap the tree into Filter implementation recognized by Parquet file reader
• A new file reader is created using the Filter
• Internally, different visitor types are used to apply record-level and block-level predicates

Parquet file metadata, standard and custom

When writing about Parquet file format before I neglected to mention one auxiliary but potentially useful API. You might remember that a Parquet file is comprised of blocks. Even though each column is compressed individually, the block determines the rows that will be compressed as a single chunk. The API I would like to discuss today provides access to the file metadata, both custom and built-in.

If you are interested in using Parquet files with no help from a Spark-like query engine chances are you will end up using metadata. It could be as simple as attaching a schema version. Or you might want to utilize column statistics to optimize query processing predicate pushdown-style.

Every Parquet file has detailed metadata associated with it. If you see it in the picture you can read it. Among the attributes most likely to be useful are:

• schema name
• row count for every block
• the number of NULLs, MIN and MAX values for every column in a block
• application-specific key/value pairs you can attach to a file

I have an example of reading standard metadata. There is a static method that can return a ParquetMetadata instance for a given file path. From there you can traverse all blocks and their columns. No surprises.

In addition to the attributes defined by the Parquet format you can also attach arbitrary String key/value pairs to a file. In the write path, your WriteSupport class can override the finalizeWrite() method to return a custom metadata Map. In the read path, you have access to the map in the same place you get access to the file schema. Alternatively, the already mentioned standard API allows you to read custom metadata as well.

Spark SQL connector design

Now that we have discussed the Data Source API in general it's time to consider what happens when you actually start coding. This post will rely on the first iteration of my experiment. I cannot promise that its design follows the only possible allocation of responsibilities. It will also not work in a real cluster because for the time being I am ignoring the need for a DFS. 

What is real are interactions between the abstractions comprising the connector, a few useful RDD/SparkContext API tidbits, and the problems one faces trying to write a connector for a non-trivial storage from scratch.

As I mentioned before, I am using Lucene to represent the data storage to integrate into Spark. For simplicity sake I am making a few assumptions here: only point fields are supported, a field visitor is used to retrieve values, no effort is made to implement a scalable iterator-based approach to search. 

Squeezing performance from Lucene quickly leads away from the standard API and so remains out of scope. I am actually curious about the latency achievable when millions of documents with DocValue fields are retrieved and one day I could end up trying it.

Connector structure

There are many Spark connectors like it, but this one is mine :) To begin with let me introduce key abstractions and their responsibilities:

• DefaultSource - the connector entry point that is essentially a factory of LuceneRelation instances
• LuceneRelation - represents an application-specific dataset (e.g. the data that belongs to a particular version of some tenant's data); technically, it incapsulates a LuceneRDD and its schema
• LuceneRDD - an RDD of LucenePartitions; one way to think about it is a facade for multiple storage partitions. Every store or retrieve operation is delegated to all the partitions.
• LucenePartition - a subset of original data operated on by Spark as an RDD partition (i.e. iterator of Rows); an entire partition can be retrieved from storage or its subset can be specified with a combination of required columns and row filters
• Spark API extensions - a package object making possible to add methods to DataFrame and SQLContext

In addition to the aforementioned classes that mostly implement Data Source API interfaces there are a couple less generic classes:

• LuceneSchema - the data schema representation used by the connector and comprised of a Spark schema (fields types) and a Lucene schema (can a field be retrieved and/or filtered on?). The latter is necessary because not all Lucene documents are required to have all the fields and so reverse engineering the schema from a Lucene index would be prohibitively expensive
• LucenePartitionStorage - a means of using different storage implementations in read and write paths. In the read path the storage is an actual Lucene index. In the write path the storage is a temporary in-memory data structure holding DataFrame rows before they are written to Lucene index.

To better understand how these abstractions fit each other let's consider interactions among them in the write path and the read path separately. Both paths are used by a unit test and can be seen in action.

Write path

The write path starts from a DataFrame instance that you want to write to the storage. You also need to come up with a file system path (e.g. a tenant-specific subdirectory under some root location) and a Lucene schema (to let Lucene know which fields to index and/or store).

• DefaultSource is asked for a new LuceneRelation instance
• DefaultSource calls the extended DataFrame API to save the data frame
• For each partition of the RDD wrapped by the data frame a new LucenePartition is created
• A new LuceneRDD instance is created from the LucenePartitions using the data frame schema and the client-provided Lucene schema
• The composite schema is written to a file (with the expectation to be shared by all partitions)
• For each LucenePartition a subdirectory is created 
• Every LucenePartition is written to a Lucene index created in the subdirectory
• A new LuceneRelation is created to represent the written data

In the diagram above:

• The RDD[Row] is the RDD wrapped by the data frame
• The RDD[LP] stands for an RDD[LucenePartition]
• When a LucenePartition instance is created from DataFrame rows it is backed by an in-memory storage implementation; currently it simply remembers the original row iterator
• The in-memory storage only hold data temporarily before it's actually written to the Lucene index
• The details of Lucene index writing are omitted because they are quite generic and have nothing Spark-specific

Read path

The read path is a mirror copy of the write path. It starts from the client calling the SparkContext to load a data frame from a given location.

• DefaultSource is asked for a new LuceneRelation instance
• A new LuceneRelation instance is created with a schema loaded from the given location
• The number of LucenePartition is found by scanning the location for subdirectories
• The sequence of found partition identifiers is parallelized to make loading of all partitions in parallel possible
• A new LucenePartition is created for each found partition
• A LuceneRDD instance is created from the LucenePartitions and the loaded schema
• At this point the LuceneRelation is created and so is the data frame the client can see
• The unit test demonstrates a few different ways to query after the loaded data frame is registered as a temporary SparkSQL table. All of them are internally translated into the same three method calls seen in the diagram above. The most interesting of the three is the buildScan method
• The LuceneRDD delegates the call to its partitions and than flatMaps partial results into a single row iterator
• Each partition compiles the required column list and the filters into a Lucene query, executes the query, collects the values from the matching documents, and converts every Lucene document into a Spark Row

As another side note, the Lucene query API is iterator-unfriendly. So in this version every partition collects all the Rows into an array which would probably kill it in real life if the query selectivity is low. 

Spark SQL Data Source design considerations

I like learning about query engine internals. Spark SQL Data Source API seems to be a reasonable gateway drug on the way to really understanding Spark executor and optimizer. In more practical terms, the efficiency of data storage integration is likely to dominate any high-performance Spark-based architecture. Especially with the core engine getting so many optimizations in v2.0.

Curiously enough, there is no official documentation other than the API scaladocs. Even though the API itself is comprised of just a handful of abstractions, in my limited experience getting a clear picture of what it takes to implement a reasonable data source integration is not easy. I have not seen too many descriptions and even my favorite ones tend to concentrate on the read path or rely on remote access to some actual storage. 

Browsing a few real integrations made me think that not all of them follow the same design. I am still trying to understand if the differences are purely cosmetic or there are indeed multiple ways to structure a Spark connector. With this question in mind I embarked on a little journey through the world of Spark data sources. I am trying to answer questions such as:

• Assuming co-locating computations with the data is the shortest path to low latency, how do I integrate a data storage into Spark without making (too many) remote calls from the workers?
• Can it be done without a distributed file system (assuming, say, AWS S3 instead)?

While reading Succinct tech report I liked their connector so much I decided to follow its design in my first iteration. I decided to use Lucene as my data storage because (A) it will force realistic constraints on my prototype (B) I am interested in using Lucene in OLAP context (C) apparently nobody tried it before so it could be fun.

Spark

Spark is a distributed query engine that by default expects an HDFS-style distributed file system to be available on worker nodes. A popular alternative is to call a remote storage using some TCP-based RPC protocol (e.g. ElasticSearch or SOLR). Some people use S3 instead of a DFS but that approach seems to be less prominent.

Spark API is somewhat confusing because it attempts to hide the complexity of distributed processing behind a rather generic RDD abstraction. And then builds even more abstractions on top of the RDD. In addition, the older low-level RDD API and the higher level modern DataFrame API are closely related. The DataFrame is an RDD<Row> with a schema. RDDs have partitions. The Data Source API operates on Relations.

Lucene

• Lucene is a library that relies on the usual local file system API to write and read index files
• Documents do not have a schema because different documents can use different fields (the set of fields could be small enough to treat it as a schema conceptually but it's application-specific)
• Document fields can be indexed (so you can filter on their values), stored (so you can retrieve a field value from a particular document), or both.
• There are at least two very different flavors of numeric fields (i.e. LongPoint and NumericDocValuesField)
• When you expect your queries to return millions of documents there are a few optimized ways to retrieve those efficiently. The more efficient, the less standard.
• The standard search API can take a callback but there is no iterator-friendly entry point

First Spark Lucene connector iteration

I am learning some crucial details as I go. So the first iteration will be more of an educational exercise demonstrating internal data source interactions rather than a usable piece of software. It remains to be seen how far I'll be able to evolve it from there.

I don't have a DFS. I also believe that trying to write a "Lucene Directory implementation using HDFS I/O Stream API" is not a good idea. Regardless of whether you use HDFS or S3 I would expect that a more promising approach is to read/write Lucene index files locally and then upload/download them in the background.

Initially I will co-locate everything in my prototype and simply use the local file system for any files. My hypothesis is that having a Lucene index for each Spark partition is a reasonable approach. Once I have the simplest possible connector implementation I will see

• If there are alternative connector designs 
• What can be done about storing index files in S3 
• What kind of performance is achievable with DocValues fields (that will likely require a fixed schema but latency should be worth it)

Asynchronous RPC server with GRPC

After a severe disappointment with the state of the Thrift RPC framework I was looking forward to the first stable release of GRPC. Of the original big three only the protobuf-based ecosystem seems to be still evolving enough to be a viable choice. With the arrival of v1.0 it is finally ready for production.

For better or worse despite using a binary transport protocol GRPC forces things with "HTTP" in their names upon us. That is unfortunate because a protocol designed for connecting to browsers outside of the data center will incur some penalties in complexity and probably latency on backend services not exposed to the public Internet. 

The question I want to investigate today is pretty much the same I had playing with Thrift last year. What does it take to configure and run a backend service exposing multiple end points with no blocking calls. Spoiler alert: it's so trivial I should probably double-check my understanding of the threading model to confirm GRPC is as reactive as it seems to be :)

To begin with I would like to illustrate a typical GRPC call. Let's assume we have a service API with one method. GRPC uses protobuf3 as its IDL and, other than not having required fields anymore, I find it hard to see any difference from protobuf2 or even Thrift. 

The first pleasant surprise is that a single maven plugin with a simple, mostly OS-specific configuration is all it takes to compile IDL into Java classes. For each XXX service you basically end up with two classes:

• xxxProto - request and response classes
• xxxGrpc - request handler base class and client-side factory method for different IO styles

The second pleasant surprise is how uniform server-side API is. When processing a request you are given an Observer with the same onNext/onComplete/onError methods used for all four currently supported RPC types. On the client side it's even simpler, all you need is a request instance and a guava callback instance.

where

• a Channel is created for a TCP destination
• a service stub using the new Channel is obtained
• a callback instance is created to process server response
• an RPC method is called on the stub
• a mirror copy method is called on the request handler but with an observer instead of a callback
• the handler calls either onNext/onSuccess or onError observer methods to finish processing
• once the server response is received, GRPC invokes the callback on a thread from a pre-configured executor

Server-side initialization is equally terse. You give it a handler instance for each service end point, an executor, and a port to bind to. Four lines of code and you are in business.

In my example I created two nearly identical protobuf files to describe two service end points. There are two abstractions to establish a connection on the client-side and bootstrap the server on the server side. I followed the same CompletableFuture-based approach to implementing request handlers as discussed in the Thrift post. A unit test wires everything together and calls both end points in parallel.

So far so good. But what ointment would be without a fly? There are a couple of things that I either don't understand or need more time to sort out. And at least one of them is probably necessary for production deployments. 

A trivial note is that GRPC is only a second library I am aware of that depends on JUL (Parquet being the first but only temporarily).

One odd question I still have is how asynchronous Java GRPC implementation really is. What confuses me is that I can see an "asynchronous basics tutorial" for C++ only. Is it some special kind of asynchronicity attainable only by those who manage memory manually? Or is there some blocking still left in Java GRPC library?

A real question is the complexity of GRPC security. My example follows the lead of GRPC Java tutorial when it calls "usePlaintext()" on channel builder. At first glance I am not even sure if SSL/TLS is necessary for the traffic inside of the data center or whether AWS ELB could interfere with it. A topic for another day I guess.

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.