The rebalance operation is used to recompute the assignment of brokers or servers in the cluster. This is not a single command, but rather a series of steps that need to be taken.
In the case of servers, rebalance operation is used to balance the distribution of the segments amongst the servers being used by a Pinot table. This is typically done after capacity changes or config changes such as replication or segment assignment strategies or table migration to a different tenant.
Below are changes that need to be followed by a rebalance.
Capacity changes
Increasing/decreasing replication for a table
Changing segment assignment for a table
Moving table from one tenant to a different tenant
These are typically done when downsizing/uplifting a cluster or replacing nodes of a cluster.
Every server added to the Pinot cluster, has tags associated with it. A group of servers with the same tag forms a Server Tenant.
By default, a server in the cluster gets added to the DefaultTenant i.e. gets tagged as DefaultTenant_OFFLINE and DefaultTenant_REALTIME.
Below is an example of how this looks in the znode, as seen in ZooInspector.
A Pinot table config has a tenants section, to define the tenant to be used by the table. The Pinot table will use all the servers which belong to the tenant as described in this config. For more details about this, see the Tenants section.
0.6.0 onwards
In order to change the server tags, the following API can be used.
PUT /instances/{instanceName}/updateTags?tags=<comma separated tags>
0.5.0 and prior
UpdateTags API is not available in 0.5.0 and prior. Instead, use this API to update the Instance.
PUT /instances/{instanceName}
For example,
NOTE
The output of GET and input of PUT don't match for this API. Please make sure to use the right payload as shown in example above. Particularly, notice that instance name "Server_host_port" gets split up into their own fields in this PUT API.
When upsizing/downsizing a cluster, you will need to make sure that the host names of servers are consistent. You can do this by setting the following config parameter:
In order to change the replication factor of a table, update the table config as follows:
OFFLINE table - update the replication field
REALTIME table - update the replicasPerPartition field
The most common segment assignment change is moving from the default segment assignment to replica group segment assignment. Discussing the details of the segment assignment is beyond the scope of this page. More details can be found in Routing and in this FAQ question.
In a scenario where you need to move table across tenants, for e.g table was assigned earlier to a different Pinot tenant and now you want to move it to a separate one, then you need to call the rebalance API with reassignInstances set to true.
Currently, two rebalance algorithms are supported; one is the default algorithm and the other one is minimal data movement algorithm.
This algorithm is used for most of the cases. When reassignInstances parameter is set to true, the final lists of instance assignment will be re-computed, and the list of instances is sorted per partition per replica group. Whenever the table rebalance is run, segment assignment will respect the sequence in the sorted list and pick up the relevant instances.
This algorithm focuses more on minimizing the data movement during table rebalance. When reassignInstances parameter is set to true and this algorithm gets enabled, the position of instances which are still alive remains the same, and vacant seats are filled with newly added instances or last instances in the existing alive instance candidate. So only the instances which change the position will involve in data movement.
In order to switch to this table rebalance algorithm, just simply set the following config to the table config before triggering table rebalance:
After any of the above described changes are done, a rebalance is needed to make those changes take effect.
To run a rebalance, use the following API.
POST /tables/{tableName}/rebalance?type=<OFFLINE/REALTIME>
This API has a lot of parameters to control its behavior. Make sure to go over them and change the defaults as needed.
Note
Typically, the flags that need to be changed from defaults are
includeConsuming=true for REALTIME
downtime=true if you have only 1 replica, or prefer faster rebalance at the cost of a momentary downtime
You can check the status of the rebalance by
Checking the controller logs
Running rebalance again after a while, you should receive status "status": "NO_OP"
Checking the External View of the table, to see the changes in capacity/replicas have taken effect.
This page describes how to rebalance a table
Rebalance operation is used to recompute assignment of brokers or servers in the cluster. This is not a single command, but more of a series of steps that need to be taken.
In case of servers, rebalance operation is used to balance the distribution of the segments amongst the servers being used by a Pinot table. This is typically done after capacity changes, or config changes such as replication or segment assignment strategies.
In case of brokers, rebalance operation is used to recalculate the broker assignment to the tables. This is typically done after capacity changes (scale up/down brokers).
| Query param | Default value | Description |
|---|---|---|
dryRun
false
If set to true, rebalance is run as a dry-run so that you can see the expected changes to the ideal state and instance partition assignment.
includeConsuming
false
Applicable for REALTIME tables. CONSUMING segments are rebalanced only if this is set to true. Moving a CONSUMING segment involves dropping the data consumed so far on old server, and re-consuming on the new server. If an application is sensitive to increased memory utilization due to re-consumption or to a momentary data staleness, they may choose to not include consuming in the rebalance. Whenever the CONSUMING segment completes, the completed segment will be assigned to the right instances, and the new CONSUMING segment will also be started on the correct instances. If you choose to includeConsuming=false and let the segments move later on, any downsized nodes need to remain untagged in the cluster, until the segment completion happens.
downtime
false
This controls whether Pinot allows downtime while rebalancing. If downtime = true, all replicas of a segment can be moved around in one go, which could result in a momentary downtime for that segment (time gap between ideal state updated to new servers and new servers downloading the segments). If downtime = false, Pinot will make sure to keep certain number of replicas (config in next row) always up. The rebalance will be done in multiple iterations under the hood, in order to fulfill this constraint.
Note: If you have only 1 replica for your table, rebalance with downtime=false is not possible.
minAvailableReplicas
1
Applicable for rebalance with downtime=false.
This is the minimum number of replicas that are expected to stay alive through the rebalance.
bestEfforts
false
Applicable for rebalance with downtime=false.
If a no-downtime rebalance cannot be performed successfully, this flag controls whether to fail the rebalance or do a best-effort rebalance.
reassignInstances
false
Applicable to tables where the instance assignment has been persisted to zookeeper. Setting this to true will make the rebalance first update the instance assignment, and then rebalance the segments.
bootstrap
false
Rebalances all segments again, as if adding segments to an empty table. If this is false, then the rebalance will try to minimize segment movements.
Rebalance operation is used to recompute assignment of brokers or servers in the cluster. This is not a single command, but more of a series of steps that need to be taken.
In case of brokers, rebalance operation is used to recalculate the broker assignment to the tables. This is typically done after capacity changes.
These are typically done when downsizing/uplifting a cluster, or replacing nodes of a cluster.
Every broker added to the Pinot cluster, has tags associated with it. A group of brokers with the same tag forms a Broker Tenant. By default, a broker in the cluster gets added to the DefaultTenant i.e. gets tagged as DefaultTenant_BROKER. Below is an example of how this tag looks in the znode, as seen in ZooInspector.
A Pinot table config has a tenants section, to define the tenant to be used by the table. More details about this in the Tenants section.
Using the tenant defined above, a mapping is created, from table name to brokers and stored in the IDEALSTATES/brokerResource. This mapping can be used by external services that need to pick a broker for querying.
If you want to scale up brokers, add new brokers to the cluster, and then tag them based on the tenant used by the table. If you're using DefaultTenant, no tagging needs to be done, as every broker node by default joins with tag DefaultTenant_BROKER.
If you want to scale down brokers, untag the brokers you wish to remove.
To update the tags on the broker, use the following API:
PUT /instances/{instanceName}/updateTags?tags=<comma separated tags>
Example for tagging the broker as per your custom tenant:
PUT /instances/Broker_10.20.151.8_8000/updateTags?tags=customTenant_BROKER
Example for untagging a broker:
PUT /instances/Broker_10.20.151.8_8000/updateTags?tags=untagged_BROKER
After making any capacity changes to the broker, the brokerResource needs to be rebuilt. This can be done with the below API:
POST /tables/{tableNameWithType}/rebuildBrokerResourceFromHelixTags
This is when you untagged and now want to remove the node from the cluster.
First, shutdown the broker. Then, use API below to remove the node from the cluster.
DELETE /instances/{instanceName}
If you encounter the below message when dropping, it means the broker process hasn't been shut down.
If you encounter below message, it means the broker has not been removed from the ideal state. Check the untagging and rebuild steps went through successfully.
