Connecter Apache Airflow à ClickHouse

Orchestrez les requêtes ClickHouse et les chargements de données depuis Apache Airflow à l’aide du provider 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.

Vous utilisez déjà le airflow-clickhouse-plugin communautaire ? Il repose sur un driver, un protocole et un port différents. Suivez le guide de migration pour transférer vos DAG et connexions existants vers le provider.

​

Installer le provider

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]'

​

Créer une connexion à ClickHouse

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 :

Champ	Description	Par défaut
Host	Nom d’hôte du serveur ClickHouse, par exemple abc123.clickhouse.cloud	localhost
Port	Port HTTP(S)	8123 (en clair), 8443 (TLS)
Login	Nom d’utilisateur ClickHouse	default
Password	Mot de passe de l’utilisateur ClickHouse	(vide)
Database	Base de données par défaut de la connexion. Dans l’UI, ce champ s’appelle Database ; il correspond au champ schema lorsque vous définissez la connexion via un URI ou du JSON.	default

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).

​

Options de connexion supplémentaires

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 :

clé extra	Champ de l’UI	Par défaut	Description
secure	Utiliser TLS (HTTPS)	false	Active HTTPS/TLS.
verify	Vérifier le certificat SSL	true	Vérifie le certificat TLS du serveur lorsque secure vaut true. Définissez cette option sur false pour les certificats autosignés.
connect_timeout	Délai d’expiration de connexion (secondes)	10	Délai d’expiration de la connexion HTTP, en secondes.
send_receive_timeout	Délai d’expiration de requête (secondes)	300	Délai d’expiration en lecture/écriture des requêtes, en secondes. Augmentez cette valeur pour les requêtes analytiques de longue durée.
compress	Activer la compression LZ4	true	Active la compression LZ4 des résultats.
client_name	Nom du client	(vide)	Libellé ajouté à l’identifiant de version d’Airflow dans l’en-tête HTTP User-Agent, enregistré dans la colonne http_user_agent de system.query_log.
session_settings	Paramètres de session (JSON)	(vide)	Paramètres de session ClickHouse appliqués à chaque requête de la connexion, par exemple {"max_execution_time": 300, "max_threads": 8}.
client_kwargs	kwargs du client (JSON)	(vide)	Arguments nommés supplémentaires transmis à clickhouse_connect.get_client(), par exemple un http_proxy.

​

Définir une connexion sans l’UI

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.

​

Exécuter des requêtes avec SQLExecuteQueryOperator

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.handlers 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.

​

Utiliser une base de données différente pour chaque tâche

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"},
)

​

Utiliser directement le hook

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]

​

Appliquer les paramètres de session

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},
)

​

Contenu associé

• Migrer depuis airflow-clickhouse-plugin
• 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