Distributed systems are mainstream in software workflows these days. These are the systems where servers communicate with the client and each other (usually through gossip and consensus protocols). Distributed systems help solve a major problem where you must spread out a common task across multiple servers for performance and availability.

You also need to ensure consistency and fault tolerance in such systems. Otherwise, either task will return an output that results from a data snapshot at two different points in time or an incomplete result because some node failed during processing.

Writing a Distributed System Is Not Easy

If only it were as easy as implementing API calls between servers. Running and maintaining a distributed system is not a simple task in itself.
As you can imagine, writing a new one is much more challenging.

Take CassandraDB as an example. The basic function of the DB is to process a CQL query, gather data from multiple nodes and return the aggregated response.

However, in the background, you have to communicate properly between all the nodes using Gossip, reach a leadership consensus via Paxos, and ensure it is fault-tolerant by distributing SSTables correctly. All of this has to be done while supporting the addition and removing existing nodes while providing the same query performance.

What if you allowed a distributed system to separate its supervision logic from its business logic? That way, the developers of a distributed DB similar to Cassandra can focus only on the query execution part while the fault tolerance, replication, and token assignment are taken care of by this new entity.

Enter Apache Helix

Apache Helix allows developers to express your complex distributed system as a simple finite-state machine. All the management operations are then represented as transitions in this state machine.

This is great because the general view is that a cluster’s operations differ so much from each system to the next that creating a generic framework is impossible. By using Helix, developers can focus on the core functionalities of their systems and let Helix take care of cluster management.

Wait, a Distributed System Is a State Machine?

If you don’t know about state machines, it’s a model that can represent a system in a fixed number of states. It also models what transitions are valid from one state to another.

Let’s take the example of the popular real-time event-processing framework Apache Kafka. A Kafka cluster consists of multiple topics. Each topic consists of multiple partitions.

For a topic partition, Apache Kafka needs to ensure the following:

• There is always one leader. The leader is the only one that can accept the writes.
• There are N replicas of the partition where the user can decide N.
• The replicas should be distributed evenly among M server nodes for a fault-tolerant setup.
• Only replicas in sync with the leader can be promoted to a leader if it crashes.

Helix can take care of all this with the following state machine:

This allows Kafka-like systems to focus only on reading/writing data from the partition logs and other admin operations such as adding/deleting topics, managing offsets, and so on.

Let’s take another example of Apache Pinot.

Pinot has multiple tables, each composed of small segment files. The analogy would be a single Cassandra Table comprised of small SSTable files.

For a particular segment, Pinot has the following constraints:

• There should be N replicas of the segment where the user can configure N
• There can only be one consuming segment for each table. All replicas of that segment should be in a consuming state as well.
• The rest of the segment should be in an online state, i.e., ready-to-serve queries.
• In case M new nodes are added, the new segments should also be distributed evenly across these nodes.
• A segment and replicas should be dropped in case users reach or manually delete a retention duration.

This can be represented with the following state machine:

So, Where Does Helix Come Into the Picture?

The user will upload these state machines (in a form of config) to Helix. Helix then keeps track of all the nodes in the cluster through inbuilt monitoring.

It also keeps track of the state of each partition in each node. Helix then creates a map of what the ideal state of each partition should be. If the partition's actual and ideal state does not match, Helix will queue up State transitions.

The attractiveness of Helix though is due to the following:

• It can do multiple state transitions in parallel.
• Take care of coordination between multiple servers. For example, if a serverB is made leader instead of serverA, remove the serverB from the leadership before adding serverA.
• It can decide the order of state transitions even when done in parallel to not violate the constraints. For example, upgrading a new server with no data to the leader OR assigning some replica partition to a node that has the leader as well.
• Decide the priority of the transitions (which can be modified using a config). For example, if a leader fails at the same time some new nodes are added to the cluster, the transition from replica to leader status should take higher priority than the transition from idle to replica state.
• Modify the ideal state based on cluster monitoring. For example, where to move the partitions residing on the failed node if a node goes down?

Here’s how the state machine for Kafka can be implemented via Helix:

    StateModelDefinition.Builder builder = new StateModelDefinition.Builder("your-state-model");
    builder.initialState("IDLE");

    builder.addState("IDLE");
    builder.addState("REPLICA");
    builder.addState("LEADER");
    // Set the initial state when the node starts

    // Add transitions between the states.
    builder.addTransition("IDLE", "REPLICA");
    builder.addTransition("REPLICA", "LEADER");
    builder.addTransition("REPLICA", "IDLE");
    builder.addTransition("LEADER", "REPLICA");
    
    // Set constraint that there can only be 1 leader at a time
    builder.dynamicUpperBound("LEADER", "1");

    // Set transition priority
    // the first inserted gets the top most priority.
    List<String> stateTransitionPriorityList = new ArrayList<String>();

    stateTransitionPriorityList.add("REPLICA-LEADER");
    stateTransitionPriorityList.add("IDLE-REPLICA");
    stateTransitionPriorityList.add("REPLICA-IDLE");
    stateTransitionPriorityList.add("LEADER-REPLICA");

    record.setListField(StateModelDefinitionProperty.STATE_TRANSITION_PRIORITYLIST.toString(),
        stateTransitionPriorityList);
    

Assigning State to Each Partition

Internally, Helix uses a modified version of the RUSH algorithm to decide the final state and then uses a greedy approach to figure out valid transactions and schedule them. By greedy approach, I mean the transactions are appended to a list of pending transactions until we get two transactions that cannot be done in parallel.

Helix also allows users to plugin any other placement strategy, e.g., ConsistentHashing depending on the use case.

How Is Helix Agnostic of Core System Logic?

It should know how to deal with data movement across servers, how to read data in each partition, and so on, right?

Well, I never said Helix performs the state transition. Helix issues the request with the transition that needs to be performed to the server that needs to perform it. Currently, it is done using Apache Zookeeper as a message queue.

// Helix leader, running on a controller node
state.setPartitionState(partitionName, serverName, "ONLINE");

// Helix Listener, running inside each server
@Transition(from = "OFFLINE", to = "ONLINE")
public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
  String partitionName = message.getPartitionName();
  // Logic to add partition
  // Create partition files
  // add partition to queryable list
}

Each server has a listener which keeps track of new messages in the queue. The listener has the business logic on what to do when asked to move from State A to State B. For example, if it needs to copy data from S3 for bootstrapping or start accepting writes since it has been promoted to leader.

The server doesn’t need to care about what other servers are doing, e.g., who is the leader, if some other server already has the same data, and so on. It simply performs what it’s asked to do and then updates the actual state of the partition. If it cannot complete the task, it reverts to an error, and Helix can take appropriate follow-up action.

Conclusion

Although Apache Helix can help you simplify cluster management to an amazing degree, it still lacks some features:

• Helix can’t take heterogeneous servers into account while calculating the final state. For example, you should be able to schedule twice as many partitions on a node that is 2X larger than the rest of the nodes.
• Helix can’t properly account for replication in a cluster distributed across multiple AZs. In such cases, you must ensure the replicas are present in different AZs.

You can use the following references to learn more about Helix:

• Helix repository — https://github.com/apache/helix/tree/master
• Helix whitepaper — https://dl.acm.org/doi/10.1145/2391229.2391248
• Helix in Apache Pinot — https://docs.pinot.apache.org/basics/architecture

Apache Helix: The Distributed System’s Orchestra Conductor was originally published in Better Programming on Medium, where people are continuing the conversation by highlighting and responding to this story.

Embedded DBs have become commonplace in a lot of new distributed systems. They offer a way to store a lot of data on a local disk, thereby providing good performance compared to a database located on a different server. They are also lightweight, so they don’t put a lot of pressure on the main application running on the same server.

Facebook (now Meta) created RocksDB, which is the crown jewel among the embedded DBs. It is used in popular libs such as Apache Flink, MyRocks SQL engine, and more.

RocksDB has a lot of information on its website, but it is hard to find all the configuration options. As a new user, it was difficult for me to understand which configurations were important to set, what they meant, and where to change them.

This is what I plan to tackle in this article. All the options discussed here are applicable for Block-based table formatand may not apply to the Plain or Blob table format. It is also not an exhaustive list of options but instead the most popular and useful selection.

Use Your Fuel Wisely!

Memory optimization is crucial for any database, and RocksDB is no exception. By maximizing the efficiency of your memory usage, you can improve the speed of both reads and writes in your database. Memory is used for two key structures in RocksDB: the cache and the memtables.

Block cache

RocksDB comes with various types and levels of cache for improving read speeds. The most common one is the Block cache. It is enabled by default with a max size of 8MB. As the name indicates, the block cache stores the SSTable blocks. They can be data blocks, index blocks, filter blocks, or something else. Most of the time, you will want to increase this cache size to improve speed.

RocksDB has two cache options: LRUCache, which is the default, and ClockCache. ClockCache is faster because it doesn't need locks, but it doesn't provide the same guarantees as LRUCache. It was also marked as experimental until November 2022.

Row cache

This cache is used for storing actual data for a key. The cache’s documentation is almost non-existent, but it provides fast data access in case you do a lot of point lookups. RocksDB first looks in the row cache and then in the block cache for the data. Like block cache, you can use either LRUCache or ClockCache as the actual implementation for row cache.

Cache filters — Other than the data blocks, you can also ask RocksDB to cache index and filter blocks. These blocks need to be accessed almost every time to need to look up a key in the data block. Caching these blocks improves the read latencies while trading off memory. The following options can be used:

• cache_index_and_filter_blocks — Set it to true to enable caching these blocks.
• cache_index_and_filter_blocks_with_high_priority — Set it to true so that these blocks are cached in the high-priority region of the LRU cache. These blocks are evicted much later than the low-priority region in case the cache is full.
• pin_l0_filter_and_index_blocks_in_cache — L0 level requires reading all the files since they have overlapping key ranges. Enabling this always keeps the L0 filter and index blocks in the cache

Memtable

RocksDB writes go first to Memtables, a data structure that resides in the memory. These tables are then flushed to the L0 level in the DB. Since memory is a precious resource, you need to control the size of the memtables in the RocksDB and the frequency of flushes.

This is possible via the following configs:

• write_buffer_size — this is the max size of each memtable that RocksDB can create. Beyond this size, a new memtable is created, and the existing one is queued up for flushing to the disk. The default value is 64MB.
• max_write_buffer_number — these are the max number of memtables that can exist in the memory before a flush is triggered. The default is two so that when the first memtable is flushed, the writes can continue on the second memtable.
• min_write_buffer_number_to_merge — minimum number of memtables that will be merged before a flush is triggered. The merge operation happens in memory. This can help you reduce disk io for flushes, especially if memtables contain a lot of duplicate keys or deletes. The default is one, so no memtables are merged.
• db_write_buffer_size — this is the max size of all memtables combined across column families. The default is -1, which means there is no global limit on memtables. If you set it to a value > 0, RocksDB will initialize a WriteBufferManager that keeps track of all memtables sizes across column families.

You can grep for flush_reason stats in the RocksDB Log file to check the size of the column family before getting flushes as well as the reason it flushed. For example, you will see values like Write Buffer Fullwhich means it was triggered because of max_write_buffer_number or you will see a value like Write Buffer Manager which is because of db_write_buffer_size.

Memtable formats

There are four supported memtable formats in RocksDB:

• SkipList
• HashSkipList
• HashLinkList
• Vector

You can go through the official wiki for more detail on these formats. The default is SkipListFactory which performs well for almost every use case. It is the only one that can support concurrent writes. The flushes are also faster for this memtable format.

The second format that you can try is the Vector memtable. It provides good performance if you do a lot of random writes. Some popular frameworks, such as Kafka Streams, use this table format.

Use the Expressway, Not Single-Lane Roads!

IncreaseParallelism — The number of threads RocksDB will use for all background operations. Even if you increase the max_background_jobs but keep parallelism as 1, it won’t lead to a lot of performance increase. Ideally, this option should be kept equal to the number of cores in your machine.

Max_Background_Jobs — These are the number of flushes and compactions that can be scheduled in parallel on your threads. Flushes take higher priority, while compactions take lower priority by default. This is so that a compaction operation doesn’t block any writes to the databases.

CPU usage is also affected by compaction. We will take a look at the compaction options in the next section since they are more I/O intensive than CPU intensive.

Avoid Walls and Slow Corners!

Since RocksDB is a database, it is expected that you’ll be hitting I/O bottlenecks sooner or later. The two most I/O-intensive operations are flush and compaction. Flushes are mostly tuned by adjusting the memtable sizes and numbers. We have already gone through those in the Memtables section.

Compaction

RocksDB performs compactions in the background where multiple SSTables files are merged. It also removes deleted keys from the file. A new SSTable file is created and then put into the same or different LSM level. The following options can be used to tune the whole process:

level0_file_num_compaction_trigger — L0 is the first level in RocksDB. It is where all Memtables are flushed. You can control the number of files that should be present in the L0 level before a flush is triggered. Increasing this config means compactions occur much less frequently but need to process a lot of data.

level0_slowdown_writes_trigger — Compactions occur on low-priority threads by default in RocksDB. Compactions may get delayed due to no available threads. In such cases, you may want to slow down writes so compactions don’t have to process extreme amounts of data when they get scheduled. This config is used for that. By default, the limit is 20 files at the L0 level.

level0_stop_writes_trigger — Similar to the previous, but once the number of files at the L0 level reaches this limit, we stop processing writes altogether and may schedule compaction. By default, the limit is 36 files.

target_file_size_base — RocksDB needs an estimate of how large files it should create on each level starting from L1. By default, this file size is 64MB for L1, and then it gets multiplied by target_file_size_multiplier for the next level. By default, the multiplier is 1, so all levels will have similar-sized files.

max_bytes_for_level_base — Max amount of data that can be present in L1-Level. By default, this is 256 MB. For levels greater than L1, this base size is multiplied by max_bytes_for_level_multiplier. So L2 can contain 256 * 10 = 2GB of data, L3 can contain 256 * 10 ** 2 = 20GB of data, and so on.

subcompactions — Compactions that occur for L0 → Ln files can’t run in parallel. The only way to speed them up is to partition the data into appropriate key ranges and then run sub-compactions on each of those partitions. By default, it is 1, which means the sub-compactions are disabled. You can refer to the official wiki for more scenarios in which sub-compactions can help.

disable_auto_compactions — Disables the compactions altogether. You should use it for Bulk loads when a lot of data is written to RocksDB. Post the bulk load, and you can trigger manual compaction, so the number of files gets reduced.

Use the Fastest Route!

RocksDB uses an LSM tree for managing data. LSM trees are quite good for write-heavy workloads but struggle under read-heavy workloads. The DB has to go through a lot of files to get a particular key. Besides enabling Cache, RocksDB offers multiple ways to improve read speeds by reducing lookups. Here’s those methods:

Bloom filters — RocksDB allows SSTable level bloom filters so you can skip reading indexes if a key is not in a particular file. You can set the number of bits to control the accuracy of a bloom filter.

whole_key_filtering — If you do a lot of point lookups, generally, using this setting can give you better results. All the filters, such as the bloom filter, will use all the bytes of the key rather than only a few prefix bytes. This option only affects the SSTables.

Hash index — RocksDB uses a binary search index by default to look up blocks and then search for keys inside a block. You can, however, switch to the hash index. This allows for faster lookups but uses more space. You can use the following configs to enable and tune the hash index.

• data_block_index_type — kDataBlockBinaryAndHash. This adds a hash index at the end of the data block. Without it, only binary search will be used to find a key inside a data block.
• data_block_hash_table_util_ratio — This is the load_factor for this hashtable, i.e., (number_of_entries / number_of_buckets). Setting it to less than 1 means there are more buckets than the number of entries reducing the number of reads for point lookups. Setting it to > 1 means each bucket can contain multiple elements, but the space used by the hash index will be lower.
• kHashSearch — This modifies the Index block to use hash search instead of binary search to lookup data blocks. This differs from the above config as it is used to find data blocks in a file. The kDataBlockBinaryAndHash is used to find keys/restart intervals inside a data block.

memtable_prefix_bloom_size_ratio — By default, the bloom filters are created at file levels. If you enable this option, a bloom filter is also created in the memtable for queries that go beyond the memtable to disk. The config determines the ratio of the memtable (i.e., write_buffer_size) that should be used for this bloom filter. Max value allowed is 0.25, i.e., 25%.

memtable_whole_key_filtering — If set to true, the bloom filter created using the previous config will contain complete key bytes instead of just prefixes. This is useful for point lookups. This option is different from whole_key_filtering, which affects only the SSTable files.

prefix_extractor — You can also tell RocksDB to use only the first few bytes of your keys in the indices instead of using the whole key. This helps in reducing space used for indices and filters. It also optimizes range queries. The most commonly used one is the FixedLen prefix extractor, where you can specify the number of bits it should use from the key.

Conclusion

The config options mentioned here are not exhaustive. You can scourge through the official wiki OR lookup comments in one of the options C++ file linked at the bottom. RocksDB offers certain helper methods, which you should use to set good enough defaults before you tune the options yourself. Examples of such methods include the following:

• options.OptimizeForPointLookup — To improve read latencies
• options.PrepareForBulkLoad — To reduce compaction and rate-limiting while loading a lot of data in the DB
• options.OptimizeLevelStyleCompaction — To improve compaction speeds and minimize the total number of compactions

Also, don’t forget to set your table format version to the latest. By default, it is 4 for backward compatibility, but the latest version is 5. By making this change, you’ll be able to take advantage of all the SSTable improvements in the latest RocksDB releases.

You can use the following references to learn more about RocksDB:

• Options C++ Code which contains most of the preset options
• Memtable Wiki
• Block Cache Wiki

Want to Connect?

You can find me on LinkedIn, Twitter, or send an email to 
kharekartik@gmail.com to share feedback.

Navigating the Minefield of RocksDB Configuration Options was originally published in Better Programming on Medium, where people are continuing the conversation by highlighting and responding to this story.

Before diving into this article, I should state that — as a developer — any situation requiring placing a language A project into a language B package should occur very rarely. Most of the time it’s preferable to consider re-designing the interaction between various language components in these situations. But what if this situation is unavoidable? Open source projects such as Apache Flink and Apache Spark serve as examples. These projects have been written completely in Java but also have python modules available for those who don’t want to use the Java API.

However, Flink and Spark depend on Apache Hadoop so they don’t have a choice to write their code in a language other than Java. Also, the architecture is such that java code can’t be deployed in the cloud as an API and the python module can simply call the API as is the case with most database drivers.

Structure of the project

We must first modify the structure of the project so that the python module can easily discover the java modules. To do this we will keep the project structure as follows

In the python project, we’ll create two files

• MANIFEST.in
• setup.py

The MANIFEST.in file contains the directories which need to be packaged in the python tar file.

The setup.py file contains the code to create the package and discover the java modules at runtime.

Create a test code

First, let’s create a java class Main.classthat just prints the class name in the stdout. We will package this class in the jar file dummy-java-module-1.0-SNAPSHOT.jar

Next, we’ll implement the file dummy.py which just executes this jar using the standard java command java -cp /path/to/dummy-java-module-1.0-SNAPSHOT.jar org.example.Main

We will put the dummy.py file in a directory called java_integration_lib . Now let’s take a look at how to discover the path of the jar file at runtime. We’ll be implementing thefind_modules code used in this step.

Discover Java Jar path

This is the most critical step of this process. As a developer, you can’t hardcode the path of the Jar in the dummy.py file. The path needs to be discovered at runtime.

If you are running the jar from the source directory, you can simply use the path of the dummy-java-module for running the jar.

But if the module is being run after being imported, then you need to use the path of the packaged jar. In the package, we will put the jar in /jars the directory. The code to do this is provided in the next section.

We will create a new file find_modules.py . In this file, we will implement the following code

The code does two things -

1. Check if the find_modules.py is in the source directory. If yes, simply use the path of the jar in thetarget folder.
2. If the find_modules.py is not in the source directory, it is being imported as a module. In that case, we get the jar from /jars the directory inside the module’s installation directory (which is generally /usr/lib/python3.7/site-packages/ )

We also check if the jar is actually present in these directories. If it is not, then the code throws an exception.

Package the code

Now the final step is to package the code in the tar.gz file which can be installed using pip . We first compile the java module into a jar file using the command

mvn clean package

Once done, change the directory to the dummy-python-module . We write the following code in the setup.py file.

The code can be divided into the following parts -

• Create a jars directory and copy the jar file in this directory
• Declare the packages and the corresponding files in those packages
  We declare a python package java_integration_lib.jars .
• Since this package doesn’t exist by default, we tell the setup module to create a directory jars for this package. Note that, this jars directory is different from the one created in the previous step.
• Then, we tell the setup to load only the *.jar files in this directory. The paths are interpreted as relative to the directory containing the package.
• Run python setup function.

For more details on how to package a simple module, you can look at the official documentation.

Now, in the MANIFEST.in the file you need to tell python that the jars the directory needs to be packaged in the python package.

You can do that by simply mentioning the command -

graft jars

Now simply run the following command

python setup.py sdist

This should create a directory dist inside the current directory and then add the python packagejava-integration-lib-1.0.0.tar.gz .

You can now install this python package using the command

pip install dist/java-integration-lib-1.0.0.tar.gz

Test the hybrid package

You can use the following code to test the package

from java_integration_lib import dummy

dummy.foo_bar()

If correctly installed you should get the output -

b'Java class org.example.Main executed!!\n'

The complete working demo code can be found in my repository — Thanks for reading!

KKcorps/python-java-hybrid-example

How to Package Java Projects in Python Tar files was originally published in codeburst on Medium, where people are continuing the conversation by highlighting and responding to this story.

Apache Pinot is a realtime distributed OLAP datastore that can answer hundreds of thousands of queries with millisecond latencies. You can head over to https://pinot.apache.org/ to get started with Apache Pinot.

While using any database, we can come across a scenario where a function required for the query is not supported out of the box. In such time, we have to resort to raising a pull request for a new function or finding a tedious workaround.

Pinot aims to solve this particular pain-point by giving users the power to add their functions with almost zero lines of code. The 0.5.0 release comes bundled with two such features —

• Support for Scalar Functions that allow users to write and add their functions as a plugin.
• Support for inline Apache Groovy scripts in SQL queries.

In this article, we’ll be focusing on Scalar functions.

What are Scalar Functions?

Scalar functions in Pinot are stateless functions which transform input A into output A. e.g. upper , lower , length etc. are scalar functions. However, sum , count are non-scalar functions.

You can take a look at the scalar functions supported out of the box in official repository.

Implement a scalar function

Let’s create a maven java project. We’ll call this project scalar-function-example.

Scalar functions require @ScalarFunction annotation. This annotation is present in pinot-spi package. You can add the package dependency in maven as follows —

<dependency>
  <groupId>org.apache.pinot</groupId>
  <artifactId>pinot-spi</artifactId>
  <version>0.11.0</version>
  <scope>provided</scope>
</dependency>

Let’s create a class CustomScalarFunctions that will contain all of our functions. Let’s write a java method that converts latitude and longitude to geohash encoding. Generally, Geohash encoders are not available out of the box in most of the database.

Now, you need to annotate this function with @ScalarFunction. The annotation supports the following arguments —

• name — The name of the function to be used while querying. e.g., the Java method name can be calculateLength , but the query name can be calculate_length_str. The default is the same as the method name.
• enable — Boolean value indicating whether this function should be registered or not for the usage. This can be used in particular scenarios where you may require to disable a function because of some error or duplicate methods.

Finally, the class should be present in the package name, which follows the following pattern — org.apache.pinot.*.function.*

This is because we find the custom methods using reflection, and currently, only the packages with the mentioned pattern are considered for the search. The final code should look as follows

Register the Scalar Function

Once you have written the code, just compile the code into a JAR file.

Next, if you haven’t downloaded Pinot, you can do that from our website.

Note that scalar functions are supported only in releases 0.5.0 and above.

Now, copy the JAR in theplugins or lib directory of the Pinot binary distribution and restart the Pinot cluster.

Use the Scalar Function

Now you can use the registered function as follows

All of the arguments can either be literal constants or column names. An exception will be thrown if the column’s data type doesn’t match the input data type.

RoadMap

We saw how easy it is to add your methods to Pinot for various use cases. The current implementation has few limitations —

• Only the following data types are supported in Input and output — Integer, Long, Float, Double, String, and Bytes. Objects such as Lists, Map are not supported currently.
• You can return only a single value.
• No support for multi-valued columns.
• Only Java methods are supported as of the moment.

We are working on removing most of these limitations in future releases.

Groovy Functions

Pinot also supports Apache Groovy script in SQL which you can use to write custom functions. It requires no additional code from the user. You can directly use the scripts in the query as follows —

Read our official documentation for more info on both Groovy and Scalar functions.

You can find the complete demo code in the pinot-examples repository.

If you liked this article, head over to https://pinot.apache.org and try it out today.

You can head over to our slack workspace for any queries or discussions.

Utilize UDFs to Supercharge Queries in Apache Pinot was originally published in Apache Pinot Developer Blog on Medium, where people are continuing the conversation by highlighting and responding to this story.

One of the primary advantages of using Pinot is its pluggable architecture. The plugins make it easy to add support for any third-party system which can be an execution framework, a filesystem, or input format.

In this tutorial, we will use three such plugins to easily ingest data and push it to our Pinot cluster. The plugins we will be using are -

• pinot-batch-ingestion-spark
• pinot-s3
• pinot-parquet

You can check out Batch Ingestion, File systems, and Input formats for all the available plugins.

Setup

We are using the following tools and frameworks for this tutorial -

• Apache Spark 2.2.3 (Although any spark 2.X should work)
• Apache Parquet 1.8.2
• Amazon S3
• Apache Pinot 0.4.0

Input Data

We need to get input data to ingest first. For our demo, we’ll just create some small Parquet files and upload them to our S3 bucket. The easiest way is to create CSV files and then convert them to Parquet. CSV makes it human-readable and thus easier to modify the input in case of some failure in our demo. We will call this file students.csv

Now, we’ll create Parquet files from the above CSV file using Spark. Since this is a small program, we will be using the Spark shell instead of writing a full-fledged Spark code.

The .parquet files can now be found in /path/to/batch_input directory. You can now upload this directory to S3 either using their UI or running the following command

aws s3 cp /path/to/batch_input s3://my-bucket/batch-input/ --recursive

Create Schema and Table

We need to create a table to query the data that will be ingested. All tables in Pinot are associated with a schema. You can check out Table configuration and Schema configuration for more details on creating configurations.

For our demo, we will have the following schema and table configs

We can now upload these configurations to Pinot and create an empty table. We will be using pinot-admin.sh CLI for this purpose.

pinot-admin.sh AddTable -tableConfigFile /path/to/student_table.json -schemaFile /path/to/student_schema.json -controllerHost localhost -controllerPort 9000 -exec

You can check out Command-Line Interface (CLI) for all the available commands.

Our table will now be available in the Pinot data explorer

Ingest Data

Now that our data is available in S3 as well as we have the Tables in Pinot, we can start the process of ingesting the data. Data ingestion in Pinot involves the following steps -

• Read data and generate compressed segment files from input
• Upload the compressed segment files to the output location
• Push the location of the segment files to the controller

Once the location is available to the controller, it can notify the servers to download the segment files and populate the tables.

The above steps can be performed using any distributed executor of your choice such as Hadoop, Spark, Flink, etc. For this demo, we will be using Apache Spark to execute the steps.

Pinot provides runners for Spark out of the box. So as a user, you don’t need to write a single line of code. You can write runners for any other executor using our provided interfaces.

First, we will create a job spec configuration file for our data ingestion process.

In the job spec, we have kept the execution framework as spark and configured the appropriate runners for each of our steps. We also need a temporary stagingDir for our spark job. This directory is cleaned up after our job has executed.

We also provide the S3 Filesystem and Parquet reader implementation in the config to use. You can refer Ingestion Job Spec for a complete list of configurations.

We can now run our Spark job to execute all the steps and populate data in Pinot.

In the command, we have included the JARs of all the required plugins in the Spark’s driver classpath. In practice, you only need to do this if you get a ClassNotFoundException.

Voila! Now our data is successfully ingested. Let’s try to query it from Pinot’s broker.

bin/pinot-admin.sh PostQuery -brokerHost localhost -brokerPort 8000 -queryType sql -query "SELECT * FROM students LIMIT 10"

If everything went right, you should receive the following output

You can also view the results in the Data explorer UI.

Pinot’s powerful pluggable architecture allowed us to successfully ingest parquet records from S3 with just a few configurations. The process described in this article is highly-scalable and can be used to ingest billions of records with minimal latency.

You can check out Pinot on the official website. Refer to our documentation to get started with the setup and in the case of any issues, the community is there to help on the official slack channel.

Leverage Plugins to Ingest Parquet Files from S3 In pinot was originally published in Apache Pinot Developer Blog on Medium, where people are continuing the conversation by highlighting and responding to this story.

The overflow of data in the world opened up a multitude of opportunities to learn and analyze the behavior of people all around the world. Most analyses require at least a few days of data if not more which results in a need of a fast queryable storage engine. OLAP databases exist to serve this purpose only i.e. to make huge amounts of data easily queryable with minimal latencies.

To minimize the latency, all databases have indices created on data. The index is generally a tree-based structure such as B-tree, R-Tree, etc. which based on some fixed key will directly provide you the row or the block containing the data rather than having you scan all the rows.

The unique issue with OLAP databases is that the queries can be done on multiple columns at once e.g. a query to get a total number of orders by users, date, and city. Technically, you can create indices on multiple dimensions as well but you will have to make assumptions on what query patterns the user will follow. You also have to assume the amount of data each column will hold to make an effective indexing decision.

What if the database itself adapted its index according to the data inserted? Learned Multi-dimensional Indexes are an effort to answer this particular problem. In this blog, we are going to take a look at one of those algorithms.

Flood

Flood algorithm has been designed for in-memory indexes. It can be modified to use in the OLTP databases as well. There are two key ideas behind Flood:

1. Use a sample query filter workload to determine how often certain dimensions are used, which ones are used together and which ones are more selective than others. Based on this info, customize the entire layout to optimize performance.
2. Use empirical CDF models to flatten multi-dimensional skewed data into a uniform space.

Let us assume an index needs to be created on d-dimensional data. In such data, there is no natural sort order. So, the algorithm first chooses a dimension that will be used for sorting. The algorithm then creates a d-1 dimensional grid where each dimension is divided into equally spaced columns. Each cell in such a grid contains multiple data points.

To ensure that data points are distributed uniformly, they are normalized using the min and max value of the particular dimension. The last dimension is used for sorting the data within each cell.

Query

A query generally consists of k dimensions where k < d. Since the query already contains the range of values for each dimension that need to be queried, we can simply select all the cells which lie within that range along with cells that have partial overlap. This step is known as Projection.

For the cells which have partial overlap, we can use the fact that the data in them is sorted and select the relevant data using binary search. This is only possible when you have the sort dimension as a part of the query. This step is known as Refinement.

Once we have all the data, we can refine it further to check if any out of range data is there or not and then return the result to the user. This step is known as Scan.

Layout optimization

The primary strength of FLOOD lies in its ability to optimize the data layout to minimize the query latency. To minimize the latency, you first need a proxy to determine the query performance. The algorithm uses a custom cost function to serve the purpose. The cost function is composed of three parts:

• wpNc, where wp is the average time to perform refinement on a cell and Nc, is the total number of cells in the grid.
• wrNc where wr is the average time to perform refinement on a cell and Nc is the total number of cells in the grid.
• wsNs, where ws is the average time to perform each scan and Ns, is the total number of scanned data points.

The model for query time then can be calculated as -

wpNc + wrNc + wsNs

The next step is to calculate the weights wp, wr, and ws. For this Flood uses a simple model that takes features such as the total number of cells, the mean, median and tail quantile of the sizes of filterable cells, the number of dimensions, etc. Flood only trains the weight model once and re-uses it for multiple data layouts.

The final step is to optimize the layout which consists of fiddling with the dimensions for the sort dimension and the number of columns in each dimension.

In each iteration, the algorithm chooses one of the d dimensions as the sort dimension. Rest all the dimensions are used to create the d-1 dimensional grid. Then, it runs a gradient descent algorithm to determine the number of columns that minimizes the query time.

Flood re-tunes the data layout for each new workload. Since it is not possible by DB administrator to generate most probable queries to provide as an input to the model, Flood itself generates queries for training by randomizing some dimensions for grouping, others for filtering and leaving the rest. The group by aggregate functions are also randomized.

Conclusion

Flood algorithm marks a significant milestone on a path of self-learning indices. It is highly practical and can be used in a real-world database as demonstrated in the paper. Flood however still suffers from drawbacks that need to be fixed to make it a completely generic algorithm. Some of those drawbacks are -

• It doesn’t support the insertion of new data. The whole dataset needs to be fitted again in case a new data arrives. The current version is meant to be used only for read-only workloads.
• It is single-threaded and can’t support concurrency in its current implementation.
• Although Flood re-tunes itself to a new workload, it still struggles to determine when has the workload changed enough for a retune operation to be triggered.

In the end, I am hopeful that these issues will be fixed in the future and OLAP databases will end up becoming at least an order of magnitude faster as demonstrated in the paper.

You can use to following references to learn more about Learned data structures and indices:

• Learning Multi-dimensional Indexes by Vikram Nathan, Jialin Ding, Mohammad Alizadeh, Tim Kraska
• The Case for Learned Index Structures by Tim Kraska, Alex Beutel, Ed H. Chi, Jeffrey Dean, Neoklis Polyzotis

Learning Multi-dimensional indices: The next big thing in OLAP DBs was originally published in TDS Archive on Medium, where people are continuing the conversation by highlighting and responding to this story.

The state of the world is gloomy and I wish it gets better as soon as possible. However, even in these tough times, there are a lot of things I am liking about WFH. I am fortunate enough in life to be able to do work from home in these tough times and this article in no way is meant to disrespect those who can’t.

The Lab

I currently work in a retail giant as a Software Engineer. My company already had all the remote working facilities such as Zoom, Slack, remote conferencing rooms, VPN, etc. I used to do WFH once a week or fortnight and the same is true for my team.

Due to the reasons mentioned above, I already had a minimal WFH setup. It consists of the following:

• LG 23 inch Full HD LCD monitor
• MacBook Pro 15 inch 2018 model
• A basic wooden table and Chair
• Sony WH-1000MX3 Noise-canceling headphones (The noise-canceling element is pretty useless these days)

All of this is set up in a separate room which is well sunlit. I used to work sitting in bed but that more often than not resulted in small unnecessary naps.

The Experiment

My sleep cycle has been completely disrupted these days. I am a latent night owl and the only reason I woke up early was to go to the office on time. Now, I usually sleep at around 2–3 AM and wake up at 10 AM because I make no compromise on 8 hours of sleep (except when deadlines are pretty strict or I am binge-watching a TV series). Strangely on weekend, I get up early around 7 because I need to buy the bread, eggs, and milk before they go out of stock.

After I am completely awake and done with shenanigans, I prepare a cup of coffee. I am a chai addict but I have switched to coffee because cleaning the tea leaves out of pan and sieve is not fun at all. I drink only half of what I have prepared so that I can have something to drink 2 hours later when I will again have the caffeine cravings.
I then have a small breakfast such as omelets or sandwiches. I scroll through Hacker News and Techcrunch for some latest news while sipping on my coffee.
Then, I proceed to clean my utensils which I left uncleaned last night due to lethargy.

By the time I have a bath, its around 11:30AM and time for my daily standup which lasts for about 30 min. Then, I sit, put on my headphones, start my Liked songs playlist on Spotify and proceed to work uninterrupted till 1:30PM.

I then proceed to prepare my lunch. I have only started learning cooking so I cook only basic stuff like Lentils, Rice, Chapati/Paratha, Rajma and 2–3 potato-based curries. The general life hack these days is to prepare enough so that I can eat it in dinner as well (not applicable on weekends though).

I am generally done with the lunch break in 1.5 hours so around 3–3:30 PM, I start working again till 5:30PM. If I have a bazillion JIRA tickets remaining, I prepare a cup of coffee and work till some of them are done. Otherwise, I generally proceed to watch some random videos on YouTube because it thinks I should watch those and shows them in my feed.

Then its time for some light 15–30 min. of exercise. I use one Android app called Fat to Fit (which likely won’t happen :P). It has a fixed routine for a month so I don’t have to decide for myself.

After I am done, with all the exercise, I generally go to my balcony to unwind for 15–20 min. and then switch on my PS4 for a couple of sessions of FIFA 2019 (need to buy 2020) since its fun and not at all exhausting.

At around 7PM, when I have finally rested its time to either do some research for my blogs or read some books. These days I am reading Lifespan by David A. Sinclair and Billion Dollar Whale by Bradley Hope and Tom Wright. I exclusively read non-fiction but balance heavy deep stuff with light and fun frequently as I have done in this case.

I then proceed to eat my dinner while watching news updates or lighthearted comedy TV series. Currently, I am watching Arrested Development on Hotstar. Since there is already so much doom and gloom around, I am avoiding dark movies completely although I certainly enjoy them more.

After dinner, if I am not exhausted, I either do some easy or medium questions on Leetcode or listen to a podcast and continue reading. I then proceed to shape up my drafts, adding one or two patches at a time or start watching some movie or anime depending on my mood. Finally, its time to sleep (in case something has not broken down in production).

The Observations

So as with all the activities in my life, there are both sides to the coin to the WFH situation. Let’s begin with what I like:

• I have become more efficient at getting my work done. This is primarily because I am not being interrupted for casual conversations or chai breaks in between.
• My time management skills have upgraded a notch due to the fact I have to cook and clean myself along with a plethora of stuff I was already doing. I use Notion to manage my daily life. Here’s what my dashboard looks like

• Not getting stuck in daily traffic and sucking all the smoke in the air while standing idle is definitely a big win.

Now, what do I dislike and miss? Well,

• Cleaning floors for sure.
• Struggle to buy groceries when all the delivery apps are assigning slots of 1 week and later.
• Connecting with teammates for random chitchat over tea
• Beer and Pizza

Another unanticipated outcome of the WFH situation is that I have been connecting with my friends a lot more than usual. I have been lucky enough in life to have amazing friends. These days we frequently hold zoom calls now that most of us are trapped within 4 walls for weeks.

As Ralph Waldo Emerson once said

It is one of the blessings of old friends that you can afford to be stupid with them

On that note, I hope the lockdown situation gets over soon and we can catch up over a glass of beer and a night of intensive board games soon. Till then the only objective is to grow my knowledge and to not grow my body.

Apache Zookeeper is probably one of the most amusing and intricate distributed frameworks. It is generally used as an intermediary to keep distributed servers in sync. Implementing sync ourselves can lead to multiple race conditions. So, Developers use Zookeeper in various systems without even thinking twice about any alternative. The wide adoption serves as a testament to its robustness and performance.

A question then pops up: How does Zookeeper manage to keep its servers in sync. You can’t use a parent zookeeper to manage a child zookeeper, thus creating inception like scenario without the limbo.

Hopefully, this article will try to answer the above query.

Leader and Followers

Zookeeper servers always have one leader which manages the complete data. All writes first happen on the leader and then are synced to followers. The leader helps in deciding authority over the correct state at a time and provides a single point of coordination.

Reads can happen from any of the servers, including the leader. It is entirely possible that some follower is not in sync with the leader, and you might end up reading an older state.

To keep Followers and Leaders in sync, Zookeeper uses the ZAB protocol.

ZAB

Zookeeper leader keeps all the followers in sync using its custom protocol called ZAB (Zookeeper Atomic Broadcast Protocol). It is a custom protocol designed for Zookeeper to solve challenges such as maintaining causal order among the data shared. All the state changes (known as transactions) in ZAB are idempotent and incremental. So applying a state change multiple times produces the same result. However, the order of changes needs to be maintained.
A transaction in ZAB is identified by a unique identifier called zxid which is a 64-bit integer composed of two 32-bit slices -

• Epoch - An integer that is incremented every time a leader election happens.
• Counter - An integer counter that is incremented after every valid transaction.

ZAB Protocol consists of three phases —

1. Discovery — determine who is the leader and how much data is missing

2. Synchronization — sync all the servers so that the missing data is updated

3. Broadcast — start transmitting transaction occurring in real-time

Before discussing each of these phases in detail, we need to understand what are the necessary guarantees that the protocol provides that make each of the above phases effective.

ZAB provides 3 necessary guarantees which are:

• Integrity — If a process receives a transaction with zxid Z than some other process has broadcast a transaction with zxid Z
• Total Order — If a process delivers transaction with zxid Z before one with zxid Z' , than any other process which delivers Z' must also deliver z, and it should deliver Z before Z'
• Agreement — If a process delivers z and another one delivers Z’ , then either the first process should deliver Z’ or the second process should deliver Z. This guarantees that the state of the two processes does not diverge.

These three safety properties guarantee that the final state of all the servers is consistent once all the transactions have been delivered to each other.

All the phases occurring in ZAB are quite similar to two-phased commits. First, a proposal is sent by the leader, the follower on the receiving end sends an ACK, and then a commit is sent by the leader to complete the transaction.

Discovery

In this phase, the leader and follower decide which server contains the true history of the transactions which have occurred until now.
A prospective leader is chosen first using a simple leader election algorithm.
The process contains the following exchanges:

1. Follower sends the last proposed epoch to the prospective leader.
2. The leader gets last accepted epoch from a quorum of followers and sends a new epoch, which is greater than all the epochs it has received.
3. If the new epoch is greater than last proposed epoch, followers update their proposed epoch and send their last acknowledged epoch along with their last zxid to leaders
4. The leader selects the history of the follower with the highest zxid and epoch as the truth.

Zookeeper performs an optimization where it selects the server with the highest epoch and zxid as the prospective leader so that it already has all the data which needs to be synced. The following piece of code in FastLeaderElection.java in Zookeeper checks the following-

/**
     * Check if a pair (server id, zxid) succeeds our
     * current vote.
     *
     * @param id    Server identifier
     * @param zxid  Last zxid observed by the issuer of this vote
     */
    protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
        LOG.debug("id: " + newId + ", proposed id: " + curId + ", zxid: 0x" +
                Long.toHexString(newZxid) + ", proposed zxid: 0x" + Long.toHexString(curZxid));
        if(self.getQuorumVerifier().getWeight(newId) == 0){
            return false;
        }

        /*
         * We return true if one of the following three cases hold:
         * 1- New epoch is higher
         * 2- New epoch is the same as current epoch, but new zxid is higher
         * 3- New epoch is the same as current epoch, new zxid is the same
         *  as current zxid, but server id is higher.
         */

        return ((newEpoch > curEpoch) ||
                ((newEpoch == curEpoch) &&
                ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
    }

Synchronisation

In this phase, the history of transactions is synced across all the followers.

1. The prospective leader proposes itself as the new leader since it has the highest zxid and epoch.
2. If the followers’ last accepted proposal has the same epoch as the new leader, it sets its current epoch as the same, sends ACK to the leader, and starts accepting all the missing transactions through a DIFF call.
3. Upon receiving the ACK from a quorum of followers, the leader sends a commit message and delivers all the missing transactions to the followers.

This synchronisation phase also prevents causal conflicts. It guarantees that all processes in the quorum deliver transactions of prior epochs before transactions of the new epoch e are proposed.

Broadcast

This is the phase that occurs after a quorum of servers has decided a leader, appended the missing data, and is ready to accept new transactions.

1. Leader proposes a transaction with zxid higher than all previous ids
2. Followers accept the proposed transaction from the leader and append it to their history. An ACK messaged is sent once the transaction is written to durable storage.
3. If the leader received ACK from a quorum of followers for the transaction, then it sends a commit message.
4. Followers on receiving a commit message broadcast the transactions among each other.

Each server in zookeeper executes one iteration of this protocol at a time. In case of an exception, such as epoch not matching with the leader, the servers can start a new iteration beginning from the first phase.

You can refer the following links to learn more about Zookeeper’s internals:

1. Zab: High-performance broadcast for primary-backup systems
2. ZooKeeper Internals
3. ZooKeeper: Distributed Process Coordination

How Does Zookeeper Servers Remain In sync? was originally published in codeburst on Medium, where people are continuing the conversation by highlighting and responding to this story.

Apache Airflow is an open-source scheduler to manage your regular jobs. It is an excellent tool to organize, execute, and monitor your workflows so that they work seamlessly.

Apache Airflow solved a lot of problems that the predecessors faced. Let’s first understand the architecture, and then we’ll take a look at what makes Airflow better.

DAGs

DAGs (Directed Acyclic Graphs) represent a workflow in Airflow. Each node in a DAG represents a task that needs to be run. The user mentions the frequency at which a particular DAG needs to be run. The user can also specify the trigger rule for each task in a DAG. e.g., You may want to trigger an alert task right after one of the previous tasks fails.

Let us try to understand the various components of Airflow.

Core components

Airflow primary consists of the following components -

1. Scheduler
2. Webserver
3. Executor
4. Backend

Scheduler

It is responsible for scheduling your tasks according to the frequency mentioned. It looks for all the eligible DAGs and then puts them in the queue. If a DAG failed and retry is enabled, the scheduler will automatically put that DAG up for retry. The number of retries can be limited on a DAG level.

Webserver

The webserver is the frontend for Airflow. Users can enable/disable, retry, and see logs for a DAG all from the UI. Users can also drill down in a DAG to see which tasks have failed, what caused the failure, how long did the task run for, and when was the task last retried.

This UI makes Airflow superior to its competitors. e.g., In Apache Oozie, seeing logs for non-MR (map-reduce) jobs is a pain.

Executor

It is responsible for actually running a task. Executor controls on which worker to run a task, how many tasks to run in parallel, and update the status of the task as it progress.

You can run your task on multiple workers managed by Celery or Dask or Kubernetes.

The tasks are pulled from a queue, which can be either Redis or RabbitMQ.

By default, Airflow uses SerialExecutor, which only runs one task at a time on a local machine. This is not advised to be done in production.

Backend

Airflow uses MySQL or PostgreSQL to store the configuration as well as the state of all the DAG and task runs. By default, Airflow uses SQLite as a backend by default, so no external setup is required. The SQLite backend is not recommended for production since data loss is probable.

So what makes Airflow the right scheduler for Data pipelines?

Monitoring

Airflow provides various methods of monitoring. You can see the status of the tasks from the UI. It sends an mail in case a DAG fails. You can also send the email if a task breaches the defined SLA. The logs for a task can also be viewed from the Airflow UI itself.

Lineage

This feature came pretty recently in Airflow v1.10. Lineage allows you to track the origins of data, what happens to it, and where it moves over time, such as Hive table or S3/HDFS partition.

It comes pretty handily when you multiple data tasks reading and writing into storage. The user needs to define the input and output data sources for each task, and a graph is created in Apache Atlas, depicting the relationship between various data sources.

Sensors

Sensors allow a user to trigger a task based on a certain pre-condition. The user needs to specify the type of sensor and the frequency at which they need to check for the condition. e.g., You can use the HDFS Partition sensor to trigger a task when a particular partition such as date is available.

Customization

Airflow also allows users to create their operators and sensors in case an already rich ecosystem of existing ones doesn’t satisfy your requirements. I wrote a SparkOperator because the official one didn’t allow me to tweak all the parameters. All the code is written in Python, which makes it easy for any developer to integrate.

Apart from all the benefits mentioned above, Airflow also has seamless integration with all the services in big data ecosystems such as Hadoop, Spark, etc. Since all the code is written in Python, getting started with Airflow will only take a couple of minutes. You can take a look at the official quickstart guide.

You can also explore https://databand.ai/ for a much more powerful setup to monitor your data pipelines powered by Apache Airflow.

Databand - Data Pipeline Observability | Observability for DAGs and ML

Why Apache Airflow Is a Great Choice for Managing Data Pipelines was originally published in TDS Archive on Medium, where people are continuing the conversation by highlighting and responding to this story.

Machine Learning has gone from zero to one in the past decade. The rise of ML can be seen as one of the most defining moments in the tech industry. Today ML models are ubiquitous in almost all the services.

One of the challenges which remain to date is the training and inference of models using real-time data. Let’s take a look at the various strategies which you can use in data streaming production jobs to make predictions.

Model alongside the pipeline

The natural approach for predictions on real-time data is to run your ML model in the pipeline processing the data.

This approach has two major complications -

1. Integration of the pipeline’s code and the model’s code.
2. Optimizing the integrated pipeline to make efficient use of the underlying resources.

Integration

Most of the real-time data pipelines are written using either Java or Python. Both Apache Spark and Apache Flink provide Python API. This allows for easy integration of models written using Scikit-Learn or Tensorflow.

You can also use Spark MLlib or Flink ML to create models. These models are convenient to integrate, and you don’t have to worry about scaling and fault-tolerance.

But what if you have a pre-existing data pipeline which is written in Java or Scala? In that case, it makes much more sense to use Tensorflow Java API or third-party libraries such as MLeap or JPMML to export your Scikit-learn models and use them inside your code. JPMML supports a lot of models but MLeap is faster.

Optimization

The choice between Python and Java/Scala represents a tradeoff between versatility and performance. It would be best if you made the decision based on the use case, the amount of expected data, and the latency expected. I prefer Scala for most of the applications since the expected input records were in the order of millions per second.

Another optimization is the number of parallel executors that you should allocate to your model. If it’s a lightweight model such as Logistic Regression or a small Random forest, you can even run a single instance of the model and re-partition data to go to the single executor (this is never a good idea in production). For heavy models such as large random forests or deep neural nets, finding the right number of executors is mostly an exercise in trial & error.

You might also need to optimize your ML models so that they can fit in the memory. There are several tools available for this purpose.

TensorFlow Lite | ML for Mobile and Edge Devices

Another complication with this approach is updating the model to a newer version. A fresh deployment is generally required for the update. This also makes A/B testing quite non-trivial.

Model as a REST service

This is one of the most popular approaches for inference. Run your python code inside a docker container and provide a REST interface to get the results. Tensorflow already provides the REST model serving out of the box.

For Java, you can use MLeap or DeepLearning4J. You can also dynamically increase/decrease the number of servers according to the throughput in this approach.

If your model calls are async, this approach fails to trigger back pressure in case there is a burst of data such as during restarts. This can lead to OOM failures in the model servers. Extra precautions must be taken to prevent such scenarios.

Latencies are also high since you need a network call to fetch the results. The latencies can slightly be reduced by using gRPC instead of REST.

Here’s How You Can Go Beyond Http 1.1

Database as a model store

If you have a fixed model architecture e.g., Linear Regression, Random Forest, or a small neural net, the weights can be stored in a distributed database such as Cassandra. You can create the model at the runtime using the weights and make the predictions on the new model.

This approach is a hybrid of the first and second approaches. It allows you to update the model at runtime without requiring a new deployment while also proving back pressure capabilities. It comes at the cost of versatility since you are limiting the number of potential options for the models.

So which approach should you choose?

Well, if you want to do a simple POC or your model is pretty lightweight, go with the REST model server. Ease of integration and very few code changes required to run your model makes it an attractive choice. A/B testing can also be done quickly.

If you require predictions to happen in tens of milliseconds, the pipelined approach is the one to prefer.

Lastly, the model store approach should only be used in the case when you have several models e.g., an ML model per city data, and they are lightweight too.

Deploying ML Models in Distributed Real-time Data Streaming Applications was originally published in TDS Archive on Medium, where people are continuing the conversation by highlighting and responding to this story.