Native batch ingestion

Apache Druid supports the following types of native batch indexing tasks:

• Parallel task indexing (index_parallel) that can run multiple indexing tasks concurrently. Parallel task works well for production ingestion tasks.
• Simple task indexing (index) that run a single indexing task at a time. Simple task indexing is suitable for development and test environments.

This topic covers the configuration for index_parallel ingestion specs.

For related information on batch indexing, see:

• Simple task indexing for index task configuration.
• Native batch input sources for a reference for inputSource configuration.
• Hadoop-based vs. native batch comparison table for a comparison of batch ingestion methods.
• Loading a file for a tutorial on native batch ingestion.

Submit an indexing task

To run either kind of native batch indexing task you can:

• Use the Load Data UI in the Druid console to define and submit an ingestion spec.
• Define an ingestion spec in JSON based upon the examples and reference topics for batch indexing. Then POST the ingestion spec to the Indexer API endpoint, /druid/indexer/v1/task, the Overlord service. Alternatively you can use the indexing script included with Druid at bin/post-index-task.

Parallel task indexing

The parallel task type index_parallel is a task for multi-threaded batch indexing. Parallel task indexing only relies on Druid resources. It does not depend on other external systems like Hadoop.

The index_parallel task is a supervisor task that orchestrates the whole indexing process. The supervisor task splits the input data and creates worker tasks to process the individual portions of data.

Druid issues the worker tasks to the Overlord. The overlord schedules and runs the workers on MiddleManagers or Indexers. After a worker task successfully processes the assigned input portion, it reports the resulting segment list to the supervisor task.

The supervisor task periodically checks the status of worker tasks. If a task fails, the supervisor retries the task until the number of retries reaches the configured limit. If all worker tasks succeed, it publishes the reported segments at once and finalizes ingestion.

The detailed behavior of the parallel task is different depending on the partitionsSpec. See partitionsSpec for more details.

Parallel tasks require:

• a splittable inputSource in the ioConfig. For a list of supported splittable input formats, see Splittable input sources.
• the maxNumConcurrentSubTasks greater than 1 in the tuningConfig. Otherwise tasks run sequentially. The index_parallel task reads each input file one by one and creates segments by itself.

Supported compression formats

Native batch ingestion supports the following compression formats:

• bz2
• gz
• xz
• zip
• sz (Snappy)
• zst (ZSTD).

Implementation considerations

This section covers implementation details to consider when you implement parallel task ingestion.

Volume control for worker tasks

You can control the amount of input data each worker task processes using different configurations depending on the phase in parallel ingestion.

• See partitionsSpec for details about how partitioning affects data volume for tasks.
• For the tasks that read data from the inputSource, you can set the Split hint spec in the tuningConfig.
• For the task that merge shuffled segments, you can set the totalNumMergeTasks in the tuningConfig.

Number of running tasks

The maxNumConcurrentSubTasks in the tuningConfig determines the number of concurrent worker tasks that run in parallel. The supervisor task checks the number of current running worker tasks and creates more if it’s smaller than maxNumConcurrentSubTasks regardless of the number of available task slots. This may affect to other ingestion performance. See Capacity planning section for more details.

Replacing or appending data

By default, batch ingestion replaces all data in the intervals in your granularitySpec‘ for any segment that it writes to. If you want to add to the segment instead, set the appendToExisting flag in the ioConfig. Batch ingestion only replaces data in segments where it actively adds data. If there are segments in the intervals for your granularitySpec that have do not have data from a task, they remain unchanged. If any existing segments partially overlap with the intervals in the granularitySpec, the portion of those segments outside the interval for the new spec remain visible.

Fully replacing existing segments using tombstones

You can set dropExisting flag in the ioConfig to true if you want the ingestion task to replace all existing segments that start and end within the intervals for your granularitySpec. This applies whether or not the new data covers all existing segments. dropExisting only applies when appendToExisting is false and the granularitySpec contains an interval. WARNING: this functionality is still in beta.

The following examples demonstrate when to set the dropExisting property to true in the ioConfig:

Consider an existing segment with an interval of 2020-01-01 to 2021-01-01 and YEAR segmentGranularity. You want to overwrite the whole interval of 2020-01-01 to 2021-01-01 with new data using the finer segmentGranularity of MONTH. If the replacement data does not have a record within every months from 2020-01-01 to 2021-01-01 Druid cannot drop the original YEAR segment even if it does include all the replacement data. Set dropExisting to true in this case to replace the original segment at YEAR segmentGranularity since you no longer need it.

Imagine you want to re-ingest or overwrite a datasource and the new data does not contain some time intervals that exist in the datasource. For example, a datasource contains the following data at MONTH segmentGranularity:

• January: 1 record
• February: 10 records
• March: 10 records

You want to re-ingest and overwrite with new data as follows:

• January: 0 records
• February: 10 records
• March: 9 records

Unless you set dropExisting to true, the result after ingestion with overwrite using the same MONTH segmentGranularity would be:

• January: 1 record
• February: 10 records
• March: 9 records

This may not be what it is expected since the new data has 0 records for January. Set dropExisting to true to replace the unneeded January segment with a tombstone.

Parallel indexing example

The following example illustrates the configuration for a parallel indexing task:

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "wikipedia_parallel_index_test",
      "timestampSpec": {
        "column": "timestamp"
      },
      "dimensionsSpec": {
        "dimensions": [
          "country",
          "page",
          "language",
          "user",
          "unpatrolled",
          "newPage",
          "robot",
          "anonymous",
          "namespace",
          "continent",
          "region",
          "city"
        ]
      },
      "metricsSpec": [
        {
          "type": "count",
              "name": "count"
            },
            {
              "type": "doubleSum",
              "name": "added",
              "fieldName": "added"
            },
            {
              "type": "doubleSum",
              "name": "deleted",
              "fieldName": "deleted"
            },
            {
              "type": "doubleSum",
              "name": "delta",
              "fieldName": "delta"
            }
        ],
        "granularitySpec": {
          "segmentGranularity": "DAY",
          "queryGranularity": "second",
          "intervals" : [ "2013-08-31/2013-09-02" ]
        }
    },
    "ioConfig": {
        "type": "index_parallel",
        "inputSource": {
          "type": "local",
          "baseDir": "examples/indexing/",
          "filter": "wikipedia_index_data*"
        },
        "inputFormat": {
          "type": "json"
        }
    },
    "tuningConfig": {
        "type": "index_parallel",
        "partitionsSpec": {
             "type": "single_dim",
             "partitionDimension": "country",
             "targetRowsPerSegment": 5000000
      },
        "maxNumConcurrentSubTasks": 2
    }
  }
}

Parallel indexing configuration

The following table defines the primary sections of the input spec:

dataSchema

This field is required. In general, it defines the way that Druid will store your data: the primary timestamp column, the dimensions, metrics, and any transformations. For an overview, see Ingestion Spec DataSchema.

When defining the granularitySpec for index parallel, consider the defining intervals explicitly if you know the time range of the data. This way locking failure happens faster and Druid won’t accidentally replace data outside the interval range some rows contain unexpected timestamps. The reasoning is as follows:

• If you explicitly define intervals, batch ingestion locks all intervals specified when it starts up. Problems with locking become evident quickly when multiple ingestion or indexing tasks try to obtain a lock on the same interval. For example, if a Kafka ingestion task tries to obtain a lock on a locked interval causing the ingestion task fail. Furthermore, if there are rows outside the specified intervals, Druid drops them, avoiding conflict with unexpected intervals.
• If you do not define intervals, batch ingestion locks each interval when the interval is discovered. In this case if the task overlaps with a higher-priority task, issues with conflicting locks occur later in the ingestion process. Also if the source data includes rows with unexpected timestamps, they may caused unexpected locking of intervals.

ioConfig

tuningConfig

The tuningConfig is optional and default parameters will be used if no tuningConfig is specified. See below for more details.

Split Hint Spec

The split hint spec is used to help the supervisor task divide input sources. Each worker task processes a single input division. You can control the amount of data each worker task reads during the first phase.

Size-based Split Hint Spec

The size-based split hint spec affects all splittable input sources except for the HTTP input source and SQL input source.

Segments Split Hint Spec

The segments split hint spec is used only for DruidInputSource and legacy IngestSegmentFirehose.

partitionsSpec

The primary partition for Druid is time. You can define a secondary partitioning method in the partitions spec. Use the partitionsSpec type that applies for your rollup method. For perfect rollup, you can use:

• hashed partitioning based on the hash value of specified dimensions for each row
• single_dim based on ranges of values for a single dimension
• range based on ranges of values of multiple dimensions.

For best-effort rollup, use dynamic.

For an overview, see Partitioning.

The partitionsSpec types have different characteristics.

Dynamic partitioning

With the Dynamic partitioning, the parallel index task runs in a single phase: it spawns multiple worker tasks (type single_phase_sub_task), each of which creates segments. How the worker task creates segments:

• Whenever the number of rows in the current segment exceeds maxRowsPerSegment.
• When the total number of rows in all segments across all time chunks reaches to maxTotalRows. At this point the task pushes all segments created so far to the deep storage and creates new ones.

Hash-based partitioning

The Parallel task with hash-based partitioning is similar to MapReduce. The task runs in up to 3 phases: partial dimension cardinality, partial segment generation and partial segment merge.

• The partial dimension cardinality phase is an optional phase that only runs if numShards is not specified. The Parallel task splits the input data and assigns them to worker tasks based on the split hint spec. Each worker task (type partial_dimension_cardinality) gathers estimates of partitioning dimensions cardinality for each time chunk. The Parallel task will aggregate these estimates from the worker tasks and determine the highest cardinality across all of the time chunks in the input data, dividing this cardinality by targetRowsPerSegment to automatically determine numShards.
• In the partial segment generation phase, just like the Map phase in MapReduce, the Parallel task splits the input data based on the split hint spec and assigns each split to a worker task. Each worker task (type partial_index_generate) reads the assigned split, and partitions rows by the time chunk from segmentGranularity (primary partition key) in the granularitySpec and then by the hash value of partitionDimensions (secondary partition key) in the partitionsSpec. The partitioned data is stored in local storage of the middleManager or the indexer.
• The partial segment merge phase is similar to the Reduce phase in MapReduce. The Parallel task spawns a new set of worker tasks (type partial_index_generic_merge) to merge the partitioned data created in the previous phase. Here, the partitioned data is shuffled based on the time chunk and the hash value of partitionDimensions to be merged; each worker task reads the data falling in the same time chunk and the same hash value from multiple MiddleManager/Indexer processes and merges them to create the final segments. Finally, they push the final segments to the deep storage at once.

Hash partition function

In hash partitioning, the partition function is used to compute hash of partition dimensions. The partition dimension values are first serialized into a byte array as a whole, and then the partition function is applied to compute hash of the byte array. Druid currently supports only one partition function.

Single-dimension range partitioning

Single dimension range partitioning is currently not supported in the sequential mode of the Parallel task.

The Parallel task will use one subtask when you set maxNumConcurrentSubTasks to 1.

When you use this technique to partition your data, segment sizes may be unequally distributed if the data in your partitionDimension is also unequally distributed. Therefore, to avoid imbalance in data layout, review the distribution of values in your source data before deciding on a partitioning strategy.

For segment pruning to be effective and translate into better query performance, you must use the partitionDimension at query time. You can concatenate values from multiple dimensions into a new dimension to use as the partitionDimension. In this case, you must use that new dimension in your native filter WHERE clause.

With single-dim partitioning, the Parallel task runs in 3 phases, i.e., partial dimension distribution, partial segment generation, and partial segment merge. The first phase is to collect some statistics to find the best partitioning and the other 2 phases are to create partial segments and to merge them, respectively, as in hash-based partitioning.

• In the partial dimension distribution phase, the Parallel task splits the input data and assigns them to worker tasks based on the split hint spec. Each worker task (type partial_dimension_distribution) reads the assigned split and builds a histogram for partitionDimension. The Parallel task collects those histograms from worker tasks and finds the best range partitioning based on partitionDimension to evenly distribute rows across partitions. Note that either targetRowsPerSegment or maxRowsPerSegment will be used to find the best partitioning.
• In the partial segment generation phase, the Parallel task spawns new worker tasks (type partial_range_index_generate) to create partitioned data. Each worker task reads a split created as in the previous phase, partitions rows by the time chunk from the segmentGranularity (primary partition key) in the granularitySpec and then by the range partitioning found in the previous phase. The partitioned data is stored in local storage of the middleManager or the indexer.
• In the partial segment merge phase, the parallel index task spawns a new set of worker tasks (type partial_index_generic_merge) to merge the partitioned data created in the previous phase. Here, the partitioned data is shuffled based on the time chunk and the value of partitionDimension; each worker task reads the segments falling in the same partition of the same range from multiple MiddleManager/Indexer processes and merges them to create the final segments. Finally, they push the final segments to the deep storage.

Because the task with single-dimension range partitioning makes two passes over the input in partial dimension distribution and partial segment generation phases, the task may fail if the input changes in between the two passes.

Multi-dimension range partitioning

Multiple dimension (multi-dimension) range partitioning is an experimental feature. Multi-dimension range partitioning is currently not supported in the sequential mode of the Parallel task.

When you use multi-dimension partitioning for your data, Druid is able to distribute segment sizes more evenly than with single dimension partitioning.

For segment pruning to be effective and translate into better query performance, you must include the first of your partitionDimensions in the WHERE clause at query time. For example, given the following partitionDimensions:

 "partitionsSpec": {
        "type": "range",
        "partitionDimensions":["coutryName","cityName"],
        "targetRowsPerSegment" : 5000
}

Use “countryName” or both “countryName” and “cityName” in the WHERE clause of your query to take advantage of the performance benefits from multi-dimension partitioning.

HTTP status endpoints

The supervisor task provides some HTTP endpoints to get running status.

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/mode

Returns ‘parallel’ if the indexing task is running in parallel. Otherwise, it returns ‘sequential’.

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/phase

Returns the name of the current phase if the task running in the parallel mode.

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/progress

Returns the estimated progress of the current phase if the supervisor task is running in the parallel mode.

An example of the result is

{
  "running":10,
  "succeeded":0,
  "failed":0,
  "complete":0,
  "total":10,
  "estimatedExpectedSucceeded":10
}

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/subtasks/running

Returns the task IDs of running worker tasks, or an empty list if the supervisor task is running in the sequential mode.

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/subtaskspecs

Returns all worker task specs, or an empty list if the supervisor task is running in the sequential mode.

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/subtaskspecs/running

Returns running worker task specs, or an empty list if the supervisor task is running in the sequential mode.

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/subtaskspecs/complete

Returns complete worker task specs, or an empty list if the supervisor task is running in the sequential mode.

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/subtaskspec/{SUB_TASK_SPEC_ID}

Returns the worker task spec of the given id, or HTTP 404 Not Found error if the supervisor task is running in the sequential mode.

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/subtaskspec/{SUB_TASK_SPEC_ID}/state

Returns the state of the worker task spec of the given id, or HTTP 404 Not Found error if the supervisor task is running in the sequential mode. The returned result contains the worker task spec, a current task status if exists, and task attempt history.

An example of the result is

{
  "spec": {
    "id": "index_parallel_lineitem_2018-04-20T22:12:43.610Z_2",
    "groupId": "index_parallel_lineitem_2018-04-20T22:12:43.610Z",
    "supervisorTaskId": "index_parallel_lineitem_2018-04-20T22:12:43.610Z",
    "context": null,
    "inputSplit": {
      "split": "/path/to/data/lineitem.tbl.5"
    },
    "ingestionSpec": {
      "dataSchema": {
        "dataSource": "lineitem",
        "timestampSpec": {
          "column": "l_shipdate",
          "format": "yyyy-MM-dd"
        },
        "dimensionsSpec": {
          "dimensions": [
            "l_orderkey",
            "l_partkey",
            "l_suppkey",
            "l_linenumber",
            "l_returnflag",
            "l_linestatus",
            "l_shipdate",
            "l_commitdate",
            "l_receiptdate",
            "l_shipinstruct",
            "l_shipmode",
            "l_comment"
          ]
        },
        "metricsSpec": [
          {
            "type": "count",
            "name": "count"
          },
          {
            "type": "longSum",
            "name": "l_quantity",
            "fieldName": "l_quantity",
            "expression": null
          },
          {
            "type": "doubleSum",
            "name": "l_extendedprice",
            "fieldName": "l_extendedprice",
            "expression": null
          },
          {
            "type": "doubleSum",
            "name": "l_discount",
            "fieldName": "l_discount",
            "expression": null
          },
          {
            "type": "doubleSum",
            "name": "l_tax",
            "fieldName": "l_tax",
            "expression": null
          }
        ],
        "granularitySpec": {
          "type": "uniform",
          "segmentGranularity": "YEAR",
          "queryGranularity": {
            "type": "none"
          },
          "rollup": true,
          "intervals": [
            "1980-01-01T00:00:00.000Z/2020-01-01T00:00:00.000Z"
          ]
        },
        "transformSpec": {
          "filter": null,
          "transforms": []
        }
      },
      "ioConfig": {
        "type": "index_parallel",
        "inputSource": {
          "type": "local",
          "baseDir": "/path/to/data/",
          "filter": "lineitem.tbl.5"
        },
        "inputFormat": {
          "format": "tsv",
          "delimiter": "|",
          "columns": [
            "l_orderkey",
            "l_partkey",
            "l_suppkey",
            "l_linenumber",
            "l_quantity",
            "l_extendedprice",
            "l_discount",
            "l_tax",
            "l_returnflag",
            "l_linestatus",
            "l_shipdate",
            "l_commitdate",
            "l_receiptdate",
            "l_shipinstruct",
            "l_shipmode",
            "l_comment"
          ]
        },
        "appendToExisting": false,
        "dropExisting": false
      },
      "tuningConfig": {
        "type": "index_parallel",
        "maxRowsPerSegment": 5000000,
        "maxRowsInMemory": 1000000,
        "maxTotalRows": 20000000,
        "numShards": null,
        "indexSpec": {
          "bitmap": {
            "type": "roaring"
          },
          "dimensionCompression": "lz4",
          "metricCompression": "lz4",
          "longEncoding": "longs"
        },
        "indexSpecForIntermediatePersists": {
          "bitmap": {
            "type": "roaring"
          },
          "dimensionCompression": "lz4",
          "metricCompression": "lz4",
          "longEncoding": "longs"
        },
        "maxPendingPersists": 0,
        "reportParseExceptions": false,
        "pushTimeout": 0,
        "segmentWriteOutMediumFactory": null,
        "maxNumConcurrentSubTasks": 4,
        "maxRetry": 3,
        "taskStatusCheckPeriodMs": 1000,
        "chatHandlerTimeout": "PT10S",
        "chatHandlerNumRetries": 5,
        "logParseExceptions": false,
        "maxParseExceptions": 2147483647,
        "maxSavedParseExceptions": 0,
        "forceGuaranteedRollup": false
      }
    }
  },
  "currentStatus": {
    "id": "index_sub_lineitem_2018-04-20T22:16:29.922Z",
    "type": "index_sub",
    "createdTime": "2018-04-20T22:16:29.925Z",
    "queueInsertionTime": "2018-04-20T22:16:29.929Z",
    "statusCode": "RUNNING",
    "duration": -1,
    "location": {
      "host": null,
      "port": -1,
      "tlsPort": -1
    },
    "dataSource": "lineitem",
    "errorMsg": null
  },
  "taskHistory": []
}

• http://{PEON_IP}:{PEON_PORT}/druid/worker/v1/chat/{SUPERVISOR_TASK_ID}/subtaskspec/{SUB_TASK_SPEC_ID}/history

Returns the task attempt history of the worker task spec of the given id, or HTTP 404 Not Found error if the supervisor task is running in the sequential mode.

Segment pushing modes

While ingesting data using the parallel task indexing, Druid creates segments from the input data and pushes them. For segment pushing, the parallel task index supports the following segment pushing modes based upon your type of rollup:

• Bulk pushing mode: Used for perfect rollup. Druid pushes every segment at the very end of the index task. Until then, Druid stores created segments in memory and local storage of the service running the index task. This mode can cause problems if you have limited storage capacity, and is not recommended to use in production. To enable bulk pushing mode, set forceGuaranteedRollup in your TuningConfig. You cannot use bulk pushing with appendToExisting in your IOConfig.
• Incremental pushing mode: Used for best-effort rollup. Druid pushes segments are incrementally during the course of the indexing task. The index task collects data and stores created segments in the memory and disks of the services running the task until the total number of collected rows exceeds maxTotalRows. At that point the index task immediately pushes all segments created up until that moment, cleans up pushed segments, and continues to ingest the remaining data.

Capacity planning

The supervisor task can create up to maxNumConcurrentSubTasks worker tasks no matter how many task slots are currently available. As a result, total number of tasks which can be run at the same time is (maxNumConcurrentSubTasks + 1) (including the supervisor task). Please note that this can be even larger than total number of task slots (sum of the capacity of all workers). If maxNumConcurrentSubTasks is larger than n (available task slots), then maxNumConcurrentSubTasks tasks are created by the supervisor task, but only n tasks would be started. Others will wait in the pending state until any running task is finished.

If you are using the Parallel Index Task with stream ingestion together, we would recommend to limit the max capacity for batch ingestion to prevent stream ingestion from being blocked by batch ingestion. Suppose you have t Parallel Index Tasks to run at the same time, but want to limit the max number of tasks for batch ingestion to b. Then, (sum of maxNumConcurrentSubTasks of all Parallel Index Tasks + t (for supervisor tasks)) must be smaller than b.

If you have some tasks of a higher priority than others, you may set their maxNumConcurrentSubTasks to a higher value than lower priority tasks. This may help the higher priority tasks to finish earlier than lower priority tasks by assigning more task slots to them.

Splittable input sources

Use the inputSource object to define the location where your index can read data. Only the native parallel task and simple task support the input source.

For details on available input sources see:

• S3 input source (s3) reads data from AWS S3 storage.
• Google Cloud Storage input source (gs) reads data from Google Cloud Storage.
• Azure input source (azure) reads data from Azure Blob Storage and Azure Data Lake.
• HDFS input source (hdfs) reads data from HDFS storage.
• HTTP input Source (http) reads data from HTTP servers.
• Inline input Source reads data you paste into the Druid console.
• Local input Source (local) reads data from local storage.
• Druid input Source (druid) reads data from a Druid datasource.
• SQL input Source (sql) reads data from a RDBMS source.

For information on how to combine input sources, see Combining input sources.