Tuning Real-time Performance

See the section on Ingesting Real-time Data before reading this section.

Pinot servers ingest rows into a consuming segment that resides in volatile memory. Therefore, pinot servers hosting consuming segments tend to be memory bound. They may also have long garbage collection cycles when the segment is completed and memory is released.

Controlling memory allocation

You can configure Pinot servers to use off-heap memory for dictionary and forward indices of consuming segments by setting the value of pinot.server.instance.realtime.alloc.offheap to true. With this configuration in place, the server allocates off-heap memory by memory-mapping files. These files are never flushed to stable storage by Pinot (the Operating System may do so depending on demand for memory on the host). The files are discarded when the consuming segment is turned into a completed segment.

By default, the files are created under the directory where the table’s segments are stored in local disk attached to the consuming server. You can set a specific directory for consuming segments with the configuration pinot.server.consumerDir. Given that there is no control over flushing of pages from the memory mapped for consuming segments, you may want to set the directory to point to a memory-based file system, eliminating wasteful disk I/O.

If you don't want to use memory-mapping, set pinot.server.instance.realtime.alloc.offheap.direct to true. In this case, pinot allocates direct objects for consuming segments. Using direct allocation can potentially result in address space fragmentation.

Note that we still use heap memory to store inverted indices for consuming segments.

Controlling number of rows in consuming segment

The number of rows in a consuming segment needs to be balanced. Having too many rows can result in memory pressure. On the other hand, having too few rows results in having too many small segments. Having too many segments can be detrimental to query performance, and also increase pressure on the Helix.

The recommended way to do this is to use the realtime.segment.flush.threshold.segment.size setting as described in . You can run the administrative tool pinot-admin.sh RealtimeProvisioningHelper that will help you to come up with an optimal setting for the segment size.

Moving completed segments to different hosts

This feature is available only if the consumption type is LowLevel.

The structure of the consuming segments and the completed segments are very different. The memory, CPU, I/O and GC characteristics could be very different while processing queries on these segments. Therefore it may be useful to move the completed segments onto different set of hosts in some use cases.

You can host completed segments on a different set of hosts using the tagOverrideConfig as described in . Pinot will automatically move them once the consuming segments are completed.

If you require more fine-tuned control over how segments are hosted on different hosts, we recommend that you use the feature to accomplish this.

Controlling segment build vs. segment download on real-time servers

This feature is available only if the consumption type is LowLevel.

When a real-time segment completes, a winner server is chosen as a committer amongst all replicas by the controller. That committer builds the segment and uploads to the controller. The non-committer servers are asked to catchup to the winning offset. If the non-committer servers are able to catch up, they are asked to build the segment and replace the in-memory segment. If they are unable to catchup, they are asked to download the segment from the controller.

Building a segment can cause excessive garbage and may result in GC pauses on the server. Long GC pauses can affect query processing. You might want to force the non-committer servers to download the segment from the controller instead of building it again. The completionConfig as described in can be used to configure this.

Fine tuning the segment commit protocol

This feature is available only if the consumption type is LowLevel.

Once a committer is asked to commit the segment, it builds a segment, and issues an HTTP POST to the controller, with the segment. The controller then commits the segment in Zookeeper and starts the next consuming segment.

It is possible to configure the servers to do a split commit, in which the committer performs the following steps:

• Build the segment

• Start a transaction with the lead controller to commit the segment (CommitStart phase)

• Post the completed segment to any of the controllers (and the controller posts it to segment store)

This method of committing can be useful if the network bandwidth on the lead controller is limiting segment uploads.In order to accomplish this, you will need to set the following configurations:

• On the controller, set pinot.controller.enable.split.commit to true (default is false).

• On the server, set pinot.server.enable.split.commit to true (default is false).

RealtimeProvisioningHelper

This tool can help decide the optimum segment size and number of hosts for your table. You will need one sample Pinot segment from your table before you run this command. There are three ways to get a sample segment:

1. If you have an offline segment, you can use that.

2. You can provision a test version of your table with some minimum number of hosts that can consume the stream, let it create a few segments with large enough number of rows (say, 500k to 1M rows), and use one of those segments to run the command. You can drop the test version table, and re-provision it once the command outputs some parameters to set.

3. If you don't have a segment in hand or provisioning of a test version of your table is not an easy option, you can provide . Then the tool generates a segment based on the provided characteristics behind the scene and proceeds with the real-time analysis. In case the characteristics of real data is very different, you may need to modify the parameters. You can always change the config after you get segments from real data.

As of Pinot version 0.5.0, this command has been improved to display the number of pages mapped, as well as take in the push frequency as an argument if the real-time table being provisioned is a part of a hybrid table. If you are using an older version of this command, download a later version and re-run the command. The arguments to the command are as follows:

• tableConfigFile: This is the path to the table config file

• numPartitions: Number of partitions in your stream

• numHosts

One you run the command, it produces an output as below:

The idea here is to choose an optimal segment size so that :

1. The number of segments searched for your queries are minimized

2. The segment size is neither too large not too small (where "large" and "small" are as per the range for your table).

3. Overall memory is optimized for your table, considering the other tables in the host, the query traffic, etc.

You can pick the appropriate value for segment size and number of hours in the table config, and set the number of rows to zero. Note that you don't have to pick values exactly as given in each of these combinations (they are calculated guesses anyway). Feel free to choose some values in between or out of range as you feel fit, and adjust them after your table is in production (no restarts required, things will slowly adjust themselves to the new configuration). The example given below chooses from the output.

Case 1: Optimize for performance, high QPS

From the above output you may decide that 6 hours is an optimal consumption time given the number of active segments looked at for a query, and you can afford about 4G of active memory per host. You can choose either 8 or 10 hosts, you choose 10. In this case, the optimal segment size will be 111.98M. You can then enter your real-time table config as below:

Case 2: Optimize for cost, low QPS

You may decide from the output that you want to make do with 6 hosts. You have only 2G of memory per host for active segments but you are willing to map 8G of active memory on that, with plenty of paging for each query. Since QPS is low, you may have plenty of CPU per query so huge segments may not be a problem. Choose 12 or 24h or consumption and pick an appropriate segment size. You may then configure something like:

Pinot supports various different query scheduling mechanisms to allow for more control around which queries have priority when executing on the server instances.

Currently, there are the following options that can be configured using the pinot.query.scheduler.name configuration:

• First Come First Serve (fcfs) (Default)

Tuning Pinot

This section provides information on various options to tune Pinot cluster for storage and query efficiency. Unlike key-value store, tuning Pinot sometimes can be tricky because the cost of query can vary depending on the workload and data characteristics.

If you want to improve query latency for your use case, you can refer to Index Techniques section. If your use case faces the scalability issue after tuning index, you can refer Optimizing Scatter and Gather for improving query throughput for Pinot cluster. If you have identified a performance issue on the specific component (broker or server), you can refer to the

Adaptive Server Selection is a new routing capability for Pinot Brokers where incoming queries are routed to the best available server instead of following the default round robin approach while choosing servers. With this feature, Brokers will be sensitive to changes on the Servers like GC issues, slowness, network slowness, etc. The broker will thus adaptively route more queries to faster servers and lesser queries to slower servers

How this works

There are two main components:

1. Stats Collection

2. Routing using Adaptive Server Selection

Stats Collection

Each broker maintains stats individually for all servers. These stats are collected at the broker during query processing when the query is routed to the servers and after the response is received from the servers. These stats are maintained in-memory. Some of the stats collected at broker per server are as follows:

1. Number of in-progress / in-flight queries

2. EWMA (Exponential Weighted Moving Average) for latencies seen by queries

3. EWMA (Exponential Weighted Moving Average) for number of ongoing queries at any time

Adaptive Routing

When the broker receives a query, it will use the above stats to pick the best available server. This enables the broker to automatically reduces the number of queries it sends to slow servers and increase the number of queries it sends to faster servers. We currently support the following strategies:

1. NO_OP : Uses the default RoundRobin approach. In other words, this will give existing behavior where stats are not used by broker when picking the servers to route the query to.

2. NUM_INFLIGHT_REQ : Uses the number of in-flight requests stat to determine the best server

3. LATENCY : Uses the EWMA latency stat to determine the best server

The above strategies works in tandem with the following available Routing mechanisms today:

1. Balanced Routing

2. ReplicaGroup Routing

So, a table can be configured to use Balanced or Replica group segment assignment + routing and can still leverage the adaptive server selection feature.

Configs

The configuration for enabling/disabling this feature and the knobs for performance tuning are present at the Broker instance level. The feature is currently turned off by default.

Enabling Stats Collection and Adaptive Routing

1. To enable Stats Collection, set pinot.broker.adaptive.server.selector.enable.stats.collection = true. Note that setting this property alone will only enable stats collection and not perform Adaptive Routing

2. To enable an Adaptive Routing Strategy, use one of the following configs. The HYBRID strategy works well for most use cases. Unless you are an advanced user, we recommend using the HYBRID strategy.

Tuning Knobs

The following configs are already set to default values that work well for most usecases. For advanced users, the following knobs are available to tune Adaptive Routing Strategies

Optimizing routing

As explained in , Apache Pinot is a distributed system where different components have specific roles. Queries arrive at one of the , that calculates which are going to participate in the query. In most clusters, a single server cannot hold all the segments, so each server has a partial view of the complete data. At the same time, segments are usually replicated on different servers in order to have high availability and better performance. In order to produce a complete result, the broker needs to calculate a subset of servers that contains all the segments that are required to resolve the query. Once this subset is calculated, the router uses a scatter and gather algorithm that sends the query to each server and then merges the partial results into the complete one.