Pinot enables its users to write a PinotFS abstraction layer to store data in a data layer of their choice for realtime and offline segments.

Some examples of storage backends(other than local storage) currently supported are:

• HadoopFS

• Azure Data Lake

If the above two filesystems do not meet your needs, you can extend the current to customize for your needs.

New Storage Type implementation

In order to add a new type of storage backend (say, Amazon s3) implement the following class:

S3FS extends

Configurations for Realtime Tables

The example here uses the existing org.apache.pinot.filesystem.HadoopPinotFS to store realtime segments in a HDFS filesytem. In the Pinot controller config, add the following new configs:

In the Pinot controller config, add the following new configs:

Note: currently there is a bug in the controller (issue <), for now you can cherrypick the PR to fix the issue as tested already. The PR is under review now.

Configurations for Offline Tables

These properties for the stream implementation are to be set in your controller and server configurations.

In your controller and server configs, please set the FS class you would like to support. pinot.controller.storage.factory.class.${YOUR_URI_SCHEME} to the full path of the FS class you would like to include

You also need to configure pinot.controller.local.temp.dir for the local dir on the controller machine.

For filesystem specific configs, you can pass in the following with either the pinot.controller prefix or the pinot.server prefix.

All the following configs need to be prefixed with storage.factory.

AzurePinotFS requires the following configs according to your environment:

adl.accountId, adl.authEndpoint, adl.clientId, adl.clientSecret

Sample Controller Config

Sample Server Config

You can find the parameters in your account as follows:

Please also make sure to set the following config with the value “adl”

To see how to upload segments to different storage systems, check ../segment_fetcher.rst.

HadoopPinotFS requires the following configs according to your environment:

hadoop.kerberos.principle, hadoop.kerberos.keytab, hadoop.conf.path

Please make sure to also set the following config with the value “hdfs”

Prior to commit , Pinot was only able to support consuming from stream.

Pinot now enables its users to write plug-ins to consume from pub-sub streams other than Kafka. (Please refer to )

Some of the streams for which plug-ins can be added are:

Pinot has many inbuilt Aggregation Functions such as MIN, MAX, SUM, AVG etc. See PQL page for the list of aggregation functions.

Adding a new AggregationFunction requires two things

• Implement AggregationFunction interface and make it available as part of the classpath

• Register the function in . As of today, this requires code change in Pinot but we plan to add the ability to plugin Functions without having to change Pinot code.

To get an overall idea, see Aggregation Function implementation. All other implementations can be found .

Lets look at the key methods to implements in AggregationFunction

Before getting into the implementation, it's important to understand how Aggregation works in Pinot.

This is advanced topic and assumes you know Pinot . All the data in Pinot is stored in segments across multiple nodes. The query plan at a high level comprises of 3 phases

1. Map phase

This phase works on the individual segments in Pinot.

• Initialization: Depending on the query type the following methods are invoked to setup the result holder. While having different methods and return types adds complexity, it helps in performance.

  • AGGREGATION : createAggregationResultHolderThis must return an instance of type . You can either use the or

2. Combine phase

In this phase, the results from all segments within a single pinot server are combined into IntermediateResult. The type of IntermediateResult is based on the Generic Type defined in the AggregationFunction implementation.

3. Reduce phase

There are two steps in the Reduce Phase

• Merge all the IntermediateResult's from various servers using the merge function

• Extract the final results by invoking the extractFinalResult method. In most cases, FinalResult is same type as IntermediateResult. is an example where IntermediateResult (AvgPair) is different from FinalResult(Double)

When Pinot segment files are created in external systems (hadoop/spark/etc), there are several ways to push those data to Pinot Controller and Server:

1. push segment to shared NFS and let Pinot pull segment files from the location of that NFS.

2. push segment to a Web server and let Pinot pull segment files from the Web server with http/https link.

Pinot supports indexing data from various file formats. To support reading from a file format, a record reader need to be provided to read the file and convert records into the general format which the indexing engine can understand. The record reader serves as the connector from each individual file format to Pinot record format.

Pinot package provides the following record readers out of the box:

• Avro record reader: record reader for Avro format files

• CSV record reader: record reader for CSV format files

• JSON record reader: record reader for JSON format files

• ORC record reader: record reader for ORC format files

• Thrift record reader: record reader for Thrift format files

• Pinot segment record reader: record reader for Pinot segment

Initialize Record Reader

To initialize a record reader, the data file and table schema should be provided (for Pinot segment record reader, only need to provide the index directory because schema can be derived from the segment). The output record will follow the table schema provided.

For Avro/JSON/ORC/Pinot segment record reader, no extra configuration is required as column names and multi-values are embedded in the data file.

For CSV/Thrift record reader, extra configuration might be provided to determine the column names and multi-values for the data.

CSV Record Reader Config

The CSV record reader config contains the following settings:

• Header: the header for the CSV file (column names)

• Column delimiter: delimiter for each column

• Multi-value delimiter: delimiter for each value for a multi-valued column

If no config provided, use the default setting:

• Use the first row in the data file as the header

• Use ‘,’ as the column delimiter

• Use ‘;’ as the multi-value delimiter

Thrift Record Reader Config

The Thrift record reader config is mandatory. It contains the Thrift class name for the record reader to de-serialize the Thrift objects.

ORC Record Reader Config

The following property is to be set during segment generation in your Hadoop properties.

record.reader.path: ${FULL_PATH_OF_YOUR_RECORD_READER_CLASS}

For ORC, it would be:

record.reader.path: org.apache.pinot.orc.data.readers.ORCRecordReader

Implement Your Own Record Reader

For other file formats, we provide a general interface for record reader - . To index the file into Pinot segment, simply implement the interface and plug it into the index engine - . We use a 2-passes algorithm to index the file into Pinot segment, hence the rewind() method is required for the record reader.

Generic Row

is the record abstraction which the index engine can read and index with. It is a map from column name (String) to column value (Object). For multi-valued column, the value should be an object array (Object[]).

Contracts for Record Reader

There are several contracts for record readers that developers should follow when implementing their own record readers:

• The output GenericRow should follow the table schema provided, in the sense that:

  • All the columns in the schema should be preserved (if column does not exist in the original record, put default value instead)

  • Columns not in the schema should not be included