Connecter Apache Airflow à ClickHouse

Apache Airflow est une plateforme open-source permettant de définir, d’ordonnancer et de superviser des workflows sous forme de code. Les workflows sont définis comme des graphes orientés acycliques (DAG) de tâches écrites en Python.

Le provider apache-airflow-providers-clickhousedb connecte Airflow à ClickHouse, ce qui vous permet d’exécuter des requêtes, de créer des tables et de charger des données dans le cadre d’un DAG. Il se connecte via l’interface HTTP à l’aide du client clickhouse-connect, et expose ClickHouse via le framework SQL commun d’Airflow, de sorte que l’opérateur standard SQLExecuteQueryOperator gère les requêtes DDL, DML et analytiques, sans nécessiter d’opérateur spécifique à ClickHouse.

Installez le provider dans l’environnement où s’exécutent le scheduler Airflow et les workers :

pip install apache-airflow-providers-clickhousedb

Le provider dépend de apache-airflow-providers-common-sql et de clickhouse-connect, qui sont installés en même temps que lui. Pour transmettre les résultats de la requête à des DataFrames pandas ou polars, installez les extras facultatifs :

pip install 'apache-airflow-providers-common-sql[pandas,polars]'

Le provider enregistre un type de connexion clickhouse. Créez une connexion depuis l'UI d'Airflow, dans Admin > Connections, ou définissez-en une via la CLI ou une variable d'environnement.

Dans l'UI, sélectionnez ClickHouse comme type de connexion et renseignez les champs suivants :

Pour ClickHouse Cloud ou tout cluster auto-hébergé avec TLS activé, définissez secure sur true dans le champ Extra et utilisez le port TLS (8443).

Le provider expose des options supplémentaires sous forme de champs dédiés dans le formulaire de connexion. Si vous définissez la connexion via un URI, du JSON ou une variable d’environnement, indiquez-les plutôt comme clés dans l’objet JSON extra. Elles sont toutes facultatives :

Configurez la connexion à l’aide d’une variable d’environnement. Le format URI inclut l’hôte, les identifiants et la base de données :

export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='clickhouse://default:password@localhost:8123/my_database'

Tous les éléments de l’URI doivent être codés en URL. Pour TLS, les délais d’expiration ou les paramètres de session, utilisez le format JSON, qui expose les champs Extra :

export AIRFLOW_CONN_CLICKHOUSE_DEFAULT='{
    "conn_type": "clickhouse",
    "host": "abc123.clickhouse.cloud",
    "port": 8443,
    "login": "default",
    "password": "secret",
    "schema": "my_database",
    "extra": {
        "secure": true,
        "session_settings": {
            "max_execution_time": 300,
            "max_memory_usage": 10000000000
        }
    }
}'

Tous les hooks et les opérateurs utilisent l’identifiant de connexion clickhouse_default, sauf si vous en indiquez un autre.

Définissez le conn_id de l’opérateur sur votre connexion à ClickHouse. Le DAG suivant crée une table, y insère des lignes, les relit, puis supprime la table :

from datetime import datetime

from airflow import DAG
from airflow.providers.common.sql.hooks.sql import fetch_all_handler
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

CLICKHOUSE_CONN_ID = "clickhouse_default"
CLICKHOUSE_TABLE = "airflow_example"

with DAG(
    dag_id="example_clickhouse",
    start_date=datetime(2021, 1, 1),
    default_args={"conn_id": CLICKHOUSE_CONN_ID},
    schedule="@once",
    catchup=False,
) as dag:
    create_table = SQLExecuteQueryOperator(
        task_id="create_table",
        sql=f"""
            CREATE TABLE IF NOT EXISTS {CLICKHOUSE_TABLE} (
                id   UInt32,
                name String,
                ts   DateTime DEFAULT now()
            ) ENGINE = MergeTree()
            ORDER BY id
        """,
    )

    insert_rows = SQLExecuteQueryOperator(
        task_id="insert_rows",
        sql=f"""
            INSERT INTO {CLICKHOUSE_TABLE} (id, name) VALUES
                (1, 'Alice'),
                (2, 'Bob'),
                (3, 'Charlie')
        """,
    )

    read_rows = SQLExecuteQueryOperator(
        task_id="read_rows",
        sql=f"SELECT id, name FROM {CLICKHOUSE_TABLE} ORDER BY id",
        handler=fetch_all_handler,
    )

    drop_table = SQLExecuteQueryOperator(
        task_id="drop_table",
        sql=f"DROP TABLE IF EXISTS {CLICKHOUSE_TABLE}",
    )

    create_table >> insert_rows >> read_rows >> drop_table

Le résultat de la requête est récupéré avec le handler par défaut (fetch_all_handler). Pour renvoyer autre chose que l’ensemble complet des résultats, transmettez un autre handler, par exemple fetch_one_handler pour récupérer uniquement la première ligne.

Lorsqu'une connexion pointe vers un cluster et que certaines tâches interrogent différentes bases de données, remplacez la base de données via hook_params au lieu de créer une connexion distincte :

read_rows = SQLExecuteQueryOperator(
    task_id="read_rows",
    conn_id=CLICKHOUSE_CONN_ID,
    sql="SELECT count() FROM events",
    hook_params={"database": "analytics"},
)

Pour les opérations qui ne se prêtent pas à un opérateur SQL — insertions en masse, streaming ou appels client propres à ClickHouse — utilisez ClickHouseHook dans une tâche Python.

La méthode bulk_insert_rows du hook utilise le mécanisme d’insertion colonnaire natif de clickhouse-connect, bien plus rapide que les insertions ligne par ligne pour de grands jeux de données. Définissez batch_size afin de limiter le pic de mémoire sur des entrées très volumineuses :

from airflow.providers.clickhousedb.hooks.clickhouse import ClickHouseHook

hook = ClickHouseHook(clickhouse_conn_id="clickhouse_default")

hook.bulk_insert_rows(
    table="events",
    rows=[("user1", "click"), ("user2", "view")],
    column_names=["user_id", "action"],
    batch_size=1000,
)

Appelez get_client() pour accéder au client clickhouse-connect sous-jacent afin de gérer tout ce que le hook n’expose pas directement :

client = hook.get_client()
total = client.query("SELECT count() FROM events").result_rows[0][0]

Transmettez des paramètres de session lors de l’initialisation du hook, soit directement, soit via le hook_params d’un opérateur. Les paramètres transmis au constructeur sont fusionnés avec les session_settings définis dans le champ Extra de la connexion, et les valeurs du constructeur prévalent en cas de conflit de clés :

hook = ClickHouseHook(
    clickhouse_conn_id="clickhouse_default",
    session_settings={"max_execution_time": 60, "max_threads": 4},
)

• Client Python clickhouse-connect
• Interface HTTP de ClickHouse
• Référence des paramètres de session de ClickHouse
• Documentation de référence pour apache-airflow-providers-clickhousedb
• Paquet du provider sur PyPI