Stream Ingestion with Upsert
Upsert support in Apache Pinot.
Pinot provides native support of upsert during the real-time ingestion (v0.6.0+). There are scenarios that the records need modifications, such as correcting a ride fare and updating a delivery status.
With the foundation of full upsert support in Pinot, another category of use cases on partial upsert are enabled (v0.8.0+). Partial upsert is convenient to users so that they only need to specify the columns whose value changes, and ignore the others.
To enable upsert on a Pinot table, there are a couple of configurations to make on the table configurations as well as on the input stream.
To update a record, a primary key is needed to uniquely identify the record. To define a primary key, add the field
primaryKeyColumnsto the schema definition. For example, the schema definition of
UpsertMeetupRSVPin the quick start example has this definition.
Note this field expects a list of columns, as the primary key can be composite.
When two records of the same primary key are ingested, the record with the greater comparison value (timeColumn by default) is used. When records with the same primary key and event time, then the order is not determined. In most cases, the later ingested record will be used, but may not be so in the cases when the table has a column to sort by.
There are a few configurations needed in the table configurations to enable upsert.
The upsert mode defaults to
NONEfor realtime tables. To enable the full upsert, set the
FULLfor the full update. FULL upsert means that a new record will replace the older record completely if they have same primary key. Example config:
Partial upsert support is also added in
release-0.8.0. With this feature, users can choose to update only specific columns and ignore the rest.
To enable the partial upsert, set the
partialUpsertStrategiesfor partial upsert columns. Since
OVERWRITEis used as the default strategy for columns without a specified strategy.
defaultPartialUpsertStrategyis also introduced to change the default strategy for all columns. For example:
Pinot supports the following partial upsert strategies:
By default, Pinot uses the value in the time column (
timeColumnin tableConfig) to determine the latest record. That means, for two records with the same primary key, the record with the larger value of the time column is picked as the latest update. However, there are cases when users need to use another column to determine the order. In such case, you can use option
comparisonColumnto override the column used for comparison. For example,
For partial upsert table, the out-of-order events won't be consumed and indexed. For example, for two records with the same primary key, if the record with the smaller value of the comparison column came later than the other record, it will be skipped.
The upsert Pinot table can use only the low-level consumer for the input streams. As a result, it uses the partitioned replica-group assignment for the segments. Moreover, upsert poses the additional requirement that all segments of the same partition must be served from the same server to ensure the data consistency across the segments. Accordingly, it requires to use
strictReplicaGroupas the routing strategy. To use that, configure
Routingas the following:
Upsert snapshot support is also added in
release-0.12.0. To enable the snapshot, set the
true. For example:
Upsert maintains metadata in memory containing which docIds are valid in a particular segment (ValidDocIndexes). This metadata gets lost during server restarts and needs to be recreated again. ValidDocIndexes can not be recovered easily after out-of-TTL primary keys get removed. Enabling snapshots addresses this problem by adding functions to store and recover validDocIds snapshot for Immutable Segments We recommend that you enable this feature so as to speed up server boot times during restarts.
There are some limitations for the upsert Pinot tables.
- The high-level consumer is not allowed for the input stream ingestion, which means
stream.[consumerName].consumer.typemust always be
- The star-tree index cannot be used for indexing, as the star-tree index performs pre-aggregation during the ingestion.
- Unlike append-only tables, out-of-order events (with comparison value in incoming record less than the latest available value) won't be consumed and indexed by Pinot partial upsert table, these late events will be skipped.
Unlike other real-time tables, Upsert table takes up more memory resources as it needs to bookkeep the record locations in memory. As a result, it's important to plan the capacity beforehand, and monitor the resource usage. Here are some recommended practices of using Upsert table.
The number of partitions in input streams determines the partition numbers of the Pinot table. The more partitions you have in input topic/stream, more Pinot servers you can distribute the Pinot table to and therefore more you can scale the table horizontally. Do note that you can't increase the partitions in future for upsert enabled tables so you need to start with good enough partitions (atleast 2-3X the number of pinot servers)
Upsert table maintains an in-memory map from the primary key to the record location. So it's recommended to use a simple primary key type and avoid composite primary keys to save the memory cost. In addition, consider the
hashFunctionconfig in the Upsert config, which can be
MURMUR3, to store the 128-bit hashcode of the primary key instead. This is useful when your primary key takes more space. But keep in mind, this hash may introduce collisions, though the chance is very low.
Set up a dashboard over the metric
pinot.server.upsertPrimaryKeysCount.tableNameto watch the number of primary keys in a table partition. It's useful for tracking its growth which is proportional to the memory usage growth. The total memory usage by upsert is roughly
(primaryKeysCount * (sizeOfKeyInBytes + 24))
It's useful to plan the capacity beforehand to ensure you will not run into resource constraints later. A simple way is to measure the rate of the primary keys in the input stream per partition and extrapolate the data to a specific time period (based on table retention) to approximate the memory usage. A heap dump is also useful to check the memory usage so far on an upsert table instance.
Putting these together, you can find the table configurations of the quick start example as the following:
To illustrate how the full upsert works, the Pinot binary comes with a quick start example. Use the following command to creates a realtime upsert table
# stop previous quick start cluster, if any
You can also run partial upsert demo with the following command
# stop previous quick start cluster, if any
As soon as data flows into the stream, the Pinot table will consume it and it will be ready for querying. Head over to the Query Console to checkout the realtime data.
Query the upsert table
For partial upsert you can see only the value from configured column changed based on specified partial upsert strategy.
Query the partial upsert table
An example for partial upsert is shown below, each of the event_id kept being unique during ingestion, meanwhile the value of rsvp_count incremented.
Explain partial upsert table
To see the difference from the non-upsert table, you can use a query option
skipUpsertto skip the upsert effect in the query result.
Disable the upsert during query via query option
Can I change primary key columns in existing upsert table?
Yes, you can add or delete columns to primary keys as long as input stream is partitioned on one of the primary key columns. However, you need to restart all Pinot servers so that it can rebuild the primary key to record location map with the new columns.