Splout SQL User Guide

You need to have Java >= 1.6 preinstalled. Download a release of Splout from Maven central:

You’ll need to set properly a few environmental variables (i.e. at ~/.bashrc)

For Hadoop >= 2.X (YARN):

Example of env variables for Cloudera CDH5 using parcels:

For Hadoop < 2.X:

HADOOP_HOME must be properly defined pointing to your Hadoop installation folder.

Launching the server daemons is as easy as:

It is possible to have more than one QNode and more than one DNode in a single machine, but for that it is important to understand the Configuration and modify a few properties, specially if you launch the services from the same folder.

Logs are stored in logs/ folder from where daemons are launched. Commons-logging is used for logging.

If you find any strange problem or issue please contact us or raise a bug in Github.

> Back to table of contents

A table schema’s is defined the same way a Pangool Tuple schema is defined. However, the data types are adjusted to match those which are compatible with SQLite. The following table shows the correspondence between a Pangool type and the underlying SQLite type used:

A table is either partitioned or replicated to all partitions. However, a tablespace should have at least one partitioned table for the indexer to be able to distribute the data among partitions!

Partitioning is the basis of Splout and it allows it to balance data before indexing and deploying it.

The most usual case of partitioning is columnar partitioning, meaning that a table is partitioned using one or more columns of its schema.

When more than one table is partitioned in the same tablespace, they must be co-partitioned using the same kind of fields. For example, if a tablespace A contains tables A1, A2 and A3, and A1 and A2 are partitioned tables and A3 is replicated to all partitions, then if A1 is partitioned by a pair of (string, int) columns, then A2 should also be partitioned by a pair of (string, int) columns.

Therefore, it is also possible to partition using arbitrary functions, for example a javascript function that takes the first two characters of a field. You can check this in the Advanced API section.

Because data is explicitly partitioned, the user must also explicitly provide a partitioning key when querying Splout. For example, if a dataset has been partitioned by "customer_id", then the user will provide the appropriated "customer_id" together with the SQL query when querying Splout through its REST interface.

> Back to table of contents

It is fairly easy to install Splout in a distributed environment. By default, Splout will use Hazelcast in Multicast mode for finding members, but it is possible to configure Splout for explicit TCP/IP or Amazon AWS auto-discovery. Following there are some examples of distributed configurations:

> Back to table of contents

The command line tools have been developed to ease the most common use cases. There are two "generator" tools that are responsible for launching a Hadoop process that will balance and index the input files, and there is one "deployer" tool that is able to deploy the result of any of the generators to an alive Splout cluster.

The tools allow to process either textual files (CSV or fixed-width), Cascading binary files, Hive tables and Pangool Tuple files.

All the command-line tools use the underlying Java API that we have implemented for Splout. You can also use this Java API directly in your Java project and you can have access to more advanced features such as specifying custom partitioning functions, record processing functions and such.

Note: For using Hive with Splout, it is recommended to add Hive conf/ and lib/ folder to the HADOOP_CLASSPATH environment variable:

In this way, Splout will be able to locate the appropriate Hive metastore and sample the Schema of Hive tables implicitly.

For importing tables using the "simple-generate" tool we can use arguments "-hdb" and "-htn" for indicating the Hive database and the Hive table we want to import. We will need to specify that the input type is "HIVE" through "-it" property and add the rest of parameters needed (number of partitions, partitioning columns, output folder, table name and tablespace name). Note how we don’t need to specify input paths anymore as the input is read directly from Hive.

Or we can use the "generate" tool. For that we can create a JSON tablespace descriptor like the example one. We specify the input type and the Hive database and table names. Note how we don’t need to specify input paths.

We can then execute the JSON descriptor tablespace generation with the "generate" tool as usual:

And finally, with either methods, we can deploy the generated database as usual:

When using the Java API, we can add Hive table inputs with methods addHiveTable() from TableBuilder.

Keep in mind that if you are using Hive with default embedded Derby database you can’t import Hive tables while you have another session opened in Hive at the same time, as the import process will try to connect to it too.

Note: When using advanced Hive features such as the OCRInputFormat, it is also needed to add the hive-exec JAR via "-libjars", so that Mappers and Reducers can understand this native format:

Splout SQL can import binary Cascading files directly without needing to convert them to text. For that we just need to specify the input path where the output of the Cascading process is, and a list of comma-separated column names. The type of each column will be automatically discovered from the binary Cascading file.

For importing a Cascading table using the "simple-generate" tool we can use argument "-cc" for indicating the comma-separated column names, together with CASCADING in "-it" argument, and the rest of arguments as usual. See the following example:

Or we can use the "generate" tool. For that we can write a JSON descriptor as follows:

We can then execute the JSON descriptor tablespace generation with the "generate" tool as usual:

And finally, with either methods, we can then deploy the generated database as usual:

When using the Java API, we can add Cascading table inputs with methods addCascadingTable() from TableBuilder. Keep in mind that in order to have this functionality in your Java project you will need to explicitly import the appropriated Maven dependency:

Which you can get from Conjar’s Maven repo:

Binary Pangool Tuple files can be imported directly using inputType "TUPLE". Both the types and the column names will be read from those files so we need to do nothing special besides indicating the inputType and the input path.

We can import data from Pig by storing it as Pangool Tuple files. For that we need to obtain the pangool-core JAR. We can download it from Maven Central if we don’t have it somewhere else. We then register the JAR in our Pig sessiona s follows:

And we can save any output as:

Note how the first argument to the StoreFunc is the table name, and the rest are column names. The following example executes a simple word-count task in Pig and saves the output as a binary Pangool Tuple file. We will use an input file called mary with the following content:

This is the script code:

We can then import the resulting binary file as a normal Pangool Tuple file, for example with the "simple-generate" tool:

Use this method for obtaining an overview on the cluster status. Returns a QNodeStatus status object. Example response:

Use this method for getting the list of alive DNodes. Returns a list of strings. Example response:

Use this method for getting the list of tablespaces being served in the cluster. Returns a list of strings. Example reponse:

Use this method for getting the associated information of a tablespace being served by the cluster. Returns a Tablespace object. Example response:

Returns all available versions for the specified tablespace, including the one which may be being served at the moment by the cluster. Returns a Map<Long, [Tablespace]>. For every version in the key of the map, returns the associated Tablespace object information. Example response:

Returns a DNodeSystemStatus object filled with the detailed information of the specified DNode. Example response:

Perform a SQL query and get a JSON result back. Returns a QueryStatus object with some metadata about the query execution. Example response for key=Up, sql=SELECT * FROM pagecounts LIMIT 1 and tablespace pagecounts (http://localhost:4412/api/query/pagecounts?key=Up&sql=SELECT%20*%20FROM%20pagecounts%20LIMIT%201;):

By providing a list of DeployRequest objects we can perform a deploy through this POST method. We need to specify a tablespace name, an URI from where binary objects must be fetched, a PartitionMap and a ReplicationMap and optionally a list of SQL statements that will be executed each time a connection is made to the database.

Example post body:

The deploy is an asynchronous operation coordinated by the cluster members. A deploy request returns inmediately with a [DeployInfo] object indicating whether it could be started or whether there was some error trying to start it:

By providing a list of SwitchVersionRequest objects we can perform a rollback through this POST method. We just need to specify the tablespace name (which must be being served by the cluster at the moment of the rollback) and the version we want to set it to.

Rollback is a synchronous operation and the return type is [StatusMessage]:

> Back to table of contents

There are a wide variety of reasons why queries may or may not perform well. It is important to understand that each query is executed in a SQLite connection of a binary SQLite file. When troubleshooting query performance, it is advisable to take a look to the SQLite documentation and in addition check things like:

Remember that you can fine-tune your data indexing process with initialStatements, preInsertStatements, postInsertStatements and finalStatements (see [JSONTableDefinition]) and initStatements (see [DeployRequest]). All these features are available as part of the standard Java API of TableBuilder and TablespaceBuilder. Use them for adding custom PRAGMA or custom CREATE INDEX commands if needed.

It is also very important to keep in mind that colocating data in disk is crucial for query speed. For example, if your query impacts 1000 records and these records need to be loaded from the main table, if they are not colocated in disk and the database doesn’t fit in memory, it would mean that the server has to peform 1000 potential seeks. You can control the data colocation policy with "insertionOrderBy" as explained in [JSONTableDefinition], which is also available in TableBuilder.

As an example, we have used data colocation techniques within Splout SQL to obtain < 50ms average query time with 10 threads on dynamic GROUP BY’s that hit an average of 2000 records each in a multi-gigabyte database that exceeded available RAM in orders of magnitude in a m1.small EC2 machine.

Because deployments are asynchronous operations, it can be tricky to know if they have succeeded or not. You can monitor a deployment using the information returned by the API (deployInProgress flags).

A deploy may fail if some of the DNodes fail, if a timeout is reached (see Configuration) or if the leader QNode dies. The default timeout is 10 hours, so it shouldn’t be hit under normal circumstances.

In big networks in can take some time for Hazelcast to negotiate the membership. It can be a matter of minutes.

You might have to tune your open files limit as Splout is opening one connection to each partition file it has for every serving thread. So if you have 64 serving threads and 20 partitions to serve, this means up to 1280 opened files which is more than the default in some machines (1024). You can check and change your limits: http://www.cyberciti.biz/faq/linux-increase-the-maximum-number-of-open-files/