Configuration reference

This page documents all of the configuration properties for each Druid service type.

Recommended Configuration File Organization

A recommended way of organizing Druid configuration files can be seen in the conf directory in the Druid package root, shown below:

$ ls -R conf
druid
conf/druid:
_common       broker        coordinator   historical    middleManager overlord
conf/druid/_common:
common.runtime.properties log4j2.xml
conf/druid/broker:
jvm.config         runtime.properties
conf/druid/coordinator:
jvm.config         runtime.properties
conf/druid/historical:
jvm.config         runtime.properties
conf/druid/middleManager:
jvm.config         runtime.properties
conf/druid/overlord:
jvm.config         runtime.properties

Each directory has a runtime.properties file containing configuration properties for the specific Druid process corresponding to the directory (e.g., historical).

The jvm.config files contain JVM flags such as heap sizing properties for each service.

Common properties shared by all services are placed in _common/common.runtime.properties.

Configuration Interpolation

Configuration values can be interpolated from System Properties, Environment Variables, or local files. Below is an example of how this can be used:

druid.metadata.storage.type=${env:METADATA_STORAGE_TYPE}
druid.processing.tmpDir=${sys:java.io.tmpdir}
druid.segmentCache.locations=${file:UTF-8:/config/segment-cache-def.json}

Interpolation is also recursive so you can do:

druid.segmentCache.locations=${file:UTF-8:${env:SEGMENT_DEF_LOCATION}}

If the property is not set an exception will be thrown on startup, but a default can be provided if desired. Setting a default value will not work with file interpolation as an exception will be thrown if the file does not exist.

druid.metadata.storage.type=${env:METADATA_STORAGE_TYPE:-mysql}
druid.processing.tmpDir=${sys:java.io.tmpdir:-/tmp}

If you need to set a variable that is wrapped by ${...} but do not want it to be interpolated you can escape it by adding another $. For example:

config.name=$${value}

Common Configurations

The properties under this section are common configurations that should be shared across all Druid services in a cluster.

JVM Configuration Best Practices

There are four JVM parameters that we set on all of our processes:

• -Duser.timezone=UTC: This sets the default timezone of the JVM to UTC. We always set this and do not test with other default timezones, so local timezones might work, but they also might uncover weird and interesting bugs. To issue queries in a non-UTC timezone, see query granularities

• -Dfile.encoding=UTF-8 This is similar to timezone, we test assuming UTF-8. Local encodings might work, but they also might result in weird and interesting bugs.

• -Djava.io.tmpdir=<a path> Various parts of Druid use temporary files to interact with the file system. These files can become quite large. This means that systems that have small /tmp directories can cause problems for Druid. Therefore, set the JVM tmp directory to a location with ample space.

  Also consider the following when configuring the JVM tmp directory:

  • The temp directory should not be volatile tmpfs.
  • This directory should also have good read and write speed.
  • Avoid NFS mount.
  • The org.apache.druid.java.util.metrics.SysMonitor requires execute privileges on files in java.io.tmpdir. If you are using the system monitor, do not set java.io.tmpdir to noexec.
• -Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager This allows log4j2 to handle logs for non-log4j2 components (like jetty) which use standard java logging.

Extensions

Many of Druid’s external dependencies can be plugged in as modules. Extensions can be provided using the following configs:

Modules

ZooKeeper

We recommend just setting the base ZK path and the ZK service host, but all ZK paths that Druid uses can be overwritten to absolute paths.

ZooKeeper Behavior

Path Configuration

Druid interacts with ZK through a set of standard path configurations. We recommend just setting the base ZK path, but all ZK paths that Druid uses can be overwritten to absolute paths.

The indexing service also uses its own set of paths. These configs can be included in the common configuration.

If druid.zk.paths.base and druid.zk.paths.indexer.base are both set, and none of the other druid.zk.paths.* or druid.zk.paths.indexer.* values are set, then the other properties will be evaluated relative to their respective base. For example, if druid.zk.paths.base is set to /druid1 and druid.zk.paths.indexer.base is set to /druid2 then druid.zk.paths.announcementsPath will default to /druid1/announcements while druid.zk.paths.indexer.announcementsPath will default to /druid2/announcements.

The following path is used for service discovery. It is not affected by druid.zk.paths.base and must be specified separately.

TLS

General Configuration

Although not recommended but both HTTP and HTTPS connectors can be enabled at a time and respective ports are configurable using druid.plaintextPort and druid.tlsPort properties on each process. Please see Configuration section of individual processes to check the valid and default values for these ports.

Jetty Server TLS Configuration

Druid uses Jetty as an embedded web server. To learn more about TLS/SSL, certificates, and related concepts in Jetty, including explanations of the configuration settings below, see “Configuring SSL/TLS KeyStores” in the Jetty Operations Guide.

For information about TLS/SSL support in Java in general, see the Java Secure Socket Extension (JSSE) Reference Guide. The Java Cryptography Architecture Standard Algorithm Name Documentation for JDK 8 lists all possible values for the following properties, among others provided by the Java implementation.

Following table contains non-mandatory advanced configuration options, use caution.

Internal Client TLS Configuration (requires simple-client-sslcontext extension)

These properties apply to the SSLContext that will be provided to the internal HTTP client that Druid services use to communicate with each other. These properties require the simple-client-sslcontext extension to be loaded. Without it, Druid services will be unable to communicate with each other when TLS is enabled.

This document lists all the possible values for the above mentioned configs among others provided by Java implementation.

Authentication and Authorization

For more information, please see Authentication and Authorization.

For configuration options for specific auth extensions, please refer to the extension documentation.

Startup Logging

All processes can log debugging information on startup.

Note that some sensitive information may be logged if these settings are enabled.

Request Logging

All processes that can serve queries can also log the query requests they see. Broker processes can additionally log the SQL requests (both from HTTP and JDBC) they see. For an example of setting up request logging, see Request logging.

Note that you can enable sending all the HTTP requests to log by setting org.apache.druid.jetty.RequestLog to the DEBUG level. See Logging for more information.

File request logging

The file request logger stores daily request logs on disk.

The format of request logs is TSV, one line per requests, with five fields: timestamp, remote_addr, native_query, query_context, sql_query.

For native JSON request, the sql_query field is empty. Example

2019-01-14T10:00:00.000Z        127.0.0.1   {"queryType":"topN","dataSource":{"type":"table","name":"wikiticker"},"virtualColumns":[],"dimension":{"type":"LegacyDimensionSpec","dimension":"page","outputName":"page","outputType":"STRING"},"metric":{"type":"LegacyTopNMetricSpec","metric":"count"},"threshold":10,"intervals":{"type":"LegacySegmentSpec","intervals":["2015-09-12T00:00:00.000Z/2015-09-13T00:00:00.000Z"]},"filter":null,"granularity":{"type":"all"},"aggregations":[{"type":"count","name":"count"}],"postAggregations":[],"context":{"queryId":"74c2d540-d700-4ebd-b4a9-3d02397976aa"},"descending":false}    {"query/time":100,"query/bytes":800,"success":true,"identity":"user1"}

For SQL query request, the native_query field is empty. Example

2019-01-14T10:00:00.000Z        127.0.0.1       {"sqlQuery/time":100, "sqlQuery/planningTimeMs":10, "sqlQuery/bytes":600, "success":true, "identity":"user1"}  {"query":"SELECT page, COUNT(*) AS Edits FROM wikiticker WHERE TIME_IN_INTERVAL(\"__time\", '2015-09-12/2015-09-13') GROUP BY page ORDER BY Edits DESC LIMIT 10","context":{"sqlQueryId":"c9d035a0-5ffd-4a79-a865-3ffdadbb5fdd","nativeQueryIds":"[490978e4-f5c7-4cf6-b174-346e63cf8863]"}}

Emitter request logging

The emitter request logger emits every request to the external location specified in the emitter configuration.

SLF4J request logging

The slf4j request logger logs every request using SLF4J. It serializes native queries into JSON in the log message regardless of the SLF4J format specification. Requests are logged under the class org.apache.druid.server.log.LoggingRequestLogger.

For a native query, the following MDC fields are populated when setMDC is true:

Filtered request logging

The filtered request logger filters requests based on the query type or how long a query takes to complete. For native queries, the logger only logs requests when the query/time metric exceeds the threshold provided in queryTimeThresholdMs. For SQL queries, it only logs requests when the sqlQuery/time metric exceeds threshold provided in sqlQueryTimeThresholdMs. See Metrics for more details on query metrics.

Requests that meet the threshold are logged using the request logger type set in druid.request.logging.delegate.type.

Composing request logging

The composing request logger emits request logs to multiple request loggers.

Switching request logging

The switching request logger routes native query request logs to one request logger and SQL query request logs to another request logger.

Audit Logging

Coordinator and Overlord log changes to lookups, segment load/drop rules, dynamic configuration changes for auditing

Enabling Metrics

You can configure Druid processes to emit metrics regularly from a number of monitors via emitters.

Metrics monitors

Metric monitoring is an essential part of Druid operations. The following monitors are available:

For example, you might configure monitors on all processes for system and JVM information within common.runtime.properties as follows:

druid.monitoring.monitors=["org.apache.druid.java.util.metrics.SysMonitor","org.apache.druid.java.util.metrics.JvmMonitor"]

You can override cluster-wide configuration by amending the runtime.properties of individual processes.

Metrics emitters

There are several emitters available:

• noop (default) disables metric emission.
• logging emits logs using Log4j2.
• http sends POST requests of JSON events.
• parametrized operates like the http emitter but fine-tunes the recipient URL based on the event feed.
• composing initializes multiple emitter modules.
• graphite emits metrics to a Graphite Carbon service.
• switching initializes and emits to multiple emitter modules based on the event feed.

Logging Emitter Module

The use this emitter module, set druid.emitter=logging. The logging emitter uses a Log4j2 logger named druid.emitter.logging.loggerClass to emit events. Each event is logged as a single json object with a Marker as the feed of the event. Users may wish to edit the log4j config to route these logs to different sources based on the feed of the event.

HTTP Emitter Module

HTTP Emitter Module TLS Overrides

By default, when sending events to a TLS-enabled receiver, the HTTP Emitter uses an SSLContext obtained from the process described at Druid’s internal communication over TLS, i.e., the same SSLContext that would be used for internal communications between Druid processes.

In some use cases it may be desirable to have the HTTP Emitter use its own separate truststore configuration. For example, there may be organizational policies that prevent the TLS-enabled metrics receiver’s certificate from being added to the same truststore used by Druid’s internal HTTP client.

The following properties allow the HTTP Emitter to use its own truststore configuration when building its SSLContext.

Parametrized HTTP Emitter Module

The parametrized emitter takes the same configs as the http emitter using the prefix druid.emitter.parametrized.httpEmitting.. For example:

• druid.emitter.parametrized.httpEmitting.flushMillis
• druid.emitter.parametrized.httpEmitting.flushCount
• druid.emitter.parametrized.httpEmitting.ssl.trustStorePath

Do not specify recipientBaseUrl with the parametrized emitter. Instead use recipientBaseUrlPattern described in the table below.

Composing Emitter Module

Graphite Emitter

To use graphite as emitter set druid.emitter=graphite. For configuration details, see Graphite emitter for the Graphite emitter Druid extension.

Switching Emitter

To use switching as emitter set druid.emitter=switching.

Metadata storage

These properties specify the JDBC connection and other configuration around the metadata storage. The only processes that connect to the metadata storage with these properties are the Coordinator and Overlord.

Deep storage

The configurations concern how to push and pull Segments from deep storage.

Local Deep Storage

Local deep storage uses the local filesystem.

Noop Deep Storage

This deep storage doesn’t do anything. There are no configs.

S3 Deep Storage

This deep storage is used to interface with Amazon’s S3. Note that the druid-s3-extensions extension must be loaded. The below table shows some important configurations for S3. See S3 Deep Storage for full configurations.

HDFS Deep Storage

This deep storage is used to interface with HDFS. Note that the druid-hdfs-storage extension must be loaded.

Cassandra Deep Storage

This deep storage is used to interface with Cassandra. Note that the druid-cassandra-storage extension must be loaded.

Ingestion Security Configuration

HDFS input source

You can set the following property to specify permissible protocols for the HDFS input source.

HTTP input source

You can set the following property to specify permissible protocols for the HTTP input source.

External Data Access Security Configuration

JDBC Connections to External Databases

You can use the following properties to specify permissible JDBC options for:

• SQL input source
• globally cached JDBC lookups
• JDBC Data Fetcher for per-lookup caching.

These properties do not apply to metadata storage connections.

Task Logging

You can use the druid.indexer configuration to set a long-term storage location for task log files, and to set a retention policy.

For more information about ingestion tasks and the process of generating logs, see the task reference.

Log Long-term Storage

File Task Logs

Store task logs in the local filesystem.

S3 Task Logs

Store task logs in S3. Note that the druid-s3-extensions extension must be loaded.

Azure Blob Store Task Logs

Store task logs in Azure Blob Store.

Note: The druid-azure-extensions extension must be loaded, and this uses the same storage account as the deep storage module for azure.

Google Cloud Storage Task Logs

Store task logs in Google Cloud Storage.

Note: The druid-google-extensions extension must be loaded, and this uses the same storage settings as the deep storage module for google.

HDFS Task Logs

Store task logs in HDFS. Note that the druid-hdfs-storage extension must be loaded.

Log Retention Policy

API error response

You can configure Druid API error responses to hide internal information like the Druid class name, stack trace, thread name, servlet name, code, line/column number, host, or IP address.

Error response transform strategy

You can use an error response transform strategy to transform error responses from within Druid services to hide internal information. When you specify an error response transform strategy other than none, Druid transforms the error responses from Druid services as follows:

• For any query API that fails in the Router service, Druid sets the fields errorClass and host to null. Druid applies the transformation strategy to the errorMessage field.
• For any SQL query API that fails, for example POST /druid/v2/sql/..., Druid sets the fields errorClass and host to null. Druid applies the transformation strategy to the errorMessage field.
• For any JDBC related exceptions, Druid will turn all checked exceptions into QueryInterruptedException otherwise druid will attempt to keep the exception as the same type. For example if the original exception isn’t owned by Druid it will become QueryInterruptedException. Druid applies the transformation strategy to the errorMessage field.

No error response transform strategy

In this mode, Druid leaves error responses from underlying services unchanged and returns the unchanged errors to the API client. This is the default Druid error response mode. To explicitly enable this strategy, set druid.server.http.errorResponseTransform.strategy to “none”.

Allowed regular expression error response transform strategy

In this mode, Druid validates the error responses from underlying services against a list of regular expressions. Only error messages that match a configured regular expression are returned. To enable this strategy, set druid.server.http.errorResponseTransform.strategy to allowedRegex.

For example, consider the following error response:

{"error":"Plan validation failed","errorMessage":"org.apache.calcite.runtime.CalciteContextException: From line 1, column 15 to line 1, column 38: Object 'nonexistent-datasource' not found","errorClass":"org.apache.calcite.tools.ValidationException","host":null}

If druid.server.http.errorResponseTransform.allowedRegex is set to [], Druid transforms the query error response to the following:

{"error":"Plan validation failed","errorMessage":null,"errorClass":null,"host":null}

On the other hand, if druid.server.http.errorResponseTransform.allowedRegex is set to [".*CalciteContextException.*"] then Druid transforms the query error response to the following:

{"error":"Plan validation failed","errorMessage":"org.apache.calcite.runtime.CalciteContextException: From line 1, column 15 to line 1, column 38: Object 'nonexistent-datasource' not found","errorClass":null,"host":null}

Overlord Discovery

This config is used to find the Overlord using Curator service discovery. Only required if you are actually running an Overlord.

Coordinator Discovery

This config is used to find the Coordinator using Curator service discovery. This config is used by the realtime indexing processes to get information about the segments loaded in the cluster.

Announcing Segments

You can configure how to announce and unannounce Znodes in ZooKeeper (using Curator). For normal operations you do not need to override any of these configs.

Batch Data Segment Announcer

In current Druid, multiple data segments may be announced under the same Znode.

If you want to turn off the batch data segment announcer, you can add a property to skip announcing segments. You do not want to enable this config if you have any services using batch for druid.serverview.type

JavaScript

Druid supports dynamic runtime extension through JavaScript functions. This functionality can be configured through the following properties.

info

JavaScript-based functionality is disabled by default. Please refer to the Druid JavaScript programming guide for guidelines about using Druid’s JavaScript functionality, including instructions on how to enable it.

Double Column storage

Prior to version 0.13.0, Druid’s storage layer used a 32-bit float representation to store columns created by the doubleSum, doubleMin, and doubleMax aggregators at indexing time. Starting from version 0.13.0 the default will be 64-bit floats for Double columns. Using 64-bit representation for double column will lead to avoid precision loss at the cost of doubling the storage size of such columns. To keep the old format set the system-wide property druid.indexing.doubleStorage=float. You can also use floatSum, floatMin and floatMax to use 32-bit float representation. Support for 64-bit floating point columns was released in Druid 0.11.0, so if you use this feature then older versions of Druid will not be able to read your data segments.

SQL compatible null handling

Prior to version 0.13.0, Druid string columns treated '' and null values as interchangeable, and numeric columns were unable to represent null values, coercing null to 0. Druid 0.13.0 introduced a mode which enabled SQL compatible null handling, allowing string columns to distinguish empty strings from nulls, and numeric columns to contain null rows.

This mode does have a storage size and query performance cost, see segment documentation for more details.

HTTP Client

All Druid components can communicate with each other over HTTP.

Common endpoints Configuration

This section contains the configuration options for endpoints that are supported by all processes.

Master Server

This section contains the configuration options for the processes that reside on Master servers (Coordinators and Overlords) in the suggested three-server configuration.

Coordinator

For general Coordinator Process information, see here.

Static Configuration

These Coordinator static configurations can be defined in the coordinator/runtime.properties file.

Coordinator Process Config

Coordinator Operation

Metadata Management

Segment Management

Additional config when “http” loadqueuepeon is used

Metadata Retrieval

Dynamic Configuration

The Coordinator has dynamic configurations to tune certain behavior on the fly, without requiring a service restart.

It is recommended that you use the web console to configure these parameters. However, if you need to do it via HTTP, the JSON object can be submitted to the Coordinator via a POST request at:

http://<COORDINATOR_IP>:<PORT>/druid/coordinator/v1/config

Optional Header Parameters for auditing the config change can also be specified.

A sample Coordinator dynamic config JSON object is shown below:

{
  "millisToWaitBeforeDeleting": 900000,
  "mergeBytesLimit": 100000000,
  "mergeSegmentsLimit" : 1000,
  "maxSegmentsToMove": 5,
  "replicantLifetime": 15,
  "replicationThrottleLimit": 10,
  "killDataSourceWhitelist": ["wikipedia", "testDatasource"],
  "decommissioningNodes": ["localhost:8182", "localhost:8282"],
  "decommissioningMaxPercentOfMaxSegmentsToMove": 70,
  "pauseCoordination": false,
  "replicateAfterLoadTimeout": false,
  "maxNonPrimaryReplicantsToLoad": 2147483647
}

Issuing a GET request at the same URL will return the spec that is currently in place. A description of the config setup spec is shown below.

Smart segment loading

The smartSegmentLoading mode simplifies Coordinator configuration for segment loading and balancing. In this mode, the Coordinator does not require the user to provide values of the following parameters and computes them automatically instead.

info

If you enable smartSegmentLoading mode and provide values for the following properties, Druid ignores your values. The Coordinator computes them automatically. The computed values are based on the current state of the cluster and are designed to optimize Coordinator performance.

When smartSegmentLoading is disabled, Druid uses the configured values of these properties. Disable smartSegmentLoading only if you want to explicitly set the values of any of the above properties.

Audit history

To view the audit history of Coordinator dynamic config issue a GET request to the URL -

http://<COORDINATOR_IP>:<PORT>/druid/coordinator/v1/config/history?interval=<interval>

default value of interval can be specified by setting druid.audit.manager.auditHistoryMillis (1 week if not configured) in Coordinator runtime.properties

To view last n entries of the audit history of Coordinator dynamic config issue a GET request to the URL -

http://<COORDINATOR_IP>:<PORT>/druid/coordinator/v1/config/history?count=<n>

Lookups Dynamic Configuration

These configuration options control Coordinator lookup management. See dynamic configuration for lookups configurations that affect lookup propagation.

Automatic compaction dynamic configuration

You can set or update automatic compaction properties dynamically using the Automatic compaction API without restarting Coordinators.

For details about segment compaction, see Segment size optimization.

You can configure automatic compaction through the following properties:

Automatic compaction config example:

{
  "dataSource": "wikiticker",
  "granularitySpec" : {
    "segmentGranularity" : "none"
  }
}

Compaction tasks fail when higher priority tasks cause Druid to revoke their locks. By default, realtime tasks like ingestion have a higher priority than compaction tasks. Frequent conflicts between compaction tasks and realtime tasks can cause the Coordinator’s automatic compaction to hang. You may see this issue with streaming ingestion from Kafka and Kinesis, which ingest late-arriving data.

To mitigate this problem, set skipOffsetFromLatest to a value large enough so that arriving data tends to fall outside the offset value from the current time. This way you can avoid conflicts between compaction tasks and realtime ingestion tasks. For example, if you want to skip over segments from thirty days prior to the end time of the most recent segment, assign "skipOffsetFromLatest": "P30D". For more information, see Avoid conflicts with ingestion.

Automatic compaction tuningConfig

Auto-compaction supports a subset of the tuningConfig for Parallel task. The below is a list of the supported configurations for auto-compaction.

Automatic compaction granularitySpec

Automatic compaction dimensionsSpec

Automatic compaction transformSpec

Automatic compaction ioConfig

Auto-compaction supports a subset of the ioConfig for Parallel task. The below is a list of the supported configurations for auto-compaction.

Overlord

For general Overlord Process information, see here.

Overlord Static Configuration

These Overlord static configurations can be defined in the overlord/runtime.properties file.

Overlord Process Configs

Overlord Operations

The following configs only apply if the Overlord is running in remote mode. For a description of local vs. remote mode, see Overlord Process.

There are additional configs for autoscaling (if it is enabled):

Supervisors

The druid.supervisor.idleConfig.* specified in the runtime properties of the overlord defines the default behavior for the entire cluster. See Idle Configuration in Kafka Supervisor IOConfig to override it for an individual supervisor.

Overlord dynamic configuration

The Overlord can be dynamically configured to specify how tasks are assigned to workers.

The JSON object can be submitted to the Overlord via a POST request at:

http://<OVERLORD_IP>:<port>/druid/indexer/v1/worker

Optional header parameters for auditing the config change can also be specified.

An example Overlord dynamic config is shown below:

{
  "selectStrategy": {
    "type": "fillCapacity",
    "affinityConfig": {
      "affinity": {
        "datasource1": ["host1:port", "host2:port"],
        "datasource2": ["host3:port"]
      }
    }
  },
  "autoScaler": {
    "type": "ec2",
    "minNumWorkers": 2,
    "maxNumWorkers": 12,
    "envConfig": {
      "availabilityZone": "us-east-1a",
      "nodeData": {
        "amiId": "${AMI}",
        "instanceType": "c3.8xlarge",
        "minInstances": 1,
        "maxInstances": 1,
        "securityGroupIds": ["${IDs}"],
        "keyName": "${KEY_NAME}"
      },
      "userData": {
        "impl": "string",
        "data": "${SCRIPT_COMMAND}",
        "versionReplacementString": ":VERSION:",
        "version": null
      }
    }
  }
}

Issuing a GET request to the same URL returns the current Overlord dynamic config.

To view the audit history of worker config issue a GET request to the URL -

http://<OVERLORD_IP>:<port>/druid/indexer/v1/worker/history?interval=<interval>

The default value of interval can be specified by setting druid.audit.manager.auditHistoryMillis (1 week if not configured) in Overlord runtime.properties.

To view the last n entries of the audit history of worker config, issue a GET request to the following URL:

http://<OVERLORD_IP>:<port>/druid/indexer/v1/worker/history?count=<n>

Worker select strategy

The select strategy controls how Druid assigns tasks to workers (MiddleManagers). At a high level, the select strategy determines the list of eligible workers for a given task using either an affinityConfig or a categorySpec. Then, Druid assigns the task by either trying to distribute load equally (equalDistribution) or to fill as many workers as possible to capacity (fillCapacity). There are 4 options for select strategies:

• equalDistribution
• equalDistributionWithCategorySpec
• fillCapacity
• fillCapacityWithCategorySpec

A javascript option is also available but should only be used for prototyping new strategies.

If an affinityConfig is provided (as part of fillCapacity and equalDistribution strategies) for a given task, the list of workers eligible to be assigned is determined as follows:

• a non-affinity worker if no affinity is specified for that datasource. Any worker not listed in the affinityConfig is considered a non-affinity worker.
• a non-affinity worker if preferred workers are not available and the affinity is weak i.e. strong: false.
• a preferred worker listed in the affinityConfig for this datasource if it has available capacity
• no worker if preferred workers are not available and affinity is strong i.e. strong: true. In this case, the task remains in “pending” state. The chosen provisioning strategy (e.g. pendingTaskBased) may then use the total number of pending tasks to determine if a new node should be provisioned.

Note that every worker listed in the affinityConfig will only be used for the assigned datasources and no other.

If a categorySpec is provided (as part of fillCapacityWithCategorySpec and equalDistributionWithCategorySpec strategies), then a task of a given datasource may be assigned to:

• any worker if no category config is given for task type
• any worker if category config is given for task type but no category is given for datasource and there’s no default category
• a preferred worker (based on category config and category for datasource) if available
• any worker if category config and category are given but no preferred worker is available and category config is weak
• not assigned at all if preferred workers are not available and category config is strong

In both the cases, Druid determines the list of eligible workers and selects one depending on their load with the goal of either distributing the load equally or filling as few workers as possible.

If you are using auto-scaling, use the fillCapacity select strategy since auto-scaled nodes can not be assigned a category, and you want the work to be concentrated on the fewest number of workers to allow the empty ones to scale down.

equalDistribution

Tasks are assigned to the MiddleManager with the most free slots at the time the task begins running. This evenly distributes work across your MiddleManagers.

equalDistributionWithCategorySpec

This strategy is a variant of equalDistribution, which supports workerCategorySpec field rather than affinityConfig. By specifying workerCategorySpec, you can assign tasks to run on different categories of MiddleManagers based on the type and dataSource of the task. This strategy doesn’t work with AutoScaler since the behavior is undefined.

Example: tasks of type “index_kafka” default to running on MiddleManagers of category c1, except for tasks that write to datasource “ds1,” which run on MiddleManagers of category c2.

{
  "selectStrategy": {
    "type": "equalDistributionWithCategorySpec",
    "workerCategorySpec": {
      "strong": false,
      "categoryMap": {
        "index_kafka": {
          "defaultCategory": "c1", 
          "categoryAffinity": {
            "ds1": "c2"
          }
        }
      }
    }
  }
}

fillCapacity

Tasks are assigned to the worker with the most currently-running tasks. This is useful when you are auto-scaling MiddleManagers since it tends to pack some full and leave others empty. The empty ones can be safely terminated.

Note that if druid.indexer.runner.pendingTasksRunnerNumThreads is set to N > 1, then this strategy will fill N MiddleManagers up to capacity simultaneously, rather than a single MiddleManager.

fillCapacityWithCategorySpec

This strategy is a variant of fillCapacity, which supports workerCategorySpec instead of an affinityConfig. The usage is the same as equalDistributionWithCategorySpec strategy. This strategy doesn’t work with AutoScaler since the behavior is undefined.

javascript

Allows defining arbitrary logic for selecting workers to run task using a JavaScript function. The function is passed remoteTaskRunnerConfig, map of workerId to available workers and task to be executed and returns the workerId on which the task should be run or null if the task cannot be run. It can be used for rapid development of missing features where the worker selection logic is to be changed or tuned often. If the selection logic is quite complex and cannot be easily tested in JavaScript environment, its better to write a druid extension module with extending current worker selection strategies written in java.

Example: a function that sends batch_index_task to workers 10.0.0.1 and 10.0.0.2 and all other tasks to other available workers.

{
  "type":"javascript",
  "function":"function (config, zkWorkers, task) {\nvar batch_workers = new java.util.ArrayList();\nbatch_workers.add(\"middleManager1_hostname:8091\");\nbatch_workers.add(\"middleManager2_hostname:8091\");\nworkers = zkWorkers.keySet().toArray();\nvar sortedWorkers = new Array()\n;for(var i = 0; i < workers.length; i++){\n sortedWorkers[i] = workers[i];\n}\nArray.prototype.sort.call(sortedWorkers,function(a, b){return zkWorkers.get(b).getCurrCapacityUsed() - zkWorkers.get(a).getCurrCapacityUsed();});\nvar minWorkerVer = config.getMinWorkerVersion();\nfor (var i = 0; i < sortedWorkers.length; i++) {\n var worker = sortedWorkers[i];\n  var zkWorker = zkWorkers.get(worker);\n  if(zkWorker.canRunTask(task) && zkWorker.isValidVersion(minWorkerVer)){\n    if(task.getType() == 'index_hadoop' && batch_workers.contains(worker)){\n      return worker;\n    } else {\n      if(task.getType() != 'index_hadoop' && !batch_workers.contains(worker)){\n        return worker;\n      }\n    }\n  }\n}\nreturn null;\n}"
}

info

JavaScript-based functionality is disabled by default. Please refer to the Druid JavaScript programming guide for guidelines about using Druid’s JavaScript functionality, including instructions on how to enable it.

affinityConfig

Use the affinityConfig field to pass affinity configuration to the equalDistribution and fillCapacity strategies. If not provided, the default is to have no affinity.

workerCategorySpec

WorkerCategorySpec can be provided to the equalDistributionWithCategorySpec and fillCapacityWithCategorySpec strategies using the workerCategorySpec field. If not provided, the default is to not use it at all.

CategoryConfig

Autoscaler

Amazon’s EC2 together with Google’s GCE are currently the only supported autoscalers.

EC2’s autoscaler properties are:

For GCE’s properties, please refer to the gce-extensions.

Data Server

This section contains the configuration options for the processes that reside on Data servers (MiddleManagers/Peons and Historicals) in the suggested three-server configuration.

Configuration options for the Indexer process are also provided here.

MiddleManager and Peons

These MiddleManager and Peon configurations can be defined in the middleManager/runtime.properties file.

MiddleManager Process Config

MiddleManager Configuration

Middle managers pass their configurations down to their child peons. The MiddleManager requires the following configs:

Peon Processing

Processing properties set on the MiddleManager will be passed through to Peons.

The amount of direct memory needed by Druid is at least druid.processing.buffer.sizeBytes * (druid.processing.numMergeBuffers + druid.processing.numThreads + 1). You can ensure at least this amount of direct memory is available by providing -XX:MaxDirectMemorySize=<VALUE> in druid.indexer.runner.javaOptsArray as documented above.

Peon query configuration

See general query configuration.

Peon Caching

You can optionally configure caching to be enabled on the peons by setting caching configs here.

See cache configuration for how to configure cache settings.

Additional Peon Configuration

Although peons inherit the configurations of their parent MiddleManagers, explicit child peon configs in MiddleManager can be set by prefixing them with:

druid.indexer.fork.property

Additional peon configs include:

If the peon is running in remote mode, there must be an Overlord up and running. Peons in remote mode can set the following configurations:

SegmentWriteOutMediumFactory

When new segments are created, Druid temporarily stores some preprocessed data in some buffers. Currently three types of medium exist for those buffers: temporary files, off-heap memory, and on-heap memory.

Temporary files (tmpFile) are stored under the task working directory (see druid.worker.baseTaskDirs configuration above) and thus share it’s mounting properties, e. g. they could be backed by HDD, SSD or memory (tmpfs). This type of medium may do unnecessary disk I/O and requires some disk space to be available.

Off-heap memory medium (offHeapMemory) creates buffers in off-heap memory of a JVM process that is running a task. This type of medium is preferred, but it may require to allow the JVM to have more off-heap memory, by changing -XX:MaxDirectMemorySize configuration. It is not yet understood how does the required off-heap memory size relates to the size of the segments being created. But definitely it doesn’t make sense to add more extra off-heap memory, than the configured maximum heap size (-Xmx) for the same JVM.

On-heap memory medium (onHeapMemory) creates buffers using the allocated heap memory of the JVM process running a task. Using on-heap memory introduces garbage collection overhead and so is not recommended in most cases. This type of medium is most helpful for tasks run on external clusters where it may be difficult to allocate and work with direct memory effectively.

For most types of tasks SegmentWriteOutMediumFactory could be configured per-task (see Tasks page, “TuningConfig” section), but if it’s not specified for a task, or it’s not supported for a particular task type, then the value from the configuration below is used:

Indexer

Indexer Process Configuration

Indexer General Configuration

Indexer Concurrent Requests

Druid uses Jetty to serve HTTP requests.

Indexer Processing Resources

The amount of direct memory needed by Druid is at least druid.processing.buffer.sizeBytes * (druid.processing.numMergeBuffers + druid.processing.numThreads + 1). You can ensure at least this amount of direct memory is available by providing -XX:MaxDirectMemorySize=<VALUE> at the command line.

Query Configurations

See general query configuration.

Indexer Caching

You can optionally configure caching to be enabled on the Indexer by setting caching configs here.

See cache configuration for how to configure cache settings.

Note that only local caches such as the local-type cache and caffeine cache are supported. If a remote cache such as memcached is used, it will be ignored.

Historical

For general Historical Process information, see here.

These Historical configurations can be defined in the historical/runtime.properties file.

Historical Process Configuration

Historical General Configuration

Storing Segments

In druid.segmentCache.locations, freeSpacePercent was added because maxSize setting is only a theoretical limit and assumes that much space will always be available for storing segments. In case of any druid bug leading to unaccounted segment files left alone on disk or some other process writing stuff to disk, This check can start failing segment loading early before filling up the disk completely and leaving the host usable otherwise.

In druid.segmentCache.locationSelector.strategy, one of leastBytesUsed, roundRobin, random, or mostAvailableSize could be specified to represent the strategy to distribute segments across multiple segment cache locations.

Note that if druid.segmentCache.numLoadingThreads > 1, multiple threads can download different segments at the same time. In this case, with the leastBytesUsed strategy or mostAvailableSize strategy, historicals may select a sub-optimal storage location because each decision is based on a snapshot of the storage location status of when a segment is requested to download.

Historical query configs

Concurrent Requests

Druid uses Jetty to serve HTTP requests.

Processing

The amount of direct memory needed by Druid is at least druid.processing.buffer.sizeBytes * (druid.processing.numMergeBuffers + druid.processing.numThreads + 1). You can ensure at least this amount of direct memory is available by providing -XX:MaxDirectMemorySize=<VALUE> at the command line.

Historical query configuration

See general query configuration.

Historical Caching

You can optionally only configure caching to be enabled on the Historical by setting caching configs here.

See cache configuration for how to configure cache settings.

Query Server

This section contains the configuration options for the processes that reside on Query servers (Brokers) in the suggested three-server configuration.

Configuration options for the experimental Router process are also provided here.

Broker

For general Broker process information, see here.

These Broker configurations can be defined in the broker/runtime.properties file.

Broker Process Configs

Query configuration

Query routing

Query prioritization and laning

Laning strategies allow you to control capacity utilization for heterogeneous query workloads. With laning, the broker examines and classifies a query for the purpose of assigning it to a ‘lane’. Lanes have capacity limits, enforced by the broker, that can be used to ensure sufficient resources are available for other lanes or for interactive queries (with no lane), or to limit overall throughput for queries within the lane. Requests in excess of the capacity are discarded with an HTTP 429 status code.

Prioritization strategies

Manual prioritization strategy

With this configuration, queries are never assigned a priority automatically, but will preserve a priority manually set on the query context with the priority key. This mode can be explicitly set by setting druid.query.scheduler.prioritization.strategy to manual.

Threshold prioritization strategy

This prioritization strategy lowers the priority of queries that cross any of a configurable set of thresholds, such as how far in the past the data is, how large of an interval a query covers, or the number of segments taking part in a query.

This strategy can be enabled by setting druid.query.scheduler.prioritization.strategy to threshold.

Laning strategies

No laning strategy

In this mode, queries are never assigned a lane, and the concurrent query count will only be limited by druid.server.http.numThreads or druid.query.scheduler.numThreads, if set. This is the default Druid query scheduler operating mode. Enable this strategy explicitly by setting druid.query.scheduler.laning.strategy to none.

‘High/Low’ laning strategy

This laning strategy splits queries with a priority below zero into a low query lane, automatically. Queries with priority of zero (the default) or above are considered ‘interactive’. The limit on low queries can be set to some desired percentage of the total capacity (or HTTP thread pool size), reserving capacity for interactive queries. Queries in the low lane are not guaranteed their capacity, which may be consumed by interactive queries, but may use up to this limit if total capacity is available.

If the low lane is specified in the query context lane parameter, this will override the computed lane.

This strategy can be enabled by setting druid.query.scheduler.laning.strategy=hilo.

Guardrails for materialization of subqueries

Druid stores the subquery rows in temporary tables that live in the Java heap. It is a good practice to avoid large subqueries in Druid. Therefore there are guardrails that are built in Druid to prevent the queries from generating subquery results which can exhaust the heap space. They can be set on a cluster level or modified per query level as desired. Note the following guardrails that can be set by the cluster admin to limit the subquery results:

1. druid.server.http.maxSubqueryRows in broker’s config to set a default for the entire cluster or maxSubqueryRows in the query context to set an upper limit on the number of rows a subquery can generate
2. druid.server.http.maxSubqueryBytes in broker’s config to set a default for the entire cluster or maxSubqueryBytes in the query context to set an upper limit on the number of bytes a subquery can generate

Note that limiting the subquery by bytes is a newer feature therefore it is experimental as it materializes the results differently.

If you choose to modify or set any of the above limits, you must also think about the heap size of all Brokers, Historicals, and task Peons that process data for the subqueries to accommodate the subquery results. There is no formula to calculate the correct value. Trial and error is the best approach.

‘Manual’ laning strategy

This laning strategy is best suited for cases where one or more external applications which query Druid are capable of manually deciding what lane a given query should belong to. Configured with a map of lane names to percent or exact max capacities, queries with a matching lane parameter in the query context will be subjected to those limits.

Server Configuration

Druid uses Jetty to serve HTTP requests. Each query being processed consumes a single thread from druid.server.http.numThreads, so consider defining druid.query.scheduler.numThreads to a lower value in order to reserve HTTP threads for responding to health checks, lookup loading, and other non-query, (in most cases) comparatively very short-lived, HTTP requests.

Client Configuration

Druid Brokers use an HTTP client to communicate with data servers (Historical servers and real-time tasks). This client has the following configuration options.

Retry Policy

Druid broker can optionally retry queries internally for transient errors.

Processing

The broker uses processing configs for nested groupBy queries.

The amount of direct memory needed by Druid is at least druid.processing.buffer.sizeBytes * (druid.processing.numMergeBuffers + 1). You can ensure at least this amount of direct memory is available by providing -XX:MaxDirectMemorySize=<VALUE> at the command line.

Broker query configuration

See general query configuration.

Broker Generated Query Configuration Supplementation

The Broker generates queries internally. This configuration section describes how an operator can augment the configuration of these queries.

As of now the only supported augmentation is overriding the default query context. This allows an operator the flexibility to adjust it as they see fit. A common use of this configuration is to override the query priority of the cluster generated queries in order to avoid running as a default priority of 0.

SQL

The Druid SQL server is configured through the following properties on the Broker.

info

Previous versions of Druid had properties named druid.sql.planner.maxQueryCount and druid.sql.planner.maxSemiJoinRowsInMemory. These properties are no longer available. Since Druid 0.18.0, you can use druid.server.http.maxSubqueryRows to control the maximum number of rows permitted across all subqueries.

Broker Caching

You can optionally only configure caching to be enabled on the Broker by setting caching configs here.

See cache configuration for how to configure cache settings.

info

Note: Even if cache is enabled, for groupBy v2 queries, both of non-result level cache and result level cache do not work on Brokers. See Differences between v1 and v2 and Query caching for more information.

Segment Discovery

Cache Configuration

This section describes caching configuration that is common to Broker, Historical, and MiddleManager/Peon processes.

Caching could optionally be enabled on the Broker, Historical, and MiddleManager/Peon processes. See Broker, Historical, and Peon configuration options for how to enable it for different processes.

Druid uses a local in-memory cache by default, unless a different type of cache is specified. Use the druid.cache.type configuration to set a different kind of cache.

Cache settings are set globally, so the same configuration can be re-used for both Broker and Historical processes, when defined in the common properties file.

Cache Type

Local Cache

info

DEPRECATED: Use caffeine (default as of v0.12.0) instead

The local cache is deprecated in favor of the Caffeine cache, and may be removed in a future version of Druid. The Caffeine cache affords significantly better performance and control over eviction behavior compared to local cache, and is recommended in any situation where you are using JRE 8u60 or higher.

A simple in-memory LRU cache. Local cache resides in JVM heap memory, so if you enable it, make sure you increase heap size accordingly.

Caffeine Cache

A highly performant local cache implementation for Druid based on Caffeine. Requires a JRE8u60 or higher if using COMMON_FJP.

Configuration

Below are the configuration options known to this module:

druid.cache.cacheExecutorFactory

Here are the possible values for druid.cache.cacheExecutorFactory, which controls how maintenance tasks are run

• COMMON_FJP (default) use the common ForkJoinPool. Should use with JRE 8u60 or higher. Older versions of the JRE may have worse performance than newer JRE versions.
• SINGLE_THREAD Use a single-threaded executor.
• SAME_THREAD Cache maintenance is done eagerly.

Metrics

In addition to the normal cache metrics, the caffeine cache implementation also reports the following in both total and delta

Memcached

Uses memcached as cache backend. This allows all processes to share the same cache.

Hybrid

Uses a combination of any two caches as a two-level L1 / L2 cache. This may be used to combine a local in-memory cache with a remote memcached cache.

Cache requests will first check L1 cache before checking L2. If there is an L1 miss and L2 hit, it will also populate L1.

General query configuration

This section describes configurations that control behavior of Druid’s query types, applicable to Broker, Historical, and MiddleManager processes.

Overriding default query context values

Any query context general parameter default value can be overridden by setting the runtime property in the format of druid.query.default.context.{query_context_key}. The druid.query.default.context.{query_context_key} runtime property prefix applies to all current and future query context keys, the same as how query context parameter passed with the query works. Note that the runtime property value can be overridden if value for the same key is explicitly specify in the query contexts.

The precedence chain for query context values is as follows:

hard-coded default value in Druid code <- runtime property not prefixed with druid.query.default.context <- runtime property prefixed with druid.query.default.context <- context parameter in the query

Note that not all query context key has a runtime property not prefixed with druid.query.default.context that can override the hard-coded default value. For example, maxQueuedBytes has druid.broker.http.maxQueuedBytes but joinFilterRewriteMaxSize does not. Hence, the only way of overriding joinFilterRewriteMaxSize hard-coded default value is with runtime property druid.query.default.context.joinFilterRewriteMaxSize.

To further elaborate on the previous example:

If neither druid.broker.http.maxQueuedBytes or druid.query.default.context.maxQueuedBytes is set and the query does not have maxQueuedBytes in the context, then the hard-coded value in Druid code is use. If runtime property only contains druid.broker.http.maxQueuedBytes=x and query does not have maxQueuedBytes in the context, then the value of the property, x, is use. However, if query does have maxQueuedBytes in the context, then that value is use instead. If runtime property only contains druid.query.default.context.maxQueuedBytes=y OR runtime property contains both druid.broker.http.maxQueuedBytes=x and druid.query.default.context.maxQueuedBytes=y, then the value of druid.query.default.context.maxQueuedBytes, y, is use (given that query does not have maxQueuedBytes in the context). If query does have maxQueuedBytes in the context, then that value is use instead.

TopN query config

Search query config

SegmentMetadata query config

GroupBy query config

This section describes the configurations for groupBy queries. You can set the runtime properties in the runtime.properties file on Broker, Historical, and MiddleManager processes. You can set the query context parameters through the query context.

Configurations for groupBy v2

Supported runtime properties:

Supported query contexts:

Advanced configurations

Common configurations for all groupBy strategies

Supported runtime properties:

Supported query contexts:

GroupBy v2 configurations

Supported runtime properties:

Supported query contexts:

GroupBy v1 configurations

Supported runtime properties:

Supported query contexts:

Expression processing configurations

Router

Router Process Configs

Runtime Configuration