SQL-based ingestion and multi-stage query task API

The Query view in the web console provides a friendly experience for the multi-stage query task engine (MSQ task engine) and multi-stage query architecture. We recommend using the web console if you do not need a programmatic interface.

When using the API for the MSQ task engine, the action you want to take determines the endpoint you use:

  • endpoint: Submit a query for ingestion.
  • /druid/indexer/v1/task endpoint: Interact with a query, including getting its status, getting its details, or canceling it. This page describes a few of the Overlord Task APIs that you can use with the MSQ task engine. For information about Druid APIs, see the .

You submit queries to the MSQ task engine using the POST /druid/v2/sql/task/ endpoint.

Request

The SQL task endpoint accepts SQL requests in the JSON-over-HTTP form using the query, context, and parameters fields, but ignoring the resultFormat, header, typesHeader, and sqlTypesHeader fields.

This endpoint accepts and REPLACE statements.

As an experimental feature, this endpoint also accepts SELECT queries. SELECT query results are collected from workers by the controller, and written into the as an array of arrays. The behavior and result format of plain SELECT queries (without INSERT or REPLACE) is subject to change.

HTTP

curl

  1. {  "query": "INSERT INTO wikipedia\nSELECT\n  TIME_PARSE(\"timestamp\") AS __time,\n  *\nFROM TABLE(\n  EXTERN(\n    '{\"type\": \"http\", \"uris\": [\"https://druid.apache.org/data/wikipedia.json.gz\"]}',\n    '{\"type\": \"json\"}',\n    '[{\"name\": \"added\", \"type\": \"long\"}, {\"name\": \"channel\", \"type\": \"string\"}, {\"name\": \"cityName\", \"type\": \"string\"}, {\"name\": \"comment\", \"type\": \"string\"}, {\"name\": \"commentLength\", \"type\": \"long\"}, {\"name\": \"countryIsoCode\", \"type\": \"string\"}, {\"name\": \"countryName\", \"type\": \"string\"}, {\"name\": \"deleted\", \"type\": \"long\"}, {\"name\": \"delta\", \"type\": \"long\"}, {\"name\": \"deltaBucket\", \"type\": \"string\"}, {\"name\": \"diffUrl\", \"type\": \"string\"}, {\"name\": \"flags\", \"type\": \"string\"}, {\"name\": \"isAnonymous\", \"type\": \"string\"}, {\"name\": \"isMinor\", \"type\": \"string\"}, {\"name\": \"isNew\", \"type\": \"string\"}, {\"name\": \"isRobot\", \"type\": \"string\"}, {\"name\": \"isUnpatrolled\", \"type\": \"string\"}, {\"name\": \"metroCode\", \"type\": \"string\"}, {\"name\": \"namespace\", \"type\": \"string\"}, {\"name\": \"page\", \"type\": \"string\"}, {\"name\": \"regionIsoCode\", \"type\": \"string\"}, {\"name\": \"regionName\", \"type\": \"string\"}, {\"name\": \"timestamp\", \"type\": \"string\"}, {\"name\": \"user\", \"type\": \"string\"}]'\n  )\n)\nPARTITIONED BY DAY",  "context": {    "maxNumTasks": 3  }}
  1. # Make sure you replace `username`, `password`, `your-instance`, and `port` with the values for your deployment.curl --location --request POST 'https://<username>:<password>@<your-instance>:<port>/druid/v2/sql/task/' \  --header 'Content-Type: application/json' \  --data-raw '{    "query": "INSERT INTO wikipedia\nSELECT\n  TIME_PARSE(\"timestamp\") AS __time,\n  *\nFROM TABLE(\n  EXTERN(\n    '\''{\"type\": \"http\", \"uris\": [\"https://druid.apache.org/data/wikipedia.json.gz\"]}'\'',\n    '\''{\"type\": \"json\"}'\'',\n    '\''[{\"name\": \"added\", \"type\": \"long\"}, {\"name\": \"channel\", \"type\": \"string\"}, {\"name\": \"cityName\", \"type\": \"string\"}, {\"name\": \"comment\", \"type\": \"string\"}, {\"name\": \"commentLength\", \"type\": \"long\"}, {\"name\": \"countryIsoCode\", \"type\": \"string\"}, {\"name\": \"countryName\", \"type\": \"string\"}, {\"name\": \"deleted\", \"type\": \"long\"}, {\"name\": \"delta\", \"type\": \"long\"}, {\"name\": \"deltaBucket\", \"type\": \"string\"}, {\"name\": \"diffUrl\", \"type\": \"string\"}, {\"name\": \"flags\", \"type\": \"string\"}, {\"name\": \"isAnonymous\", \"type\": \"string\"}, {\"name\": \"isMinor\", \"type\": \"string\"}, {\"name\": \"isNew\", \"type\": \"string\"}, {\"name\": \"isRobot\", \"type\": \"string\"}, {\"name\": \"isUnpatrolled\", \"type\": \"string\"}, {\"name\": \"metroCode\", \"type\": \"string\"}, {\"name\": \"namespace\", \"type\": \"string\"}, {\"name\": \"page\", \"type\": \"string\"}, {\"name\": \"regionIsoCode\", \"type\": \"string\"}, {\"name\": \"regionName\", \"type\": \"string\"}, {\"name\": \"timestamp\", \"type\": \"string\"}, {\"name\": \"user\", \"type\": \"string\"}]'\''\n  )\n)\nPARTITIONED BY DAY",    "context": {        "maxNumTasks": 3    }  }'
  1. import jsonimport requests# Make sure you replace `username`, `password`, `your-instance`, and `port` with the values for your deployment.url = "https://<username>:<password>@<your-instance>:<port>/druid/v2/sql/task/"payload = json.dumps({  "query": "INSERT INTO wikipedia\nSELECT\n  TIME_PARSE(\"timestamp\") AS __time,\n  *\nFROM TABLE(\n  EXTERN(\n    '{\"type\": \"http\", \"uris\": [\"https://druid.apache.org/data/wikipedia.json.gz\"]}',\n    '{\"type\": \"json\"}',\n    '[{\"name\": \"added\", \"type\": \"long\"}, {\"name\": \"channel\", \"type\": \"string\"}, {\"name\": \"cityName\", \"type\": \"string\"}, {\"name\": \"comment\", \"type\": \"string\"}, {\"name\": \"commentLength\", \"type\": \"long\"}, {\"name\": \"countryIsoCode\", \"type\": \"string\"}, {\"name\": \"countryName\", \"type\": \"string\"}, {\"name\": \"deleted\", \"type\": \"long\"}, {\"name\": \"delta\", \"type\": \"long\"}, {\"name\": \"deltaBucket\", \"type\": \"string\"}, {\"name\": \"diffUrl\", \"type\": \"string\"}, {\"name\": \"flags\", \"type\": \"string\"}, {\"name\": \"isAnonymous\", \"type\": \"string\"}, {\"name\": \"isMinor\", \"type\": \"string\"}, {\"name\": \"isNew\", \"type\": \"string\"}, {\"name\": \"isRobot\", \"type\": \"string\"}, {\"name\": \"isUnpatrolled\", \"type\": \"string\"}, {\"name\": \"metroCode\", \"type\": \"string\"}, {\"name\": \"namespace\", \"type\": \"string\"}, {\"name\": \"page\", \"type\": \"string\"}, {\"name\": \"regionIsoCode\", \"type\": \"string\"}, {\"name\": \"regionName\", \"type\": \"string\"}, {\"name\": \"timestamp\", \"type\": \"string\"}, {\"name\": \"user\", \"type\": \"string\"}]'\n  )\n)\nPARTITIONED BY DAY",  "context": {    "maxNumTasks": 3  }})headers = {  'Content-Type': 'application/json'}response = requests.request("POST", url, headers=headers, data=payload)print(response.text)

Response

  1. {
  2.   "taskId": "query-f795a235-4dc7-4fef-abac-3ae3f9686b79",
  3.   "state": "RUNNING",
  4. }

Response fields

You can retrieve status of a query to see if it is still running, completed successfully, failed, or got canceled.

Request

HTTP

curl

Python

  1. # Make sure you replace `username`, `password`, `your-instance`, `port`, and `taskId` with the values for your deployment.curl --location --request GET 'https://<username>:<password>@<hostname>:<port>/druid/indexer/v1/task/<taskId>/status'
  1. import requests# Make sure you replace `username`, `password`, `your-instance`, `port`, and `taskId` with the values for your deployment.url = "https://<username>:<password>@<hostname>:<port>/druid/indexer/v1/task/<taskId>/status"payload={}headers = {}response = requests.request("GET", url, headers=headers, data=payload)print(response.text)

Response

  1. {
  2.   "task": "query-3dc0c45d-34d7-4b15-86c9-cdb2d3ebfc4e",
  3.   "status": {
  4.     "id": "query-3dc0c45d-34d7-4b15-86c9-cdb2d3ebfc4e",
  5.     "groupId": "query-3dc0c45d-34d7-4b15-86c9-cdb2d3ebfc4e",
  6.     "type": "query_controller",
  7.     "createdTime": "2022-09-14T22:12:00.183Z",
  8.     "queueInsertionTime": "1970-01-01T00:00:00.000Z",
  9.     "statusCode": "RUNNING",
  10.     "status": "RUNNING",
  11.     "runnerStatusCode": "RUNNING",
  12.     "duration": -1,
  13.     "location": {
  14.       "host": "localhost",
  15.       "port": 8100,
  16.       "tlsPort": -1
  17.     },
  18.     "dataSource": "kttm_simple",
  19.     "errorMsg": null
  20.   }
  21. }

A report provides detailed information about a query task, including things like the stages, warnings, and errors.

Keep the following in mind when using the task API to view reports:

  • The task report for an entire job is associated with the query_controller task. The query_worker tasks do not have their own reports; their information is incorporated into the controller report.
  • The task report API may report 404 Not Found temporarily while the task is in the process of starting up.
  • As an experimental feature, the MSQ task engine supports running SELECT queries. SELECT query results are written into the multiStageQuery.payload.results.results task report key as an array of arrays. The behavior and result format of plain SELECT queries (without INSERT or REPLACE) is subject to change.

For an explanation of the fields in a report, see .

Request

curl

Python

  1. GET /druid/indexer/v1/task/<taskId>/reports
  1. import requests# Make sure you replace `username`, `password`, `your-instance`, `port`, and `taskId` with the values for your deployment.url = "https://<username>:<password>@<hostname>:<port>/druid/indexer/v1/task/<taskId>/reports"headers = {}response = requests.request("GET", url, headers=headers)print(response.text)

Response

The response shows an example report for a query.

Show the response

  1. {
  2.   "multiStageQuery": {
  3.     "type": "multiStageQuery",
  4.     "taskId": "query-3dc0c45d-34d7-4b15-86c9-cdb2d3ebfc4e",
  5.     "payload": {
  6.       "status": {
  7.         "status": "SUCCESS",
  8.         "startTime": "2022-09-14T22:12:09.266Z",
  9.         "durationMs": 28227
  10.       },
  11.       "stages": [
  12.         {
  13.           "stageNumber": 0,
  14.           "definition": {
  15.             "id": "71ecb11e-09d7-42f8-9225-1662c8e7e121_0",
  16.             "input": [
  17.               {
  18.                 "type": "external",
  19.                 "inputSource": {
  20.                   "type": "http",
  21.                   "uris": [
  22.                     "https://static.imply.io/example-data/kttm-v2/kttm-v2-2019-08-25.json.gz"
  23.                   ],
  24.                   "httpAuthenticationUsername": null,
  25.                   "httpAuthenticationPassword": null
  26.                 },
  27.                 "inputFormat": {
  28.                   "type": "json",
  29.                   "flattenSpec": null,
  30.                   "featureSpec": {},
  31.                   "keepNullColumns": false
  32.                 },
  33.                 "signature": [
  34.                   {
  35.                     "name": "timestamp",
  36.                     "type": "STRING"
  37.                   },
  38.                   {
  39.                     "name": "agent_category",
  40.                     "type": "STRING"
  41.                   },
  42.                   {
  43.                     "name": "agent_type",
  44.                     "type": "STRING"
  45.                   }
  46.                 ]
  47.               }
  48.             ],
  49.             "processor": {
  50.               "type": "scan",
  51.               "query": {
  52.                 "queryType": "scan",
  53.                 "dataSource": {
  54.                   "type": "inputNumber",
  55.                   "inputNumber": 0
  56.                 },
  57.                 "intervals": {
  58.                   "type": "intervals",
  59.                   "intervals": [
  60.                     "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
  61.                   ]
  62.                 },
  63.                 "resultFormat": "compactedList",
  64.                 "columns": [
  65.                   "agent_category",
  66.                   "timestamp"
  67.                 ],
  69.                 "context": {
  70.                   "finalize": false,
  71.                   "finalizeAggregations": false,
  72.                   "groupByEnableMultiValueUnnesting": false,
  73.                   "scanSignature": "[{\"name\":\"agent_category\",\"type\":\"STRING\"},{\"name\":\"agent_type\",\"type\":\"STRING\"},{\"name\":\"timestamp\",\"type\":\"STRING\"}]",
  74.                   "sqlInsertSegmentGranularity": "{\"type\":\"all\"}",
  75.                   "sqlQueryId": "3dc0c45d-34d7-4b15-86c9-cdb2d3ebfc4e",
  76.                   "sqlReplaceTimeChunks": "all"
  77.                 },
  78.                 "granularity": {
  79.                   "type": "all"
  80.                 }
  81.               }
  82.             },
  83.             "signature": [
  84.               {
  85.                 "name": "__boost",
  86.                 "type": "LONG"
  87.               },
  88.               {
  89.                 "name": "agent_category",
  90.                 "type": "STRING"
  91.               },
  92.               {
  93.                 "name": "agent_type",
  94.                 "type": "STRING"
  95.               },
  96.               {
  97.                 "name": "timestamp",
  98.                 "type": "STRING"
  99.               }
  100.             ],
  101.             "shuffleSpec": {
  102.               "type": "targetSize",
  103.               "clusterBy": {
  104.                 "columns": [
  105.                   {
  106.                     "columnName": "__boost"
  107.                   }
  108.                 ]
  109.               },
  110.               "targetSize": 3000000
  111.             },
  112.             "maxWorkerCount": 1,
  113.             "shuffleCheckHasMultipleValues": true
  114.           },
  115.           "phase": "FINISHED",
  116.           "workerCount": 1,
  117.           "partitionCount": 1,
  118.           "startTime": "2022-09-14T22:12:11.663Z",
  119.           "duration": 19965,
  120.           "sort": true
  121.         },
  122.         {
  123.           "stageNumber": 1,
  124.           "definition": {
  125.             "id": "71ecb11e-09d7-42f8-9225-1662c8e7e121_1",
  126.             "input": [
  127.               {
  128.                 "type": "stage",
  129.                 "stage": 0
  130.               }
  131.             ],
  132.             "processor": {
  133.               "type": "segmentGenerator",
  134.               "dataSchema": {
  135.                 "dataSource": "kttm_simple",
  136.                 "timestampSpec": {
  137.                   "column": "__time",
  138.                   "format": "millis",
  139.                   "missingValue": null
  140.                 },
  141.                 "dimensionsSpec": {
  142.                   "dimensions": [
  143.                     {
  144.                       "type": "string",
  145.                       "name": "timestamp",
  146.                       "multiValueHandling": "SORTED_ARRAY",
  147.                       "createBitmapIndex": true
  148.                     },
  149.                     {
  150.                       "type": "string",
  151.                       "name": "agent_category",
  152.                       "multiValueHandling": "SORTED_ARRAY",
  153.                       "createBitmapIndex": true
  154.                     },
  155.                     {
  156.                       "type": "string",
  157.                       "name": "agent_type",
  158.                       "multiValueHandling": "SORTED_ARRAY",
  159.                       "createBitmapIndex": true
  160.                     }
  161.                   ],
  162.                   "dimensionExclusions": [
  163.                     "__time"
  164.                   ],
  165.                   "includeAllDimensions": false
  166.                 },
  167.                 "metricsSpec": [],
  168.                 "granularitySpec": {
  169.                   "type": "arbitrary",
  170.                   "queryGranularity": {
  171.                     "type": "none"
  172.                   },
  173.                   "rollup": false,
  174.                   "intervals": [
  175.                     "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
  176.                   ]
  177.                 },
  178.                 "transformSpec": {
  179.                   "filter": null,
  180.                 }
  182.               "columnMappings": [
  183.                 {
  184.                   "queryColumn": "timestamp",
  185.                   "outputColumn": "timestamp"
  186.                 },
  187.                 {
  188.                   "queryColumn": "agent_category",
  189.                   "outputColumn": "agent_category"
  190.                 },
  191.                 {
  192.                   "queryColumn": "agent_type",
  193.                   "outputColumn": "agent_type"
  194.                 }
  195.               ],
  196.               "tuningConfig": {
  197.                 "maxNumWorkers": 1,
  198.                 "maxRowsInMemory": 100000,
  199.                 "rowsPerSegment": 3000000
  200.               }
  201.             },
  202.             "signature": [],
  203.             "maxWorkerCount": 1
  204.           },
  205.           "phase": "FINISHED",
  206.           "workerCount": 1,
  207.           "partitionCount": 1,
  208.           "startTime": "2022-09-14T22:12:31.602Z",
  209.           "duration": 5891
  210.         }
  211.       ],
  212.       "counters": {
  213.         "0": {
  214.           "0": {
  215.             "input0": {
  216.               "type": "channel",
  217.               "rows": [
  218.                 465346
  219.               ],
  220.               "files": [
  221.                 1
  222.               ],
  223.               "totalFiles": [
  224.                 1
  225.               ]
  226.             },
  227.             "output": {
  228.               "type": "channel",
  229.               "rows": [
  230.                 465346
  231.               ],
  232.               "bytes": [
  233.                 43694447
  234.               ],
  235.               "frames": [
  236.                 7
  237.               ]
  238.             },
  239.             "shuffle": {
  240.               "type": "channel",
  241.               "rows": [
  242.                 465346
  243.               ],
  244.               "bytes": [
  245.                 41835307
  246.               ],
  247.               "frames": [
  248.                 73
  249.               ]
  250.             },
  251.             "sortProgress": {
  252.               "type": "sortProgress",
  253.               "totalMergingLevels": 3,
  254.               "levelToTotalBatches": {
  255.                 "0": 1,
  256.                 "1": 1,
  257.                 "2": 1
  258.               },
  259.               "levelToMergedBatches": {
  260.                 "0": 1,
  261.                 "1": 1,
  262.                 "2": 1
  263.               },
  264.               "totalMergersForUltimateLevel": 1,
  265.               "progressDigest": 1
  266.             }
  267.           }
  268.         },
  269.         "1": {
  270.           "0": {
  271.             "input0": {
  272.               "type": "channel",
  273.               "rows": [
  274.                 465346
  275.               ],
  276.               "bytes": [
  277.                 41835307
  278.               ],
  279.               "frames": [
  280.                 73
  281.               ]
  282.             }
  283.           }
  284.         }
  285.       }
  286.     }
  287.   }
  288. }

The following table describes the response fields when you retrieve a report for a MSQ task engine using the /druid/indexer/v1/task/<taskId>/reports endpoint:

FieldDescription
multiStageQuery.taskIdController task ID.
multiStageQuery.payload.statusQuery status container.
multiStageQuery.payload.status.statusRUNNING, SUCCESS, or FAILED.
multiStageQuery.payload.status.startTimeStart time of the query in ISO format. Only present if the query has started running.
multiStageQuery.payload.status.durationMsMilliseconds elapsed after the query has started running. -1 denotes that the query hasn’t started running yet.
multiStageQuery.payload.status.errorReportError object. Only present if there was an error.
multiStageQuery.payload.status.errorReport.taskIdThe task that reported the error, if known. May be a controller task or a worker task.
multiStageQuery.payload.status.errorReport.hostThe hostname and port of the task that reported the error, if known.
multiStageQuery.payload.status.errorReport.stageNumberThe stage number that reported the error, if it happened during execution of a specific stage.
multiStageQuery.payload.status.errorReport.errorError object. Contains errorCode at a minimum, and may contain other fields as described in the error code table. Always present if there is an error.
multiStageQuery.payload.status.errorReport.error.errorCodeOne of the error codes from the . Always present if there is an error.
multiStageQuery.payload.status.errorReport.error.errorMessageUser-friendly error message. Not always present, even if there is an error.
multiStageQuery.payload.status.errorReport.exceptionStackTraceJava stack trace in string form, if the error was due to a server-side exception.
multiStageQuery.payload.stagesArray of query stages.
multiStageQuery.payload.stages[].stageNumberEach stage has a number that differentiates it from other stages.
multiStageQuery.payload.stages[].phaseEither NEW, READING_INPUT, POST_READING, RESULTS_COMPLETE, or FAILED. Only present if the stage has started.
multiStageQuery.payload.stages[].workerCountNumber of parallel tasks that this stage is running on. Only present if the stage has started.
multiStageQuery.payload.stages[].partitionCountNumber of output partitions generated by this stage. Only present if the stage has started and has computed its number of output partitions.
multiStageQuery.payload.stages[].startTimeStart time of this stage. Only present if the stage has started.
multiStageQuery.payload.stages[].durationThe number of milliseconds that the stage has been running. Only present if the stage has started.
multiStageQuery.payload.stages[].sortA boolean that is set to true if the stage does a sort as part of its execution.
multiStageQuery.payload.stages[].definitionThe object defining what the stage does.
multiStageQuery.payload.stages[].definition.idThe unique identifier of the stage.
multiStageQuery.payload.stages[].definition.inputArray of inputs that the stage has.
multiStageQuery.payload.stages[].definition.broadcastArray of input indexes that get broadcasted. Only present if there are inputs that get broadcasted.
multiStageQuery.payload.stages[].definition.processorAn object defining the processor logic.
multiStageQuery.payload.stages[].definition.signatureThe output signature of the stage.

Request

HTTP

curl

Python

  1. POST /druid/indexer/v1/task/<taskId>/shutdown
  1. # Make sure you replace `username`, `password`, `your-instance`, `port`, and `taskId` with the values for your deployment.curl --location --request POST 'https://<username>:<password>@<your-instance>:<port>/druid/indexer/v1/task/<taskId>/shutdown'

Response

  1. {
  3. }