Implement Clickhouse Destination

[Pipboyguy]

Objective

Implement Clickhouse as a destination in the dlt library.

Features:

1. Default Clickhouse configuration via dlt resources adapter:

   • Provide a way to configure the Clickhouse connection details using dlt's configuration system using the clickhouse-connect driver. (clickhouse_connect.driver)
   • Allow users to specify additional connection-specific settings if needed.

2. Schema mapping and data type compatibility:

   • Map dlt's schema to the appropriate Clickhouse data types.
   • The sqlalchemy dialect contained in (clickhouse_connect.cc_sqlalchemy) could be used to define these mappings. The docs mention that it is limited though.
   • Handle necessary data type conversions to ensure compatibility between dlt's data and Clickhouse's supported types.
   • Handle Clickhouse's definition of primary keys.

3. Efficient data loading:

   • Utilize Clickhouse's bulk loading capabilities to optimize data ingestion performance.
   • Explore options for parallel data loading to leverage Clickhouse's distributed architecture.

4. Support for Clickhouse-specific features:

   • Provide support for Clickhouse-specific features that can enhance data loading and querying, such as:
     • Partitioning and sharding strategies
     • Materialized views
     • Aggregation and compression settings

5. SQL client integration:

   • Ensure integration with dlt's SQL client to enable users to query and interact with the loaded data in Clickhouse.

Development Tasks:

• Define the interface and configuration options for the Clickhouse destination.
• Implement the core functionality for schema mapping, data loading, and destination-specific features.
• Write unit and integration tests to ensure the correctness and reliability of the Clickhouse destination in test/load/pipelines.
• Update documentation and provide examples showcasing the usage of the Clickhouse destination in dlt pipelines.

Requested Community Feedback

We would appreciate any feedback or suggestions from the community regarding:

• Clickhouse features or optimizations that would be valuable to support in the dlt destination.
• Use cases or scenarios where a Clickhouse destination would be beneficial.
• Any potential challenges or considerations to keep in mind when implementing the Clickhouse destination.

Your input will help us prioritize the development efforts and ensure that the Clickhouse destination meets the needs of the dlt community. Thank you for your feedback.

[rudolfix]

@Pipboyguy
please take a look at the update guide on new destinations. I bet you'll be able to reuse most of the code of other similar warehouses ie. databricks or snowflake. https://dlthub.com/docs/walkthroughs/create-new-destination

what you need is a db-api client: https://clickhouse-driver.readthedocs.io/en/latest/dbapi.html

the base class (sql_job_client and sql_client) will handle 90% of things for you, type mapping is just (mostly) writing a dict.

my biggest question is how people load data to clickhouse so they can use this on production. this will affect the scope the most:

• do people use staging and load files with COPY jobs
• maybe clickhouse cloud has internal staging?
• maybe people do INSERT statements?

we should answer the above. for all those options we have scaffolding code that should be quick to adapt. we can also greatly help with writting tests once you have prototype that loads data with a few test pipelines

[Pipboyguy]

@rudolfix ClickHouse's primary method for bulk data ingestion is through the use of table functions, for example the s3 table function. There doesn't seem to be any concept of staging tables like in Snowflake, just direct inserts from source in a streaming fashion:

When using the s3 table function with INSERT INTO...SELECT, data is read and inserted in a streaming fashion. Only a few blocks of data reside in memory while the blocks are continuously read from S3 and pushed into the destination table.

It is possible to save credentials and associated metadata from a source in a table engines, but this is simply a convenience utility.

ClickHouse has a collection of table functions that build tables. The main ingestion pattern in production, according the tutorials, example datasets and docs, is to insert the response from a table function into a table using the command

INSERT INTO table_name select from table_function(...);

Asynchronous, direct inserts are supported as well, but again not the main method.

This considered, I've opted to adapt the snowflake implementation (without staging).

dlt/dlt/destinations/impl/clickhouse/clickhouse.py

Line 89 in de66efa

class ClickhouseLoadJob(LoadJob, FollowupJob):

I'll also reuse as much of SF's tests considering its similarity.

[arniwesth]

Hey - Clickhouse user here. Thanks a lot for providing this destination❤️.

To give a short background, we use Clickhouse partly as a traditional data warehouse for BI, partly for analyzing medium-large scale time series data sets (~100s GB each) and partly (experimentally) as a feature store for ML features.

Features that are important to us (sorted by importance):

• Being able to specify Settings: This is wherestorage_policy is defined and we use both local disks and S3 for storage depending on the type of data
• Being able to specify compression codecs: Careful selection of compression codecs can result in +20x compression depending on the nature of the data. We are seeing huge compression ratios when moving data from Postgres to Clickhouse
• Being able to specify ORDER BY: This has implications for joins of large tables and query performance

My base assumption from looking at the destination implementations is that we will fairly easily be able to modify at least some of these features, but they are probably also important to the broader community.

[Pipboyguy]

Hi @arniwesth

Thank you for your input!

You're right that Settings and comporession codec is easy change.

Order by (sorting key) is a little trickier to wrap, because the primary key must be a prefix of the sorting key if both are specified. The sorting key is implictly set in any case to be equal to primary key.

I'll see if I can add Settings and comporession codec to the destination, but can't promise sorting key (for now). Please see #1097 for further details.