Integration between ADQM and S3

Elena Dvoryadkina

Contents

Integration parameters
- S3 engine and table function parameters
- Additional settings
S3 table engine
- Create a table
- Example
Table functions
- Example
Use named collections
- Parameter keys
- Example
Wildcards in a path to S3 files
- Examples
Partitioned data writes
Virtual columns

For integration with S3, ClickHouse supports the S3 table engine and special table functions, which allow reading and writing S3 data from ClickHouse.

S3 files for test examples

In examples this article provides, a bucket in Yandex Object Storage named test-bucket is used as an S3 storage. For testing the integration of ADQM and S3, the following objects (files) are loaded into this bucket:

test_1.txt:
"one",1 "two",2 "three",3
test_2.txt:
"ten",10 "twenty",20 "thirty",30

To logically structure data and group files within the bucket, the adqm prefix (directory) is used. Thus, the files are accessible from ADQM at the following URLs:

https://storage.yandexcloud.net/test-bucket/adqm/test_1.txt
https://storage.yandexcloud.net/test-bucket/adqm/test_2.txt

Integration parameters

S3 engine and table function parameters

To create a table based on the S3 engine and call table functions for integration with S3 (the basic syntax of corresponding queries is provided in sections below), specify the following parameters.

<path>

Bucket URL with a path to a file. When using a table or function for reading data from S3, you can specify wildcards in the path to address multiple files

<access_key_id>, <secret_access_key>

Access key identifier and secret key for authenticating requests to an S3 storage. These parameters are optional — you can specify credentials in the configuration file or enable the use_environment_credentials option to retrieve them from environment variables.

If a bucket is public, you can set the NOSIGN keyword instead of the access key parameters — in this case, requests will not be signed

<session_token>

Session token for a temporary access key, which can be optionally used with the specified <access_key_id> and <secret_access_key>

<format>

File format. To get a full list of formats ADQM/ClickHouse supports for input and output data, see the Formats for input and output data article of the ClickHouse documentation

<compression>

Compression method. Possible values:

none;
gzip or gz;
brotli or br;
xz or LZMA;
zstd or zst.

By default, the compression method is automatically detected by the file extension

<headers>

S3 request headers in the format headers('key1'='value1', 'key2'='value2', …)

TIP

For production environments, it is recommended to pass parameters using named collections.

Additional settings

Parameters at the query/session level

ADQM also supports the following additional parameters related to S3 integration that you can configure before executing a query or specify in a configuration file at the user level:

SELECT name, default FROM system.settings WHERE startsWith(name, 's3_');

    ┌─name───────────────────────────────────────────────┬─default────┐
 1. │ s3_strict_upload_part_size                         │ 0          │
 2. │ s3_min_upload_part_size                            │ 16777216   │
 3. │ s3_max_upload_part_size                            │ 5368709120 │
 4. │ s3_upload_part_size_multiply_factor                │ 2          │
 5. │ s3_upload_part_size_multiply_parts_count_threshold │ 500        │
 6. │ s3_max_part_number                                 │ 10000      │
 7. │ s3_max_single_operation_copy_size                  │ 33554432   │
 8. │ s3_max_inflight_parts_for_one_file                 │ 20         │
 9. │ s3_max_single_part_upload_size                     │ 33554432   │
10. │ s3_max_single_read_retries                         │ 4          │
11. │ s3_max_unexpected_write_error_retries              │ 4          │
12. │ s3_max_redirects                                   │ 10         │
13. │ s3_max_connections                                 │ 1024       │
14. │ s3_max_get_rps                                     │ 0          │
15. │ s3_max_get_burst                                   │ 0          │
16. │ s3_max_put_rps                                     │ 0          │
17. │ s3_max_put_burst                                   │ 0          │
18. │ s3_list_object_keys_size                           │ 1000       │
19. │ s3_use_adaptive_timeouts                           │ 1          │
20. │ s3_truncate_on_insert                              │ 0          │
21. │ s3_create_new_file_on_insert                       │ 0          │
22. │ s3_skip_empty_files                                │ 0          │
23. │ s3_check_objects_after_upload                      │ 0          │
24. │ s3_allow_parallel_part_upload                      │ 1          │
25. │ s3_throw_on_zero_files_match                       │ 0          │
26. │ s3_ignore_file_doesnt_exist                        │ 0          │
27. │ s3_validate_request_settings                       │ 1          │
28. │ s3_disable_checksum                                │ 0          │
29. │ s3_retry_attempts                                  │ 100        │
30. │ s3_request_timeout_ms                              │ 30000      │
31. │ s3_connect_timeout_ms                              │ 1000       │
    └────────────────────────────────────────────────────┴────────────┘

See parameter descriptions in the Session Settings article of the ClickHouse documentation.

Endpoint-related settings

In the configuration file (/etc/clickhouse-server/config.xml or another XML file located in the /etc/clickhouse-server/config.d/ directory), you can configure parameters for a specific endpoint inside the <s3> tag — these parameters will be used in queries where the prefix of the specified URL exactly matches this endpoint.

Endpoint configuration in general terms:

<clickhouse>
    <s3>
        <endpoint_config>
            <endpoint>ENDPOINT_URL</endpoint>
            <access_key_id>ACCESS_KEY_ID</access_key_id>
            <secret_access_key>SECRET_ACCESS_KEY</secret_access_key>
            <!-- Other endpoint-related parameters -->
        </endpoint_config>
    </s3>
</clickhouse>

where:

endpoint_config — configuration name.
ENDPOINT_URL — S3 endpoint URL, including a bucket URL and path to a directory where data is stored (prefix). Required.
ACCESS_KEY_ID, SECRET_ACCESS_KEY — credentials to access the endpoint. These parameters are optional — for example, you can skip them if a bucket is public or if the option to read credentials from environment variables is enabled (see the next section).

See the full list of settings for an S3 endpoint in the Endpoint-based Settings section of the ClickHouse documentation.

Retrieve S3 credentials from the environment

Instead of explicitly specifying credentials to access an S3 bucket in your queries or configuration files, you can enable the use_environment_credentials option in ADQM to obtain credentials from the AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN environment variables and Amazon EC2 metadata:

at the S3 endpoint level:

<clickhouse>
    <s3>
        <endpoint_config>
            <use_environment_credentials>true</use_environment_credentials>
            <!-- Other endpoint-related parameters -->
        </endpoint_config>
    </s3>
</clickhouse>

at the global level:

<clickhouse>
    <s3>
        <use_environment_credentials>true</use_environment_credentials>
    </s3>
</clickhouse>

S3 table engine

The S3 table engine allows you to execute SELECT and INSERT queries on data stored in S3.

Create a table

Below is the basic syntax of a query that creates an S3 table:

CREATE TABLE <table_name> (<column_name> <column_type>, ...)
ENGINE = S3('<path>' [, NOSIGN | '<access_key_id>', '<secret_access_key>' [, '<session_token>']]
            [, '<format>'] [, '<compression>'] [, <headers>])
[PARTITION BY <expr>]
[SETTINGS <setting_name> = <setting_value>, ...];

Parameters are described above.

The engine does not support ALTER and SELECT…SAMPLE queries, as well as indexes.

Example

Create an S3 table

In the clickhouse-client console client, create a table based on the S3 engine with a structure corresponding to data to be imported from S3:

CREATE TABLE s3_table (name String, value UInt32)
ENGINE=S3('https://storage.yandexcloud.net/test-bucket/adqm/test_1.txt',
          '<access_key_id>',
          '<secret_access_key>',
          'CSV');

where <access_key_id> and <secret_access_key> are credentials to access the test-bucket S3 bucket.

Select data from the table:

SELECT * FROM s3_table;

   ┌─name──┬─value─┐
1. │ one   │     1 │
2. │ two   │     2 │
3. │ three │     3 │
   └───────┴───────┘

Insert data into an S3 table

Enable the s3_create_new_file_on_insert setting to create a new file in S3 on each data insert in an ADQM table based on the S3 engine (for example, if an S3 table has been created for the file_name.txt file, names of new files will be generated according to the pattern — file_name.1.txt, file_name.2.txt, and so on):
SET s3_create_new_file_on_insert = 1;

Insert data into the table by repeating the INSERT INTO query twice:

INSERT INTO s3_table VALUES('four', 4);

INSERT INTO s3_table VALUES('five', 5);

Check the bucket contents — two new files (test_1.1.txt with the "four",4 line and test_1.2.txt with "five",5) have been added to the adqm directory.

On subsequent reading the table, you can make sure that a sample includes data from all files:

SELECT * FROM s3_table;

┌─name──┬─value─┐
│ one   │     1 │
│ two   │     2 │
│ three │     3 │
└───────┴───────┘
┌─name─┬─value─┐
│ five │     5 │
└──────┴───────┘
┌─name─┬─value─┐
│ four │     4 │
└──────┴───────┘

Activate the s3_truncate_on_insert option and perform another data insertion into the table — this will overwrite the last data file:
SET s3_truncate_on_insert = 1;
INSERT INTO s3_table VALUES('six', 6);
Check the bucket contents. The list of files in the adqm directory has not changed, but the test_1.2.txt file has been overwritten — it now contains the "six",6 line.

If the s3_truncate_on_insert setting is enabled, new data will replace the current contents of an existing file in any case, regardless of the s3_create_new_file_on_insert value.

Table functions

ADQM also provides two table functions for integration with S3:

s3 — creates a table for reading/writing data in S3.

Function syntax:
s3('<path>' [, NOSIGN | '<access_key_id>', '<secret_access_key>' [, '<session_token>']] [, '<format>'] [, '<structure>'] [, '<compression>'] [, <headers>])
where <structure> is a table structure in the format column_name1 data_type1, column_name2 data_type2, …. Other parameters are described above.
s3Cluster — allows processing S3 files in parallel from multiple nodes of a specified cluster. On an initiator node, the function creates a connection to all nodes in the cluster, discloses asterisks in the S3 file path, and dispatches each file dynamically. On a worker node, it requests the next task from the initiator and processes it. This is repeated until all tasks are completed.

Function syntax:
s3Cluster('<cluster_name>', '<path>' [, NOSIGN | '<access_key_id>', '<secret_access_key>' [, '<session_token>']] [, '<format>'] , '<structure>' [, '<compression>'] [, <headers>])
where <cluster_name> is a name of a cluster used to build a set of addresses and connection parameters to remote and local servers. Other parameters are similar to parameters of the s3 function, but specifying the table structure is required.

Example

Read data from S3

Run the following query to read data from the test_2.txt S3 file with the s3 function:

SELECT * FROM s3('https://storage.yandexcloud.net/test-bucket/adqm/test_2.txt',
                 '<access_key_id>',
                 '<secret_access_key>',
                 'CSV',
                 'name String, value UInt32');

   ┌─name───┬─value─┐
1. │ ten    │    10 │
2. │ twenty │    20 │
3. │ thirty │    30 │
   └────────┴───────┘

Write data to S3

When inserting data through the s3 function with the s3_truncate_on_insert setting enabled, existing data in a file referenced by the function will be replaced with new data.

For example, run the following query (the s3_truncate_on_insert value was set to 1 in the above example for the S3 table engine):

INSERT INTO FUNCTION s3('https://storage.yandexcloud.net/test-bucket/adqm/test_2.txt',
                        '<access_key_id>', '<secret_access_key>',
                        'CSV')
VALUES ('s3_func_test_string_1', 100);

Check that the test_2.txt file has been updated:

SELECT * FROM s3('https://storage.yandexcloud.net/test-bucket/adqm/test_2.txt',
                 '<access_key_id>',
                 '<secret_access_key>',
                 'CSV');

   ┌─c1────────────────────┬──c2─┐
1. │ s3_func_test_string_1 │ 100 │
   └───────────────────────┴─────┘

If the s3_truncate_on_insert parameter value is set to 0 and the s3_create_new_file_on_insert setting is enabled, a new file will be created when inserting data into S3 through the s3 function. For example, if the function inserts data into the file_name.txt file, it will create a file named file_name.1.txt to save new data or overwrite the file_name.1.txt file if it already exists.

Update the settings related to data insertion and call the s3 function to export data to the test_2.txt file:

SET s3_truncate_on_insert = 0, s3_create_new_file_on_insert = 1;

INSERT INTO FUNCTION s3('https://storage.yandexcloud.net/test-bucket/adqm/test_2.txt',
                        '<access_key_id>', '<secret_access_key>',
                        'CSV')
VALUES ('s3_func_test_string_2', 200);

As a result, the data will be saved to a new file named test_2.1.txt:

SELECT * FROM s3('https://storage.yandexcloud.net/test-bucket/adqm/test_2.1.txt',
                 '<access_key_id>',
                 '<secret_access_key>',
                 'CSV');

   ┌─c1────────────────────┬──c2─┐
1. │ s3_func_test_string_2 │ 200 │
   └───────────────────────┴─────┘

Use named collections

To avoid passing all S3 storage connection parameters each time you create an S3 table or call a table function, you can specify them once in a named collection as a list of key/value pairs. This approach also allows you to hide sensitive S3 integration credentials from users without administrator access rights.

Parameter keys

Main keys in an S3-related named collection correspond to names of parameters described above: url, access_key_id, secret_access_key, session_token, format, compression (or compression_method), structure, use_environment_credentials. Also, the following extra parameters are supported.

filename

File name that is appended to the URL

no_sign_request

Specifies whether to ignore credentials so that requests are not signed (for accessing public buckets)

expiration_window_seconds

Period for verifying the expiration of credentials

max_connections

Maximum number of connections per server

max_single_read_retries

Maximum number of attempts during a single S3 read

min_upload_part_size

Minimum object part size to upload during multipart upload to S3

upload_part_size_multiply_parts_count_threshold

Each time this number of object parts have been uploaded to S3, the value of min_upload_part_size is multiplied by upload_part_size_multiply_factor

upload_part_size_multiply_factor

Factor by which the min_upload_part_size value is multiplied each time a number of object parts equal to the value of upload_part_size_multiply_parts_count_threshold have been uploaded to S3 by a single write operation

max_single_part_upload_size

Maximum object size for singlepart upload to S3

Example

Create a named collection with parameters for connecting to the S3 bucket:

CREATE NAMED COLLECTION test_bucket_creds AS
url = 'https://storage.yandexcloud.net/test-bucket/adqm/',
access_key_id = '<access_key_id>',
secret_access_key = '<secret_access_key>';

Use the named collection to call the s3 table function (a data file has not been defined in the named collection, so it should be specified separately in the function parameters):

SELECT * FROM s3(test_bucket_creds, filename = 'test_1.txt');

   ┌─c1────┬─c2─┐
1. │ one   │  1 │
2. │ two   │  2 │
3. │ three │  3 │
   └───────┴────┘

Create an S3 table using the named collection and read data from the table:

CREATE TABLE s3_table_nc (name String, value UInt32)
ENGINE=S3(test_bucket_creds, filename = 'test_1.txt');

SELECT * FROM s3_table_nc;

   ┌─name──┬─value─┐
1. │ one   │     1 │
2. │ two   │     2 │
3. │ three │     3 │
   └───────┴───────┘

Wildcards in a path to S3 files

If an ADQM table should accept data from a bunch of S3 files and be read-only, you can use the following wildcards to specify the path to multiple S3 files in the path parameter of the S3 table engine or the s3/s3Cluster table function:

* — substitutes any number of any characters except /, including an empty string;
** — substitutes any number of any characters, including / and an empty string (this pattern can be used for recursive directory traversal);
? — substitutes any single character;
{first_string,second_string,third_one} — substitutes any of strings 'first_string', 'second_string', 'third_one' (you can also use numbers in this pattern — for example, {1,3,5});
{n..m} — substitutes any number in the [n, m] range. If the list of files contains a number range with leading zeros in file names, use the general pattern {0n..0m} or the separate pattern {n..m} (or the sign ?) for each digit in a file name.

Wildcard characters can be used in multiple path components (for example, in a directory name and in a file name). Only existing files whose paths and names exactly match the pattern are processed. The list of files is determined during the SELECT (not CREATE) operation.

Examples

Import data from the test_1.txt and test_2.txt above files into ADQM in one of the following ways:

SELECT * FROM s3(test_bucket_creds, filename = 'test_{1,2}.txt');

SELECT * FROM s3(test_bucket_creds, filename = 'test_?.txt');

Read data from all files in the adqm directory:

SELECT * FROM s3(test_bucket_creds, filename = '*');

Partitioned data writes

If you specify the PARTITION BY clause when creating an S3 table, a separate file will be created for each partition key value in an S3 storage when you insert data into the table. For example, it may be useful to split data into separate files for further transfer to another system — since ClickHouse data sets are often very large, and network reliability is sometimes insufficient, it makes sense to transfer data sets in relatively small parts.

A partition key can also be specified for exporting data from ADQM to S3 via a table function: INSERT INTO FUNCTION s3(…) PARTITION BY <expr> VALUES ….

NOTE

It is not recommended to use too granular data partition (for example, more granular than by month).
Partitioning data does not make queries run faster (in contrast to the ORDER BY clause). For example, do not partition your data by user IDs or user names — list the ID or user name column first in the ORDER BY clause instead.
Currently, selecting data directly from partitioned S3 tables is not supported, but you can retrieve data from individual partitions using the s3 table function.

Below is an example of splitting data inserted into an S3 table by values of a column that has a low cardinality (a small number of unique values).

Create an S3 table, specifying the value column as a partition key and using the {_partition_id} parameter in the file name:

CREATE TABLE s3_table_partition (name String, value UInt32)
ENGINE=S3('https://storage.yandexcloud.net/test-bucket/adqm/test_partition_{_partition_id}.txt',
          '<access_key_id>',
          '<secret_access_key>',
          'CSV')
PARTITION BY value;

Insert data into the table:

INSERT INTO s3_table_partition
VALUES ('one_1', 1), ('two_1', 2), ('one_2', 1), ('two_2', 2), ('three_1', 3);

As a result, data is written to three files, where the {_partition_id} parameter has been replaced with the corresponding partition key value in each file’s name: test_partition_1.txt, test_partition_2.txt, and test_partition_3.txt. The parameter can also be specified in a name of a directory (prefix).

Use the s3 table function to read data from individual files, for example:

SELECT * FROM s3('https://storage.yandexcloud.net/test-bucket/adqm/test_partition_1.txt',
                 '<access_key_id>',
                 '<secret_access_key>',
                 'CSV');

   ┌─c1────┬─c2─┐
1. │ one_1 │  1 │
2. │ one_2 │  1 │
   └───────┴────┘

SELECT * FROM s3('https://storage.yandexcloud.net/test-bucket/adqm/test_partition_2.txt',
                 '<access_key_id>',
                 '<secret_access_key>',
                 'CSV');

   ┌─c1────┬─c2─┐
1. │ two_1 │  2 │
2. │ two_2 │  2 │
   └───────┴────┘

However, if you try to get data directly from the s3_table_partition table, an error message will be displayed:

Exception: Reading from a partitioned S3 storage is not implemented yet. (NOT_IMPLEMENTED)

Virtual columns

The following virtual columns are automatically included to tables based on the S3 engine and tables returned by the s3 and s3Cluster table functions.

Column name

Data Type

Description

_path

LowCardinality(String)

File path

_file

LowCardinality(String)

File name

_size

Nullable(UInt64)

File size in bytes

_time

Nullable(DateTime)

Time of the last file change

_etag

LowCardinality(String)

File ETag

Use virtual columns in SELECT queries to get information about S3 files referenced by an ADQM table. For example, the following query returns a list of all .txt files in the adqm directory of the test-bucket bucket and calculates the number of rows in each file:

SELECT _file, count() AS count
FROM s3('https://storage.yandexcloud.net/test-bucket/adqm/*.txt', '<access_key_id>', '<secret_access_key>', 'CSV')
GROUP BY _file
ORDER BY _file;

   ┌─_file────────────────┬─count─┐
1. │ test_1.1.txt         │     1 │
2. │ test_1.2.txt         │     1 │
3. │ test_1.txt           │     3 │
4. │ test_2.1.txt         │     1 │
5. │ test_2.txt           │     1 │
6. │ test_partition_1.txt │     2 │
7. │ test_partition_2.txt │     2 │
8. │ test_partition_3.txt │     1 │
   └──────────────────────┴───────┘

Found a mistake? Seleсt text and press Ctrl+Enter to report it