Ingesting Offline data

Segments for offline tables are constructed outside of Pinot, typically in Hadoop via map-reduce jobs and ingested into Pinot via REST API provided by the Controller. Pinot provides libraries to create Pinot segments out of input files in AVRO, JSON or CSV formats in a hadoop job, and push the constructed segments to the controllers via REST APIs.

When an Offline segment is ingested, the controller looks up the table’s configuration and assigns the segment to the servers that host the table. It may assign multiple servers for each segment depending on the number of replicas configured for that table.

Pinot supports different segment assignment strategies that are optimized for various use cases.

Once segments are assigned, Pinot servers get notified via Helix to “host” the segment. The segments are downloaded from the remote segment store to the local storage, untarred, and memory-mapped.

Once the server has loaded (memory-mapped) the segment, Helix notifies brokers of the availability of these segments. The brokers start to include the new segments for queries. Brokers support different routing strategies depending on the type of table, the segment assignment strategy, and the use case.

Data in offline segments are immutable (Rows cannot be added, deleted, or modified). However, segments may be replaced with modified data.

Starting from release-0.11.0, Pinot supports uploading offline segments to real-time tables. This is useful when user wants to bootstrap a real-time table with some initial data, or add some offline data to a real-time table without changing the data stream. Note that this is different from the setup, and no time boundary is maintained between the offline segments and the real-time segments.

Ingesting Real-time Data

Segments for real-time tables are constructed by Pinot servers with rows ingested from data streams such as Kafka. Rows ingested from streams are made available for query processing as soon as they are ingested, thus enabling applications such as those that need real-time charts on analytics.

In large scale installations, data in streams is typically split across multiple stream partitions. The underlying stream may provide consumer implementations that allow applications to consume data from any subset of partitions, including all partitions (or, just from one partition).

A pinot table can be configured to consume from streams in one of two modes:

• LowLevel: This is the preferred mode of consumption. Pinot creates independent partition-level consumers for each partition. Depending on the the configured number of replicas, multiple consumers may be created for each partition, taking care that no two replicas exist on the same server host. Therefore you need to provision at least as many hosts as the number of replicas configured.

• HighLevel: Pinot creates one stream-level consumer that consumes from all partitions. Each message consumed could be from any of the partitions of the stream. Depending on the configured number of replicas, multiple stream-level consumers are created, taking care that no two replicas exist on the same server host. Therefore you need to provision exactly as many hosts as the number of replicas configured.

Of course, the underlying stream should support either mode of consumption in order for a Pinot table to use that mode. Kafka has support for both of these modes. See for more information on the support of other data streams in Pinot.

In either mode, Pinot servers store the ingested rows in volatile memory until either one of the following conditions are met:

1. A certain number of rows are consumed

2. The consumption has gone on for a certain length of time

(See on how to set these values, or have pinot compute them for you)

Upon reaching either one of these limits, the servers do the following:

• Pause consumption

• Persist the rows consumed so far into non-volatile storage

• Continue consuming new rows into volatile memory again.

The persisted rows form what we call a completed segment (as opposed to a consuming segment that resides in volatile memory).

In LowLevel mode, the completed segments are persisted into the local non-volatile store of pinot server as well as the segment store of the pinot cluster (See ). This allows for easy and automated mechanisms for replacing pinot servers, or expanding capacity, etc. Pinot has that ensure that the completed segment is equivalent across all replicas.

During segment completion, one winner is chosen by the controller from all the replicas as the committer server. The committer server builds the segment and uploads it to the controller. All the other non-committer servers follow one of these two paths:

1. If the in-memory segment is equivalent to the committed segment, the non-committer server also builds the segment locally and replaces the in-memory segment

2. If the in-memory segment is non equivalent to the committed segment, the non-committer server downloads the segment from the controller.

For more details on this protocol, refer to .

In HighLevel mode, the servers persist the consumed rows into local store (and not the segment store). Since consumption of rows can be from any partition, it is not possible to guarantee equivalence of segments across replicas.

See for details.

To query using distributed joins, window functions, and other multi-stage operators in real time, you must enable the multi-stage query engine (v2). To enable v2, do any of the following:

• Enable the multi-stage query engine in the Query Console

• Programmatically access the multi-stage query engine:

  • Query

  • Query outside of the APIs

To learn more about what the multi-stage query engine is, see .

Enable the multi-stage query engine in the Query Console

• To enable the multi-stage query engine, in the Pinot Query Console, select the Use Multi-Stage Engine check box.

Programmatically access the multi-stage query engine

To query the Pinot multi-stage query engine, use REST APIs or the query option:

Use REST APIs

The Controller admin API and the Broker query API allow optional JSON payload for configuration. For example:

Use the query option

To enable the multi-stage engine via a query outside of the API, add the useMultistageEngine=true option to the top of your query.

For example:

Many data analytics use-cases only need aggregated data. For example, data used in charts can be aggregated down to one row per time bucket per dimension combination.

Doing this results in much less storage and better query performance. Configuring this for a table is done via the Aggregation Config in the table config.

Aggregation Config

The aggregation config controls the aggregations that happen during real-time data ingestion. Offline aggregations must be handled separately.

Below is a description of the config, which is defined in the ingestion config of the table config.

Requirements

The following are required for ingestion aggregation to work:

• Ingestion aggregation config is effective only for real-time tables. (There is no ingestion time aggregation support for offline tables. We need use or pre-process aggregations in the offline data flow using batch processing engines like Spark/MapReduce).

• type must be lowLevel.

• All metrics must have aggregation configs.

Example Scenario

Here is an example of sales data, where only the daily sales aggregates per product are needed.

Example Input Data

Schema

Note that the schema only reflects the final table structure.

Table Config

From the below aggregation config example, note that price exists in the input data while total_sales exists in the Pinot Schema.

Example Final Table

Allowed Aggregation Functions

Frequently Asked Questions

Why not use a Startree?

Startrees can only be added to real-time segments after the segments has sealed, and creating startrees is CPU-intensive. Ingestion Aggregation works for consuming segments and uses no additional CPU.

Startrees take additional memory to store, while ingestion aggregation stores less data than the original dataset.

When to not use ingestion aggregation?

If the original rows in non-aggregated form are needed, then ingestion-aggregation cannot be used.

I already use the aggregateMetrics setting?

The aggregateMetrics works the same as Ingestion Aggregation, but only allows for the SUM function.

The current changes are backward compatible, so no need to change your table config unless you need a different aggregation function.

Does this config work for offline data?

Ingestion Aggregation only works for real-time ingestion. For offline data, the offline process needs to generate the aggregates separately.

Why do all metrics need to be aggregated?

If a metric isn't aggregated then it will result in more than one row per unique set of dimensions.

For performance reasons, null handling support is disabled by default in Apache Pinot. When null support is disabled, all columns are treated as not null. Predicates like IS NOT NULL evaluates to true, and IS NULL evaluates to false. Aggregation functions like COUNT, SUM, AVG, MODE, etc. treat all columns as not null.

For example, the predicate in the query below matches all records.

To handle null values in your data, you must configure your tables to . This has to be done before ingesting the data. Tables where null values are stored support basic null handling, which is limited to IS NULL and IS NOT NULL predicates, and can optionally be queried with advanced null handling support.

The following table summarizes the behavior of null handling support in Pinot:

Default behavior

Pinot always stores column values in a . Forward index never stores null values but have to store a value for each row. Therefore independent of the null handling configuration, Pinot always stores a default value for nulls rows in the forward index. The default value used in a column can be specified in the configuration by setting the defaultNullValue field spec. The defaultNullValue depends on the type of data.

Store nulls at ingestion time

To support null handling, Pinot must store null values in segments. The forward index stores the default value for null rows whether null storing is enabled or not. When null storing is enabled, Pinot creates a new index called the null index or null vector index. This index stores the document IDs of the rows that have null values for the column.

Null support is configured per table. You can configure one table to store nulls, and configure another table to not store nulls. There are two ways to define null storing support in Pinot:

1. , where each column in a table is configured as nullable or not nullable. We recommend enabling null storing support by column. This is the only way to support null handling in the .

2. , where all columns in the table are considered nullable. This is how null values were handled before Pinot 1.1.0 and now deprecated.

Column based null storing

We recommend configuring column based null storing, which lets you specify null handling per column and supports null handling in the multi-stage query engine.

To enable column based null handling:

1. Set to true in the schema configuration before ingesting data.

2. Then specify which columns are not nullable using the notNull field spec, which defaults to false.

Table based null storing

This is the only way to enable null storing in Pinot before 1.1.0, but it is deprecated since then. Table based null storing is more expensive in terms of disk space and query performance than column based null storing. Also, it is not possible to support null handling in multi-stage query engine using table based null storing.

To enable table based null storing, enable the nullHandlingEnabled configuration in before ingesting data. All columns in the table are now nullable.

As an example:

Null handling at query time

To enable basic null handling by at query time, enable Pinot to . Advanced null handling support can be optionally enabled.

If you are converting from null support for the single-stage query engine, you can simplify your model by removing nullHandlingEnabled at the same time you set enableColumnBasedNullHandling. Also, when converting:

• No reingestion is needed.

• If the columns are changed from nullable to not nullable and there is a value that was previously null, the default value will be used instead.

Basic null support

The basic null support is automatically enabled when null values are stored on a segment (see ).

In this mode, Pinot is able to handle simple predicates like IS NULL or IS NOT NULL. Other transformation functions (like CASE, COALESCE, +, etc.) and aggregations functions (like COUNT, SUM, AVG, etc.) will use the default value specified in the schema for null values.

For example, in the following table:

If the default value for col1 is 1, the following query:

Will return the following result:

While

While return the following:

And queries like

Will return

Also

Given that neither count or mode function will ignore null values as expected but read instead the default value (in this case 1) stored in the forward index.

Advanced null handling support

Advanced null handling has two requirements:

1. Segments must store null values (see ).

2. The query must enable null handling by setting the enableNullHandling to true.

The later can be done in one of the following ways:

• Set enableNullHandling=true at the beginning of the query.

• If using JDBC, set the connection option enableNullHandling=true (either in the URL or as a property).

When the enableNullHandling option is set to true, the Pinot query engine uses a different execution path that interprets nulls in a standard SQL way. This means that IS NULL and IS NOT NULL predicates will evaluate to true or false according to whether a null is detected (like in basic null support mode) but also aggregation functions like COUNT, SUM, AVG, MODE, etc. will deal with null values as expected (usually ignoring null values).

In this mode, some indexes may not be usable, and queries may be significantly more expensive. Performance degradation impacts all the columns in the table, including columns in the query that do not contain null values. This degradation happens even when table uses column base null storing.

Examples queries

Select Query

Filter Query

Aggregate Query

Aggregate Filter Query

Group By Query

Order By Query

Transform Query

Appendix: Workarounds to handle null values without storing nulls

If you're not able to generate the null index for your use case, you may filter for null values using a default value specified in your schema or a specific value included in your query.

Filter for default null value(s) specified in your schema

1. Specify a default null value (defaultNullValue) in your for dimension fields, (dimensionFieldSpecs), metric fields (metricFieldSpecs), and date time fields (dateTimeFieldSpecs).

2. Ingest the data.

Filter for a specific value in your query

Filter for a specific value in your query that will not be included in the dataset. For example, to calculate the average age, use -1 to indicate the value of Age is null.

• Rewrite the following query:

• To cover null values as follows:

Raw source data often needs to undergo some transformations before it is pushed to Pinot.

Transformations include extracting records from nested objects, applying simple transform functions on certain columns, filtering out unwanted columns, as well as more advanced operations like joining between datasets.

A preprocessing job is usually needed to perform these operations. In streaming data sources, you might write a Samza job and create an intermediate topic to store the transformed data.

For simple transformations, this can result in inconsistencies in the batch/stream data source and increase maintenance and operator overhead.

To make things easier, Pinot supports transformations that can be applied via the table config.

Transformation functions

Pinot supports the following functions:

• Groovy functions

• Built-in functions

Groovy functions

Groovy functions can be defined using the syntax:

Any valid Groovy expression can be used.

⚠️ Enabling Groovy

Allowing executable Groovy in ingestion transformation can be a security vulnerability. To enable Groovy for ingestion, set the following controller configuration:

controller.disable.ingestion.groovy=false

If not set, Groovy for ingestion transformation is disabled by default.

Built-in Pinot functions

All the functions defined in annotated with @ScalarFunction (for example, ) are supported ingestion transformation functions.

Below are some commonly used built-in Pinot functions for ingestion transformations.

DateTime functions

These functions enable time transformations.

toEpochXXX

Converts from epoch milliseconds to a higher granularity.

toEpochXXXRounded

Converts from epoch milliseconds to another granularity, rounding to the nearest rounding bucket. For example, 1588469352000 (2020-05-01 42:29:12) is 26474489 minutesSinceEpoch. `toEpochMinutesRounded(1588469352000) = 26474480 (2020-05-01 42:20:00)

fromEpochXXX

Converts from an epoch granularity to milliseconds.

Simple date format

Converts simple date format strings to milliseconds and vice versa, per the provided pattern string.

JSON functions

Types of transformation

Filtering

Records can be filtered as they are ingested. A filter function can be specified in the filterConfigs in the ingestionConfigs of the table config.

If the expression evaluates to true, the record will be filtered out. The expressions can use any of the transform functions described in the previous section.

Consider a table that has a column timestamp. If you want to filter out records that are older than timestamp 1589007600000, you could apply the following function:

Consider a table that has a string column campaign and a multi-value column double column prices. If you want to filter out records where campaign = 'X' or 'Y' and sum of all elements in prices is less than 100, you could apply the following function:

Filter config also supports SQL-like expression of built-in for filtering records (starting v 0.11.0+). Example:

Column transformation

Transform functions can be defined on columns in the ingestion config of the table config.

For example, imagine that our source data contains the prices and timestamp fields. We want to extract the maximum price and store that in the maxPrices field and convert the timestamp into the number of hours since the epoch and store it in the hoursSinceEpoch field. You can do this by applying the following transformation:

Below are some examples of commonly used functions.

String concatenation

Concat firstName and lastName to get fullName

Find an element in an array

Find max value in array bids

Time transformation

Convert timestamp from MILLISECONDS to HOURS

Column name change

Change name of the column from user_id to userId

Extract value from a column containing space

Pinot doesn't support columns that have spaces, so if a source data column has a space, we'll need to store that value in a column with a supported name. To extract the value from first Name into the column firstName, run the following:

Ternary operation

If eventType is IMPRESSION set impression to 1. Similar for CLICK.

AVRO Map

Store an AVRO Map in Pinot as two multi-value columns. Sort the keys, to maintain the mapping. 1) The keys of the map as map_keys 2) The values of the map as map_values

Chaining transformations

Transformations can be chained. This means that you can use a field created by a transformation in another transformation function.

For example, we might have the following JSON document in the data field of our source data:

We can apply one transformation to extract the userId and then another one to pull out the numerical part of the identifier:

Flattening

There are 2 kinds of flattening:

One record into many

This is not natively supported as of yet. You can write a custom Decoder/RecordReader if you want to use this. Once the Decoder generates the multiple GenericRows from the provided input record, a List<GenericRow> should be set into the destination GenericRow, with the key $MULTIPLE_RECORDS_KEY$. The segment generation drivers will treat this as a special case and handle the multiple records case.

Extract attributes from complex objects

Feature TBD

Add a new column during ingestion

If a new column is added to table or schema configuration during ingestion, incorrect data may appear in the consuming segment(s).

To ensure accurate values are reloaded, do the following:

1. Pause consumption (and wait for pause status success): $ curl -X POST {controllerHost}/tables/{tableName}/pauseConsumption

2. Apply new table or schema configurations.

3. using the or .

Start Pinot components (scripts or docker images)

Set up Pinot by starting each component individually

Start Pinot Using Config Files

Often times we need to customized the setup of Pinot components. Hence user can compile a config file and use it to start Pinot components.

Below are the examples config files and sample command to start Pinot.

Pinot Controller

Below is a sample pinot-controller.conf used in HelmChart setup.

In order to run Pinot Controller, the command is:

Configure Controller

Below are some configurations you can set in Pinot Controller. You can head over to for complete list of available configs.

Pinot Broker

Below is a sample pinot-broker.conf used in HelmChart setup.

In order to run Pinot Broker, the command is:

Configure Broker

Below are some configurations you can set in Pinot Broker. You can head over to for complete list of available configs.

Pinot Server

Below is a sample pinot-server.conf used in HelmChart setup.

In order to run Pinot Server, the command is:

Configure Server

Below are some outstanding configurations you can set in Pinot Server. You can head over to for complete list of available configs.

Create and Configure table

A TABLE in regular database world is represented as <TABLE>_OFFLINE and/or <TABLE>_REALTIME in Pinot depending on the ingestion mode (batch, real-time, hybrid)

See for all possible batch/streaming tables.

Batch Table Creation

See for table configuration details and how to customize it.

Automatically add an inverted index to your batch table

By default, the inverted index type is the only type of index that isn't created automatically during segment generation. Instead, they are generated when the segments are loaded on the server. But, waiting to build indexes until load time increases the startup time and takes up resources with every new segment push, which increases the time for other operations such as rebalance.

To automatically create an inverted index during segment generation, add an entry to your in the table configuration file.

This setting works with .

When set to true, Pinot creates an inverted index for the columns that you specify in the invertedIndexColumns list in the table configuration.

This setting is false by default.

Set createInvertedIndexDuringSegmentGeneration to true in your table config, as follows:

When you update this setting in your table configuration, you must to apply the inverted index to all existing segments.

Streaming Table Creation

See for table configuration details and how to customize it.

Use sortedColumn with streaming tables

For tables, you can use a sorted index with sortedColumn to sort data when generating segments as the segment is created. See for more information.

A sorted forward index can be used as an inverted index with better performance, but with the limitation that the search is only applied to one column per table. See to learn more.

Load Data

Now that the table is configured, let's load some data. Data can be loaded in batch mode or streaming mode. See page for details. Loading data involves generating pinot segments from raw data and pushing them to the pinot cluster.

Load Data in Batch

User can always generate and push segments to Pinot via standalone scripts or using frameworks such as Hadoop or Spark. See this for more details on setting up Data Ingestion Jobs.

Below example goes with the standalone mode.

JobSpec yaml file has all the information regarding data format, input data location and pinot cluster coordinates. Note that this assumes that the controller is RUNNING to fetch the table config and schema. If not, you will have to configure the spec to point at their location. See for more details.

Load Data in Streaming

Kafka