SQL-based ingestion reference

info

This page describes SQL-based batch ingestion using the druid-multi-stage-query extension, new in Druid 24.0. Refer to the ingestion methods table to determine which ingestion method is right for you.

SQL reference

This topic is a reference guide for the multi-stage query architecture in Apache Druid. For examples of real-world usage, refer to the Examples page.

INSERT and REPLACE load data into a Druid datasource from either an external input source, or from another datasource. When loading from an external datasource, you typically must provide the kind of input source, the data format, and the schema (signature) of the input file. Druid provides table functions to allow you to specify the external file. There are two kinds. EXTERN works with the JSON-serialized specs for the three items, using the same JSON you would use in native ingest. A set of other, input-source-specific functions use SQL syntax to specify the format and the input schema. There is one function for each input source. The input-source-specific functions allow you to use SQL query parameters to specify the set of files (or URIs), making it easy to reuse the same SQL statement for each ingest: just specify the set of files to use each time.

EXTERN Function

Use the EXTERN function to read external data or write to an external location.

EXTERN as an input source

The function has two variations. Function variation 1, with the input schema expressed as JSON:

SELECT
 <column>
FROM TABLE(
  EXTERN(
    '<Druid input source>',
    '<Druid input format>',
    '<row signature>'
  )
)

EXTERN consists of the following parts:

1. Any Druid input source as a JSON-encoded string.
2. Any Druid input format as a JSON-encoded string.
3. A row signature, as a JSON-encoded array of column descriptors. Each column descriptor must have a name and a type. The type can be string, long, double, or float. This row signature is used to map the external data into the SQL layer.

Variation 2, with the input schema expressed in SQL using an EXTEND clause. (See the next section for more detail on EXTEND). This format also uses named arguments to make the SQL a bit easier to read:

SELECT
 <column>
FROM TABLE(
  EXTERN(
    inputSource => '<Druid input source>',
    inputFormat => '<Druid input format>'
  )) (<columns>)

The input source and format are as above. The columns are expressed as in a SQL CREATE TABLE. Example: (timestamp VARCHAR, metricType VARCHAR, value BIGINT). The optional EXTEND keyword can precede the column list: EXTEND (timestamp VARCHAR...).

For more information, see Read external data with EXTERN.

EXTERN to export to a destination

EXTERN can be used to specify a destination where you want to export data to. This variation of EXTERN requires one argument, the details of the destination as specified below. This variation additionally requires an AS clause to specify the format of the exported rows.

While exporting data, some metadata files will also be created at the destination in addition to the data. These files will be created in a directory _symlink_format_manifest.

• _symlink_format_manifest/manifest: Lists the files which were created as part of the export. The file is in the symlink manifest format, and consists of a list of absolute paths to the files created.

s3://export-bucket/export/query-6564a32f-2194-423a-912e-eead470a37c4-worker2-partition2.csv
s3://export-bucket/export/query-6564a32f-2194-423a-912e-eead470a37c4-worker1-partition1.csv
s3://export-bucket/export/query-6564a32f-2194-423a-912e-eead470a37c4-worker0-partition0.csv
...
s3://export-bucket/export/query-6564a32f-2194-423a-912e-eead470a37c4-worker0-partition24.csv

Keep the following in mind when using EXTERN to export rows:

• Only INSERT statements are supported.
• Only CSV format is supported as an export format.
• Partitioning (PARTITIONED BY) and clustering (CLUSTERED BY) aren’t supported with export statements.
• You can export to Amazon S3 or local storage.
• The destination provided should contain no other files or directories.

When you export data, use the rowsPerPage context parameter to restrict the size of exported files. When the number of rows in the result set exceeds the value of the parameter, Druid splits the output into multiple files.

INSERT INTO
  EXTERN(<destination function>)
AS CSV
SELECT
  <column>
FROM <table>

S3

Export results to S3 by passing the function s3() as an argument to the EXTERN function. Note that this requires the druid-s3-extensions. The s3() function is a Druid function that configures the connection. Arguments for s3() should be passed as named parameters with the value in single quotes like the following example:

INSERT INTO
  EXTERN(
    s3(bucket => 'your_bucket', prefix => 'prefix/to/files')
  )
AS CSV
SELECT
  <column>
FROM <table>

Supported arguments for the function:

The following runtime parameters must be configured to export into an S3 destination:

GS

Export results to GCS by passing the function google() as an argument to the EXTERN function. Note that this requires the druid-google-extensions. The google() function is a Druid function that configures the connection. Arguments for google() should be passed as named parameters with the value in single quotes like the following example:

INSERT INTO
  EXTERN(
    google(bucket => 'your_bucket', prefix => 'prefix/to/files')
  )
AS CSV
SELECT
  <column>
FROM <table>

Supported arguments for the function:

The following runtime parameters must be configured to export into a GCS destination:

LOCAL

You can export to the local storage, which exports the results to the filesystem of the MSQ worker. This is useful in a single node setup or for testing but is not suitable for production use cases.

Export results to local storage by passing the function LOCAL() as an argument for the EXTERN FUNCTION. To use local storage as an export destination, the runtime property druid.export.storage.baseDir must be configured on the Indexer/Middle Manager. This value must be set to an absolute path on the local machine. Exporting data will be allowed to paths which match the prefix set by this value. Arguments to LOCAL() should be passed as named parameters with the value in single quotes in the following example:

INSERT INTO
  EXTERN(
    local(exportPath => 'exportLocation/query1')
  )
AS CSV
SELECT
  <column>
FROM <table>

Supported arguments to the function:

For more information, see Read external data with EXTERN.

INSERT

Use the INSERT statement to insert data.

Unlike standard SQL, INSERT loads data into the target table according to column name, not positionally. If necessary, use AS in your SELECT column list to assign the correct names. Do not rely on their positions within the SELECT clause.

Statement format:

INSERT INTO <table name>
< SELECT query >
PARTITIONED BY <time frame>
[ CLUSTERED BY <column list> ]

INSERT consists of the following parts:

1. Optional context parameters.
2. An INSERT INTO <dataSource> clause at the start of your query, such as INSERT INTO your-table.
3. A clause for the data you want to insert, such as SELECT ... FROM .... You can use EXTERN to reference external tables using FROM TABLE(EXTERN(...)).
4. A PARTITIONED BY clause, such as PARTITIONED BY DAY.
5. An optional CLUSTERED BY clause.

For more information, see Load data with INSERT.

REPLACE

You can use the REPLACE function to replace all or some of the data.

Unlike standard SQL, REPLACE loads data into the target table according to column name, not positionally. If necessary, use AS in your SELECT column list to assign the correct names. Do not rely on their positions within the SELECT clause.

REPLACE all data

Function format to replace all data:

REPLACE INTO <target table>
OVERWRITE ALL
< SELECT query >
PARTITIONED BY <time granularity>
[ CLUSTERED BY <column list> ]

REPLACE specific time ranges

Function format to replace specific time ranges:

REPLACE INTO <target table>
OVERWRITE WHERE __time >= TIMESTAMP '<lower bound>' AND __time < TIMESTAMP '<upper bound>'
< SELECT query >
PARTITIONED BY <time granularity>
[ CLUSTERED BY <column list> ]

REPLACE consists of the following parts:

1. Optional context parameters.
2. A REPLACE INTO <dataSource> clause at the start of your query, such as REPLACE INTO "your-table".
3. An OVERWRITE clause after the datasource, either OVERWRITE ALL or OVERWRITE WHERE:
   • OVERWRITE ALL replaces the entire existing datasource with the results of the query.
   • OVERWRITE WHERE drops the time segments that match the condition you set. Conditions are based on the __time column and use the format __time [< > = <= >=] TIMESTAMP. Use them with AND, OR, and NOT between them, inclusive of the timestamps specified. No other expressions or functions are valid in OVERWRITE.
4. A clause for the actual data you want to use for the replacement.
5. A PARTITIONED BY clause, such as PARTITIONED BY DAY.
6. An optional CLUSTERED BY clause.

For more information, see Overwrite data with REPLACE.

PARTITIONED BY

The PARTITIONED BY <time granularity> clause is required for INSERT and REPLACE. See Partitioning for details.

The following granularity arguments are accepted:

• Time unit keywords: HOUR, DAY, MONTH, or YEAR. Equivalent to FLOOR(__time TO TimeUnit).
• Time units as ISO 8601 period strings: 'PT1H', 'P1D', etc. (Druid 26.0 and later.)
• TIME_FLOOR(__time, 'granularity_string'), where granularity_string is one of the ISO 8601 periods listed below. The first argument must be __time.
• FLOOR(__time TO TimeUnit), where TimeUnit is any unit supported by the FLOOR function. The first argument must be __time.
• ALL or ALL TIME, which effectively disables time partitioning by placing all data in a single time chunk. To use LIMIT or OFFSET at the outer level of your INSERT or REPLACE query, you must set PARTITIONED BY to ALL or ALL TIME.

Earlier versions required the TIME_FLOOR notation to specify a granularity other than the keywords. In the current version, the string constant provides a simpler equivalent solution.

The following ISO 8601 periods are supported for TIME_FLOOR and the string constant:

• PT1S
• PT1M
• PT5M
• PT10M
• PT15M
• PT30M
• PT1H
• PT6H
• P1D
• P1W*
• P1M
• P3M
• P1Y

The string constant can also include any of the keywords mentioned above:

• HOUR - Same as 'PT1H'
• DAY - Same as 'P1D'
• MONTH - Same as 'P1M'
• YEAR - Same as 'P1Y'
• ALL TIME
• ALL - Alias for ALL TIME

Examples:

-- Keyword
PARTITIONED BY HOUR
-- String literal
PARTITIONED BY 'HOUR'
-- ISO 8601 period
PARTITIONED BY 'PT1H'
-- TIME_FLOOR function
PARTITIONED BY TIME_FLOOR(__time, 'PT1H')

For more information about partitioning, see Partitioning.

*Avoid partitioning by week, P1W, because weeks don’t align neatly with months and years, making it difficult to partition by coarser granularities later.

CLUSTERED BY

The CLUSTERED BY <column list> clause is optional for INSERT and REPLACE. It accepts a list of column names or expressions.

This column list is used for secondary partitioning of segments within a time chunk, and sorting of rows within a segment. For sorting purposes, Druid implicitly prepends __time to the CLUSTERED BY column list, unless forceSegmentSortByTime is set to false (an experimental feature; see Sorting for details).

For more information about clustering, see Clustering.

Context parameters

In addition to the Druid SQL context parameters, the multi-stage query task engine accepts certain context parameters that are specific to it.

Use context parameters alongside your queries to customize the behavior of the query. If you’re using the API, include the context parameters in the query context when you submit a query:

{
  "query": "SELECT 1 + 1",
  "context": {
    "<key>": "<value>",
    "maxNumTasks": 3
  }
}

If you’re using the web console, you can specify the context parameters through various UI options.

The following table lists the context parameters for the MSQ task engine:

Joins

Joins in multi-stage queries use one of two algorithms based on what you set the context parameter sqlJoinAlgorithm to:

• broadcast (default)
• sortMerge.

If you omit this context parameter, the MSQ task engine uses broadcast since it’s the default join algorithm. The context parameter applies to the entire SQL statement, so you can’t mix different join algorithms in the same query.

sqlJoinAlgorithm is a hint to the planner to execute the join in the specified manner. The planner can decide to ignore the hint if it deduces that the specified algorithm can be detrimental to the performance of the join beforehand. This intelligence is very limited as of now, and the sqlJoinAlgorithm set would be respected in most cases, therefore the user should set it appropriately. See the advantages and the drawbacks for the broadcast and the sort-merge join to determine which join to use beforehand.

Broadcast

The default join algorithm for multi-stage queries is a broadcast hash join, which is similar to how joins are executed with native queries.

To use broadcast joins, either omit the sqlJoinAlgorithm or set it to broadcast.

For a broadcast join, any adjacent joins are flattened into a structure with a “base” input (the bottom-leftmost one) and other leaf inputs (the rest). Next, any subqueries that are inputs the join (either base or other leafs) are planned into independent stages. Then, the non-base leaf inputs are all connected as broadcast inputs to the “base” stage.

Together, all of these non-base leaf inputs must not exceed the limit on broadcast table footprint. There is no limit on the size of the base (leftmost) input.

Only LEFT JOIN, INNER JOIN, and CROSS JOIN are supported with broadcast.

Join conditions, if present, must be equalities. It is not necessary to include a join condition; for example, CROSS JOIN and comma join do not require join conditions.

The following example has a single join chain where orders is the base input while products and customers are non-base leaf inputs. The broadcast inputs (products and customers) must fall under the limit on broadcast table footprint, but the base orders input can be unlimited in size.

The query reads products and customers and then broadcasts both to the stage that reads orders. That stage loads the broadcast inputs (products and customers) in memory and walks through orders row by row. The results are aggregated and written to the table orders_enriched.

REPLACE INTO orders_enriched
OVERWRITE ALL
SELECT
  orders.__time,
  products.name AS product_name,
  customers.name AS customer_name,
  SUM(orders.amount) AS amount
FROM orders
LEFT JOIN products ON orders.product_id = products.id
LEFT JOIN customers ON orders.customer_id = customers.id
GROUP BY 1, 2
PARTITIONED BY HOUR
CLUSTERED BY product_name

Sort-merge

You can use the sort-merge join algorithm to make queries more scalable at the cost of performance. If your goal is performance, consider broadcast joins. There are various scenarios where broadcast join would return a BroadcastTablesTooLarge error, but a sort-merge join would succeed.

To use the sort-merge join algorithm, set the context parameter sqlJoinAlgorithm to sortMerge.

In a sort-merge join, each pairwise join is planned into its own stage with two inputs. The two inputs are partitioned and sorted using a hash partitioning on the same key.

When using the sort-merge algorithm, keep the following in mind:

• There is no limit on the overall size of either input, so sort-merge is a good choice for performing a join of two large inputs or for performing a self-join of a large input with itself.

• There is a limit on the amount of data associated with each individual key. If both sides of the join exceed this limit, the query returns a TooManyRowsWithSameKey error. If only one side exceeds the limit, the query does not return this error.

• Join conditions are optional but must be equalities if they are present. For example, CROSS JOIN and comma join do not require join conditions.

• All join types are supported with sortMerge: LEFT, RIGHT, INNER, FULL, and CROSS.

The following example runs using a single sort-merge join stage that receives eventstream (partitioned on user_id) and users (partitioned on id) as inputs. There is no limit on the size of either input.

REPLACE INTO eventstream_enriched
OVERWRITE ALL
SELECT
  eventstream.__time,
  eventstream.user_id,
  eventstream.event_type,
  eventstream.event_details,
  users.signup_date AS user_signup_date
FROM eventstream
LEFT JOIN users ON eventstream.user_id = users.id
PARTITIONED BY HOUR
CLUSTERED BY user

The context parameter that sets sqlJoinAlgorithm to sortMerge is not shown in the above example.

Durable storage

SQL-based ingestion supports using durable storage to store intermediate files temporarily. Enabling it can improve reliability. For more information, see Durable storage.

Durable storage configurations

Durable storage is supported on Amazon S3 storage, Microsoft’s Azure Blob Storage and Google Cloud Storage. There are common configurations that control the behavior regardless of which storage service you use. Apart from these common configurations, there are a few properties specific to S3 and to Azure.

Common properties to configure the behavior of durable storage

To use S3 or Google for durable storage, you also need to configure the following properties:

To use Azure for durable storage, you also need to configure the following properties:

Durable storage cleaner configurations

Durable storage creates files on the remote storage, and these files get cleaned up once a job no longer requires those files. However, due to failures causing abrupt exits of tasks, these files might not get cleaned up. You can configure the Overlord to periodically clean up these intermediate files after a task completes and the files are no longer need. The files that get cleaned up are determined by the storage prefix you configure. Any files that match the path for the storage prefix may get cleaned up, not just intermediate files that are no longer needed.

Use the following configurations to control the cleaner:

Limits

Knowing the limits for the MSQ task engine can help you troubleshoot any errors that you encounter. Many of the errors occur as a result of reaching a limit.

The following table lists query limits:

Error codes

The following table describes error codes you may encounter in the multiStageQuery.payload.status.errorReport.error.errorCode field: