https://pinot.apache.org/ logo
Join Slack
Powered by
Hello there, how can I configure Merge Rollup task...
# troubleshooting
t
Hello there, how can I configure Merge Rollup tasks to use 
metadata
push mode to save the segments in the deep store and keep only metadata on the Controller?
m
@Shounak Kulkarni
thanks 2
s
you can pass 
"push.mode" : "METADATA"
to the task configs to enable the metdata push mode and also configure the deep store location using 
output.segment.dir.uri
. You may need to configure few more configs (
output.fs.prop.xxx
) based on which FS you are aiming to push the segments ( 
output.fs.className
)
For example if using S3 it should look like
Copy code
"MergeRollupTask": {
  "output.segment.dir.uri": "<s3://bucketname/datadir/>",
  "input.fs.className": "org.apache.pinot.plugin.filesystem.S3PinotFS",
  "input.fs.prop.accessKey": "xxxxx",
  "input.fs.prop.secretKey": "xxxx",
  "input.fs.prop.region": "us-west-2",
  "push.mode": "METADATA",
  // other task configs
}
t
Copy code
{
  "OFFLINE": {
    "tableName": "LOADED_HOURLY_OFFLINE",
    "tableType": "OFFLINE",
    "segmentsConfig": {
      "replication": "2",
      "retentionTimeUnit": "DAYS",
      "retentionTimeValue": "7",
      "timeType": "HOURS",
      "timeColumnName": "ts",
      "minimizeDataMovement": false
    },
    "tenants": {
      "broker": "DefaultTenant",
      "server": "DefaultTenant"
    },
    "tableIndexConfig": {
      "invertedIndexColumns": [],
      "rangeIndexColumns": [],
      "rangeIndexVersion": 2,
      "columnMajorSegmentBuilderEnabled": false,
      "optimizeDictionary": true,
      "optimizeDictionaryForMetrics": true,
      "noDictionarySizeRatioThreshold": 0,
      "autoGeneratedInvertedIndex": false,
      "createInvertedIndexDuringSegmentGeneration": false,
      "sortedColumn": [],
      "bloomFilterColumns": [],
      "noDictionaryColumns": [
        "counters.hll_sketch"
      ],
      "noDictionaryConfig": {
        "counters.hll_sketch": "LZ4"
      },
      "onHeapDictionaryColumns": [
        "fields.platform"
      ],
      "varLengthDictionaryColumns": [
        "fields.screen_name",
        "fields.app_version"
      ],
      "enableDefaultStarTree": false,
      "starTreeIndexConfigs": [
        ...
      ],
      "enableDynamicStarTreeCreation": true,
      "aggregateMetrics": false,
      "nullHandlingEnabled": false
    },
    "metadata": {},
    "task": {
      "taskTypeConfigsMap": {
        "SegmentGenerationAndPushTask": {},
        "MergeRollupTask": {
          "1day.mergeType": "concat",
          "1day.bucketTimePeriod": "1d",
          "1day.bufferTimePeriod": "1d",
          "1day.maxNumRecordsPerSegment": "100000",
          "1day.maxNumRecordsPerTask": "500000",
          "1day.maxNumParallelBuckets": "10",
          "minionInstanceTag": "merge",
          "push.mode": "METADATA",
          "output.segment.dir.uri": "<gs://my-bucket/LOADED_HOURLY/merged>",
          "schedule": "0 1 * * * ?"
        }
      }
    },
    "fieldConfigList": [
      {
        "name": "counters.hll_sketch",
        "encodingType": "RAW",
        "indexTypes": [],
        "compressionCodec": "LZ4",
        "indexes": null,
        "tierOverwrites": null
      }
    ],
    "ingestionConfig": {
      "batchIngestionConfig": {
        "batchConfigMaps": [
          {
            "includeFileNamePattern": "glob:**/*/*.gz",
            "input.fs.className": "org.apache.pinot.plugin.filesystem.GcsPinotFS",
            "inputDirURI": "<gs://my-bucket/source>",
            "outputDirURI": "<gs://my-bucket/LOADED_HOURLY>",
            "inputFormat": "json",
            "overwriteOutput": "true",
            "recordReaderSpec.dataFormat": "json",
            "push.mode": "metadata",
            "segmentNameGenerator.type": "inputFile",
            "segmentNameGenerator.configs.file.path.pattern": "",
            "segmentNameGenerator.configs.segment.name.template": "\\${filePathPattern:\\1}"
          }
        ],
        "segmentIngestionType": "APPEND",
        "segmentIngestionFrequency": "HOURLY",
        "consistentDataPush": false
      },
      "transformConfigs": [
        {
          "columnName": "ts",
          "transformFunction": "FromDateTime(\"timeString\", 'yyyy-MM-dd HH')"
        },
        {
          "columnName": "job_id",
          "transformFunction": "jobId"
        }
      ],
      "complexTypeConfig": {
        "fieldsToUnnest": [
          "fields",
          "counters"
        ],
        "delimiter": ".",
        "collectionNotUnnestedToJson": "NON_PRIMITIVE"
      },
      "continueOnError": true,
      "rowTimeValueCheck": false,
      "segmentTimeValueCheck": true
    },
    "isDimTable": false
  }
}
This is my table config. The resulting task config is the following:
Copy code
{
  "configs": {
    "minionInstanceTag": "merge",
    "maxNumRecordsPerSegment": "100000",
    "1day.maxNumRecordsPerSegment": "100000",
    "downloadURL": "<gs://my-bucket/files>...",
    "mergeLevel": "1day",
    "push.mode": "TAR",
    "tableName": "LOADED_HOURLY_OFFLINE",
    "1day.maxNumParallelBuckets": "10",
    "enableReplaceSegments": "true",
    "1day.mergeType": "concat",
    "push.controllerUri": "<http://pinot-prod-controller-1.pinot-prod-controller-headless.pinot.svc.cluster.local:9000>",
    "roundBucketTimePeriod": null,
    "1day.bufferTimePeriod": "1d",
    "output.segment.dir.uri": "<gs://my-bucket/LOADED_HOURLY/merged>",
    "partitionBucketTimePeriod": "1d",
    "segmentNamePrefix": "merged_1day_1759943074268_0_AD_LOADED_HOURLY",
    "mergeType": "concat",
    "segmentName": "hotpanel_graphs_by_event_event_graph_859_hour_job_id=hive_2025_10_02_00_10_00_32b6b454c8f74e361c0c5d9525b06d57_gz_71cd43d4-ef06-49f4-b7cb-9a9cb5167df4",
    "schedule": "0 1 * * * ?",
    "uploadURL": "<http://pinot-prod-controller-1.pinot-prod-controller-headless.pinot.svc.cluster.local:9000/segments>",
    "1day.bucketTimePeriod": "1d",
    "1day.maxNumRecordsPerTask": "500000",
    "TASK_ID": "Task_MergeRollupTask_5cd6364b-7012-4f60-8e8b-58ee4a2196c1_1759943074336_0",
    "overwriteOutput": "false"
  },
  "tableName": "AD_LOADED_HOURLY_OFFLINE",
  "taskId": "Task_MergeRollupTask_5cd6364b-7012-4f60-8e8b-58ee4a2196c1_1759943074336_0",
  "taskType": "MergeRollupTask"
}
It is overridden no matter what I try to do.
I'm on version 1.1, maybe I should upgrade
s
yeah try upgrading, it should pick the configured mode
t
Upgrading did NOT help. Any thoughts? Could you try to reproduce?
s
sure will check on my end
Hey looks like the code is pulling the output dir from controller data dir, in your case that's local dir, due to which the 
isLocalOutputDir
check is returning true and ending up taking the TAR mode. Seems like a bug to me as it should check for provided 
output.segment.dir.uri
and use that.
can you verify this by looking for the log line 
segment upload with METADATA push is not supported with local output dir:
in controller?
t
I’ll check tomorrow. If it’s a bug I can open an issue and fix
🥂 1
Confirmed that this is a bug
when I'll have some time next week I'll open an issue and fix
🙌 2
Just for future reference, if anybody incurs in this problem before the pr is merged. This issue can be fixed by setting this property on controller config, provided that you have all the tables in the same deep-store with the same path pattern.
Copy code
controller.data.dir=<hdfs://your-deep-store>
thanks 1
m
Thanks @Tommaso Peresson
Open in Slack
PreviousNext
Powered by