Input sources

The input source defines where your index task reads data for Apache Druid native batch ingestion. Only the native parallel task and simple task support the input source.

For general information on native batch indexing and parallel task indexing, see Native batch ingestion.

S3 input source

info

You need to include the druid-s3-extensions as an extension to use the S3 input source.

The S3 input source reads objects directly from S3. You can specify either:

• a list of S3 URI strings
• a list of S3 location prefixes that attempts to list the contents and ingest all objects contained within the locations.

The S3 input source is splittable. Therefore, you can use it with the Parallel task. Each worker task of index_parallel reads one or multiple objects.

Sample specs:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "objectGlob": "**.json",
        "uris": ["s3://foo/bar/file.json", "s3://bar/foo/file2.json"]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "objectGlob": "**.parquet",
        "prefixes": ["s3://foo/bar/", "s3://bar/foo/"]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "objectGlob": "**.json",
        "objects": [
          { "bucket": "foo", "path": "bar/file1.json"},
          { "bucket": "bar", "path": "foo/file2.json"}
        ]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "objectGlob": "**.json",
        "uris": ["s3://foo/bar/file.json", "s3://bar/foo/file2.json"],
        "properties": {
          "accessKeyId": "KLJ78979SDFdS2",
          "secretAccessKey": "KLS89s98sKJHKJKJH8721lljkd"
        }
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "objectGlob": "**.json",
        "uris": ["s3://foo/bar/file.json", "s3://bar/foo/file2.json"],
        "properties": {
          "accessKeyId": "KLJ78979SDFdS2",
          "secretAccessKey": "KLS89s98sKJHKJKJH8721lljkd",
          "assumeRoleArn": "arn:aws:iam::2981002874992:role/role-s3"
        }
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "uris": ["s3://foo/bar/file.json", "s3://bar/foo/file2.json"],
        "endpointConfig": {
             "url" : "s3-store.aws.com",
             "signingRegion" : "us-west-2"
         },
         "clientConfig": {
             "protocol" : "http",
             "disableChunkedEncoding" : true,
             "enablePathStyleAccess" : true,
             "forceGlobalBucketAccessEnabled" : false
         },
         "proxyConfig": {
             "host" : "proxy-s3.aws.com",
             "port" : 8888,
             "username" : "admin",
             "password" : "admin"
         },
        "properties": {
          "accessKeyId": "KLJ78979SDFdS2",
          "secretAccessKey": "KLS89s98sKJHKJKJH8721lljkd",
          "assumeRoleArn": "arn:aws:iam::2981002874992:role/role-s3"
        }
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

Note that the S3 input source will skip all empty objects only when prefixes is specified.

S3 Object:

Properties Object:

info

Note: If accessKeyId and secretAccessKey are not given, the default S3 credentials provider chain is used.

Google Cloud Storage input source

info

You need to include the druid-google-extensions as an extension to use the Google Cloud Storage input source.

The Google Cloud Storage input source is to support reading objects directly from Google Cloud Storage. Objects can be specified as list of Google Cloud Storage URI strings. The Google Cloud Storage input source is splittable and can be used by the Parallel task, where each worker task of index_parallel will read one or multiple objects.

Sample specs:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "google",
        "objectGlob": "**.json",
        "uris": ["gs://foo/bar/file.json", "gs://bar/foo/file2.json"]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "google",
        "objectGlob": "**.parquet",
        "prefixes": ["gs://foo/bar/", "gs://bar/foo/"]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "google",
        "objectGlob": "**.json",
        "objects": [
          { "bucket": "foo", "path": "bar/file1.json"},
          { "bucket": "bar", "path": "foo/file2.json"}
        ]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

Note that the Google Cloud Storage input source will skip all empty objects only when prefixes is specified.

Google Cloud Storage object:

Azure input source

info

You need to include the druid-azure-extensions as an extension to use the Azure input source.

The Azure input source reads objects directly from Azure Blob store or Azure Data Lake sources. You can specify objects as a list of file URI strings or prefixes. You can split the Azure input source for use with Parallel task indexing and each worker task reads one chunk of the split data.

Sample specs:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "azure",
        "objectGlob": "**.json",
        "uris": ["azure://container/prefix1/file.json", "azure://container/prefix2/file2.json"]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "azure",
        "objectGlob": "**.parquet",
        "prefixes": ["azure://container/prefix1/", "azure://container/prefix2/"]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "azure",
        "objectGlob": "**.json",
        "objects": [
          { "bucket": "container", "path": "prefix1/file1.json"},
          { "bucket": "container", "path": "prefix2/file2.json"}
        ]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

Note that the Azure input source skips all empty objects only when prefixes is specified.

The objects property is:

HDFS input source

info

You need to include the druid-hdfs-storage as an extension to use the HDFS input source.

The HDFS input source is to support reading files directly from HDFS storage. File paths can be specified as an HDFS URI string or a list of HDFS URI strings. The HDFS input source is splittable and can be used by the Parallel task, where each worker task of index_parallel will read one or multiple files.

Sample specs:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "hdfs",
        "paths": "hdfs://namenode_host/foo/bar/", "hdfs://namenode_host/bar/foo"
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "hdfs",
        "paths": "hdfs://namenode_host/foo/bar/", "hdfs://namenode_host/bar/foo"
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "hdfs",
        "paths": "hdfs://namenode_host/foo/bar/file.json", "hdfs://namenode_host/bar/foo/file2.json"
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "hdfs",
        "paths": ["hdfs://namenode_host/foo/bar/file.json", "hdfs://namenode_host/bar/foo/file2.json"]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

You can also ingest from other storage using the HDFS input source if the HDFS client supports that storage. However, if you want to ingest from cloud storage, consider using the service-specific input source for your data storage. If you want to use a non-hdfs protocol with the HDFS input source, include the protocol in druid.ingestion.hdfs.allowedProtocols. See HDFS input source security configuration for more details.

HTTP input source

The HTTP input source is to support reading files directly from remote sites via HTTP.

info

Security notes: Ingestion tasks run under the operating system account that runs the Druid processes, for example the Indexer, Middle Manager, and Peon. This means any user who can submit an ingestion task can specify an input source referring to any location that the Druid process can access. For example, using http input source, users may have access to internal network servers.

The http input source is not limited to the HTTP or HTTPS protocols. It uses the Java URI class that supports HTTP, HTTPS, FTP, file, and jar protocols by default.

For more information about security best practices, see Security overview.

The HTTP input source is splittable and can be used by the Parallel task, where each worker task of index_parallel will read only one file. This input source does not support Split Hint Spec.

Sample specs:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "http",
        "uris": ["http://example.com/uri1", "http://example2.com/uri2"]
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

Example with authentication fields using the DefaultPassword provider (this requires the password to be in the ingestion spec):

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "http",
        "uris": ["http://example.com/uri1", "http://example2.com/uri2"],
        "httpAuthenticationUsername": "username",
        "httpAuthenticationPassword": "password123"
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...

You can also use the other existing Druid PasswordProviders. Here is an example using the EnvironmentVariablePasswordProvider:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "http",
        "uris": ["http://example.com/uri1", "http://example2.com/uri2"],
        "httpAuthenticationUsername": "username",
        "httpAuthenticationPassword": {
          "type": "environment",
          "variable": "HTTP_INPUT_SOURCE_PW"
        }
      },
      "inputFormat": {
        "type": "json"
      },
      ...
    },
...
}

You can only use protocols listed in the druid.ingestion.http.allowedProtocols property as HTTP input sources. The http and https protocols are allowed by default. See HTTP input source security configuration for more details.

Inline input source

The Inline input source can be used to read the data inlined in its own spec. It can be used for demos or for quickly testing out parsing and schema.

Sample spec:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "inline",
        "data": "0,values,formatted\n1,as,CSV"
      },
      "inputFormat": {
        "type": "csv"
      },
      ...
    },
...

Local input source

The Local input source is to support reading files directly from local storage, and is mainly intended for proof-of-concept testing. The Local input source is splittable and can be used by the Parallel task, where each worker task of index_parallel will read one or multiple files.

Sample spec:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "local",
        "filter" : "*.csv",
        "baseDir": "/data/directory",
        "files": ["/bar/foo", "/foo/bar"]
      },
      "inputFormat": {
        "type": "csv"
      },
      ...
    },
...

Druid input source

The Druid input source is to support reading data directly from existing Druid segments, potentially using a new schema and changing the name, dimensions, metrics, rollup, etc. of the segment. The Druid input source is splittable and can be used by the Parallel task. This input source has a fixed input format for reading from Druid segments; no inputFormat field needs to be specified in the ingestion spec when using this input source.

The Druid input source can be used for a variety of purposes, including:

• Creating new datasources that are rolled-up copies of existing datasources.
• Changing the partitioning or sorting of a datasource to improve performance.
• Updating or removing rows using a transformSpec.

When using the Druid input source, the timestamp column shows up as a numeric field named __time set to the number of milliseconds since the epoch (January 1, 1970 00:00:00 UTC). It is common to use this in the timestampSpec, if you want the output timestamp to be equivalent to the input timestamp. In this case, set the timestamp column to __time and the format to auto or millis.

It is OK for the input and output datasources to be the same. In this case, newly generated data will overwrite the previous data for the intervals specified in the granularitySpec. Generally, if you are going to do this, it is a good idea to test out your reindexing by writing to a separate datasource before overwriting your main one. Alternatively, if your goals can be satisfied by compaction, consider that instead as a simpler approach.

An example task spec is shown below. It reads from a hypothetical raw datasource wikipedia_raw and creates a new rolled-up datasource wikipedia_rollup by grouping on hour, “countryName”, and “page”.

{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "wikipedia_rollup",
      "timestampSpec": {
        "column": "__time",
        "format": "millis"
      },
      "dimensionsSpec": {
        "dimensions": [
          "countryName",
          "page"
        ]
      },
      "metricsSpec": [
        {
          "type": "count",
          "name": "cnt"
        }
      ],
      "granularitySpec": {
        "type": "uniform",
        "queryGranularity": "HOUR",
        "segmentGranularity": "DAY",
        "intervals": ["2016-06-27/P1D"],
        "rollup": true
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "druid",
        "dataSource": "wikipedia_raw",
        "interval": "2016-06-27/P1D"
      }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "partitionsSpec": {
        "type": "hashed"
      },
      "forceGuaranteedRollup": true,
      "maxNumConcurrentSubTasks": 1
    }
  }
}

info

Note: Older versions (0.19 and earlier) did not respect the timestampSpec when using the Druid input source. If you have ingestion specs that rely on this and cannot rewrite them, set druid.indexer.task.ignoreTimestampSpecForDruidInputSource to true to enable a compatibility mode where the timestampSpec is ignored.

The secondary partitioning method determines the requisite number of concurrent worker tasks that run in parallel to complete ingestion with the Combining input source. Set this value in maxNumConcurrentSubTasks in tuningConfig based on the secondary partitioning method:

• range or single_dim partitioning: greater than or equal to 1
• hashed or dynamic partitioning: greater than or equal to 2

For more information on the maxNumConcurrentSubTasks field, see Implementation considerations.

SQL input source

The SQL input source is used to read data directly from RDBMS. The SQL input source is splittable and can be used by the Parallel task, where each worker task will read from one SQL query from the list of queries. This input source does not support Split Hint Spec. Since this input source has a fixed input format for reading events, no inputFormat field needs to be specified in the ingestion spec when using this input source. Please refer to the Recommended practices section below before using this input source.

The following is an example of an SQL input source spec:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "sql",
        "database": {
            "type": "mysql",
            "connectorConfig": {
                "connectURI": "jdbc:mysql://host:port/schema",
                "user": "user",
                "password": "password"
            }
        },
        "sqls": ["SELECT * FROM table1 WHERE timestamp BETWEEN '2013-01-01 00:00:00' AND '2013-01-01 11:59:59'", "SELECT * FROM table2 WHERE timestamp BETWEEN '2013-01-01 00:00:00' AND '2013-01-01 11:59:59'"]
      }
    },
...

The spec above will read all events from two separate SQLs for the interval 2013-01-01/2013-01-02. Each of the SQL queries will be run in its own sub-task and thus for the above example, there would be two sub-tasks.

Recommended practices

Compared to the other native batch input sources, SQL input source behaves differently in terms of reading the input data. Therefore, consider the following points before using this input source in a production environment:

• During indexing, each sub-task would execute one of the SQL queries and the results are stored locally on disk. The sub-tasks then proceed to read the data from these local input files and generate segments. Presently, there isn’t any restriction on the size of the generated files and this would require the MiddleManagers or Indexers to have sufficient disk capacity based on the volume of data being indexed.

• Filtering the SQL queries based on the intervals specified in the granularitySpec can avoid unwanted data being retrieved and stored locally by the indexing sub-tasks. For example, if the intervals specified in the granularitySpec is ["2013-01-01/2013-01-02"] and the SQL query is SELECT * FROM table1, SqlInputSource will read all the data for table1 based on the query, even though only data between the intervals specified will be indexed into Druid.

• Pagination may be used on the SQL queries to ensure that each query pulls a similar amount of data, thereby improving the efficiency of the sub-tasks.

• Similar to file-based input formats, any updates to existing data will replace the data in segments specific to the intervals specified in the granularitySpec.

Combining input source

The Combining input source lets you read data from multiple input sources. It identifies the splits from delegate input sources and uses a worker task to process each split. Use the Combining input source only if all the delegates are splittable and can be used by the Parallel task.

Similar to other input sources, the Combining input source supports a single inputFormat. Delegate input sources that require an inputFormat must have the same format for input data. If you include the Druid input source, the timestamp column is stored in the __time field. To correctly combine the data from the Druid input source with another source, ensure that other delegate input sources also store the timestamp column in __time.

The following is an example of a Combining input source spec:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "combining",
        "delegates" : [
         {
          "type": "local",
          "filter" : "*.csv",
          "baseDir": "/data/directory",
          "files": ["/bar/foo", "/foo/bar"]
         },
         {
          "type": "druid",
          "dataSource": "wikipedia",
          "interval": "2013-01-01/2013-01-02"
         }
        ]
      },
      "inputFormat": {
        "type": "csv"
      },
      ...
    },
...

Iceberg input source

info

To use the Iceberg input source, load the extension druid-iceberg-extensions.

You use the Iceberg input source to read data stored in the Iceberg table format. For a given table, the input source scans up to the latest Iceberg snapshot from the configured Hive catalog. Druid ingests the underlying live data files using the existing input source formats.

The Iceberg input source cannot be independent as it relies on the existing input sources to read from the data files. For example, if the warehouse associated with an Iceberg catalog is on S3, you must also load the druid-s3-extensions extension.

The following is a sample spec for a HDFS warehouse source:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "iceberg",
        "tableName": "iceberg_table",
        "namespace": "iceberg_namespace",
        "icebergCatalog": {
            "type": "hive",
            "warehousePath": "hdfs://warehouse/path",
            "catalogUri": "thrift://hive-metastore.x.com:8970",
            "catalogProperties": {
                "hive.metastore.connect.retries": "1",
                "hive.metastore.execute.setugi": "false",
                "hive.metastore.kerberos.principal": "KRB_PRINCIPAL",
                "hive.metastore.sasl.enabled": "true",
                "metastore.catalog.default": "catalog_test",
                "hadoop.security.authentication": "kerberos",
                "hadoop.security.authorization": "true"
            }
        },
        "icebergFilter": {
            "type": "interval",
            "filterColumn": "event_time",
            "intervals": [
              "2023-05-10T19:00:00.000Z/2023-05-10T20:00:00.000Z"
            ]
        },
        "warehouseSource": {
            "type": "hdfs"
        },
        "snapshotTime": "2023-06-01T00:00:00.000Z",
      },
      "inputFormat": {
        "type": "parquet"
      }
  },
      ...
},
...

The following is a sample spec for a S3 warehouse source:

...
        "ioConfig": {
          "type": "index_parallel",
          "inputSource": {
            "type": "iceberg",
            "tableName": "iceberg_table",
            "namespace": "iceberg_namespace",
            "icebergCatalog": {
              "type": "hive",
              "warehousePath": "s3://warehouse/path",
              "catalogUri": "thrift://hive-metastore.x.com:8970",
              "catalogProperties": {
                "hive.metastore.connect.retries": "1",
                "hive.metastore.execute.setugi": "false",
                "hive.metastore.kerberos.principal": "KRB_PRINCIPAL",
                "hive.metastore.sasl.enabled": "true",
                "metastore.catalog.default": "default_catalog",
                "fs.s3a.access.key" : "S3_ACCESS_KEY",
                "fs.s3a.secret.key" : "S3_SECRET_KEY",
                "fs.s3a.endpoint" : "S3_API_ENDPOINT"
              }
            },
            "icebergFilter": {
              "type": "interval",
              "filterColumn": "event_time",
              "intervals": [
                "2023-05-10T19:00:00.000Z/2023-05-10T20:00:00.000Z"
              ]
            },
            "warehouseSource": {
              "type": "s3",
              "endpointConfig": {
                "url": "teststore.aws.com",
                "signingRegion": "us-west-2a"
              },
              "clientConfig": {
                "protocol": "http",
                "disableChunkedEncoding": true,
                "enablePathStyleAccess": true,
                "forceGlobalBucketAccessEnabled": false
              },
              "properties": {
                "accessKeyId": {
                  "type": "default",
                  "password": "foo"
                },
                "secretAccessKey": {
                  "type": "default",
                  "password": "bar"
                }
              },
            }
          },
          "inputFormat": {
            "type": "parquet"
          }
        },
...
},

Catalog Object

The catalog object supports local and hive catalog types.

The following table lists the properties of a local catalog:

The following table lists the properties of a hive catalog:

Iceberg filter object

This input source provides the following filters: and, equals, interval, and or. You can use these filters to filter out data files from a snapshot, reducing the number of files Druid has to ingest.

equals Filter:

interval Filter:

and Filter:

or Filter:

not Filter:

range Filter:

Delta Lake input source

info

To use the Delta Lake input source, load the extension druid-deltalake-extensions.

You can use the Delta input source to read data stored in a Delta Lake table. For a given table, the input source scans the latest snapshot from the configured table. Druid ingests the underlying delta files from the table.

The following is a sample spec:

...
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "delta",
        "tablePath": "/delta-table/directory"
      },
    }
}