Get Started Today: StarTree Free Tier
Backfill offline segment

Backfill offline segment

In this recipe we'll learn how to backfill a segment moved from a real-time to offline table using the real-time to offline job.


To follow the code examples in this guide, you must install Docker (opens in a new tab) locally and download recipes.

Navigate to recipe

  1. If you haven't already, download recipes.
  2. In terminal, run the following command to navigate to the recipe:
cd pinot-recipes/recipes/backfill

Launch Pinot Cluster

You can spin up a Pinot Cluster by running the following command:

docker-compose up

or (if you're using a Mac M1/M2)

docker-compose -f docker-compose-m1.yml up

This command will run a single instance of the Pinot Controller, Pinot Server, Pinot Broker, Pinot Minion, Kafka, and Zookeeper. You can find the docker-compose.yml (opens in a new tab) file on GitHub.

Pinot Schema and Tables

Now let's create a Pinot Schema, as well as real-time and offline tables. Pinot is going to take care of populating data into the offline table, but it still expects us to configure the table.


Our schema is going to capture some order events, and looks like this:

  "schemaName": "orders",
  "dimensionFieldSpecs": [{
      "name": "order_id",
      "dataType": "INT"
      "name": "customer_id",
      "dataType": "INT"
      "name": "order_status",
      "dataType": "STRING"
  "metricFieldSpecs": [{
    "name": "amount",
    "dataType": "FLOAT"
  "dateTimeFieldSpecs": [{
    "name": "ts",
    "dataType": "LONG",
    "format": "1:MILLISECONDS:EPOCH",
    "granularity": "1:MILLISECONDS"


Real-Time Table

And the real-time table is defined below:

  "tableName": "orders",
  "tableType": "REALTIME",
  "segmentsConfig": {
    "timeColumnName": "ts",
    "timeType": "MILLISECONDS",
    "segmentPushType": "APPEND",
    "segmentAssignmentStrategy": "BalanceNumSegmentAssignmentStrategy",
    "schemaName": "orders",
    "replicasPerPartition": "1"
  "task": {
    "taskTypeConfigsMap": {
      "RealtimeToOfflineSegmentsTask": {
        "bucketTimePeriod": "2h",
        "bufferTimePeriod": "1m"
  "tenants": {},
  "tableIndexConfig": {
    "loadMode": "MMAP",
    "streamConfigs": {
      "streamType": "kafka",
      "stream.kafka.consumer.type": "lowLevel",
      "": "orders",
      "": "",
      "": "",
      "": "kafka:9093",
      "realtime.segment.flush.threshold.rows": 5
  "metadata": {
    "customConfigs": {}
  "routing": {
    "instanceSelectorType": "strictReplicaGroup"



The realtime.segment.flush.threshold.rows config is intentionally set to an extremely small value so that the segment will be committed after 5 records have been ingested. In a production system this value should be set much higher, as described in the configuring segment threshold guide.

You can create the table and schema by running the following command:

docker run \
   --network backfill \
   -v $PWD/config:/config \
   apachepinot/pinot:1.0.0 AddTable \
     -schemaFile /config/orders_schema.json \
     -tableConfigFile /config/orders_table.json \
     -controllerHost "pinot-controller" \

Offline Table

The offline table config is defined below:

  "tableName": "events",
  "tableType": "OFFLINE",
  "segmentsConfig": {
    "timeColumnName": "ts",
    "schemaName": "events",
    "replication": "1",
    "replicasPerPartition": "1"
  "ingestionConfig": {
    "batchIngestionConfig": {
      "segmentIngestionType": "APPEND",
      "segmentIngestionFrequency": "HOURLY"
  "tableIndexConfig": {
    "loadMode": "MMAP"
  "tenants": {},
  "metadata": {}


You can create this table by running the following command:

docker run \
   --network backfill \
   -v $PWD/config:/config \
   apachepinot/pinot:1.0.0 AddTable \
     -schemaFile /config/orders_schema.json \
     -tableConfigFile /config/orders_offline_table.json \
     -controllerHost "pinot-controller" \
    -update -exec

Ingesting Data

Let's ingest data into the events Kafka topic, by running the following:

docker exec -it kafka /opt/kafka/bin/ \
  --bootstrap-server kafka:9092 --topic orders

Data will make its way into the real-time table. We can see how many records have been ingested by running the following query:

select * 
from orders 
order by order_id 
limit 10

Orders query results Orders query results

Scheduling the RT2OFF Job

The Real-Time to Offline Job can be scheduled automatically via the real-time table config or manually via the REST API. We can trigger it manually by running the following command:

curl -X POST \
  "http://localhost:9000/tasks/schedule?taskType=RealtimeToOfflineSegmentsTask&tableName=${table}" \
  -H "accept: application/json" 2>/dev/null | jq '.'


  "RealtimeToOfflineSegmentsTask": "Task_RealtimeToOfflineSegmentsTask_1647620599577"

We can then check the Pinot Controller logs to see that it's been triggered:

docker exec -it pinot-controller grep -i "\[RealtimeToOff" logs/pinot-all.log


2022/03/21 10:58:29.746 INFO [RealtimeToOfflineSegmentsTaskGenerator] [grizzly-http-server-1] Start generating task configs for table: orders_REALTIME for task: RealtimeToOfflineSegmentsTask
2022/03/21 10:58:29.763 INFO [RealtimeToOfflineSegmentsTaskGenerator] [grizzly-http-server-1] Finished generating task configs for table: orders_REALTIME for task: RealtimeToOfflineSegmentsTask

Now let's navigate to localhost:9000/#/tables (opens in a new tab). You'll see the following:

Real-Time and Offline Tables

You can see that a segment has been created in the offline table:

Viewing offline segment

You can list all the segments for a table by making the following request to the HTTP API:

curl -X GET "http://localhost:9000/segments/${table}" \
  -H "accept: application/json" 2>/dev/null | jq '.'


    "OFFLINE": [
    "REALTIME": [

We have one offline segment - orders_1632463351000_1632467070000_0. You can check which records it contains by running the following query:

select * 
from orders_OFFLINE
order by order_id 
limit 10

Orders offline table query results Orders offline table query results

The offline table contains only the first 7 records that we ingested.

Replacing offline segment

Let's now backfill those 7 records to increase the value in the amount column by 20%. The documents with the updated amount value are in data/orders.json, shown below:

{"order_id": 1, "customer_id": 101, "order_status": "OPEN", "amount": 15.947999999999999, "ts": "1632463351000"}
{"order_id": 2, "customer_id": 102, "order_status": "IN_TRANSIT", "amount": 251.21999999999997, "ts": "1632463361000"}
{"order_id": 3, "customer_id": 103, "order_status": "COMPLETED", "amount": 239.21999999999997, "ts": "1632463391000"}
{"order_id": 4, "customer_id": 105, "order_status": "COMPLETED", "amount": 3.888, "ts": "1632467065000"}
{"order_id": 5, "customer_id": 103, "order_status": "OPEN", "amount": 11.723999999999998, "ts": "1632467066000"}
{"order_id": 6, "customer_id": 104, "order_status": "OPEN", "amount": 66.624, "ts": "1632467068000"}
{"order_id": 7, "customer_id": 104, "order_status": "CANCELLED", "amount": 63.047999999999995, "ts": "1632467070000"}


We'll ingest this file using the following ingestion spec:

  name: 'standalone'
  segmentGenerationJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentGenerationJobRunner'
  segmentTarPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.standalone.SegmentTarPushJobRunner'
  type: fixed
  configs: ${segmentName}
jobType: SegmentCreationAndTarPush
inputDirURI: '/data'
includeFileNamePattern: 'glob:**/orders.json'
outputDirURI: '/opt/pinot/data/orders/'
overwriteOutput: true
  - scheme: file
    className: org.apache.pinot.spi.filesystem.LocalPinotFS
  dataFormat: 'json'
  className: 'org.apache.pinot.plugin.inputformat.json.JSONRecordReader'
  tableName: 'orders'
  - controllerURI: '${pinotController}'
  pushAttempts: 2
  pushRetryIntervalMillis: 1000


Now let's run the job to replace segment orders_1632463351000_1632467070000_0:

docker run \
   --network backfill \
   -v $PWD/config:/config \
   -v $PWD/data:/data \
   apachepinot/pinot:1.0.0 LaunchDataIngestionJob \
    -jobSpecFile /config/job-spec.yml \
    -values segmentName='orders_1632463351000_1632467070000_0' \
    -values pinotController=http://pinot-controller:9000

Once we've done that, we need to update the time boundary so that it starts from the latest time in the offline table:

curl -X POST \
  "http://localhost:9000/tables/orders/timeBoundary" \
  -H "accept: application/json"

We can re-run the query on the orders table:

select $segmentName, 
from orders
ORDER BY order_id
limit 20

And we'll see that the amount column for the first 7 orders has been updated:


Query Results