### ClickHouseSqlSensor Example

Source: https://context7.com/bryzgaloff/airflow-clickhouse-plugin/llms.txt

Example DAG demonstrating ClickHouseSqlSensor for monitoring event counts and ClickHouseSQLExecuteQueryOperator for inserting data.

```python
from airflow import DAG
from airflow_clickhouse_plugin.sensors.clickhouse_dbapi import ClickHouseSqlSensor
from airflow_clickhouse_plugin.operators.clickhouse_dbapi import ClickHouseSQLExecuteQueryOperator
from airflow.utils.dates import days_ago


with DAG(
    dag_id='dbapi_sensor_example',
    start_date=days_ago(2),
) as dag:
    ClickHouseSqlSensor(
        task_id='poke_events_count',
        hook_params=dict(schema='monitor'),
        sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'",
        success=lambda cnt: cnt[0][0] > 10000,
        conn_id=None,  # Use default connection
        poke_interval=60,
        timeout=3600,
    ) >> ClickHouseSQLExecuteQueryOperator(
        task_id='create_alert',
        database='alerts',
        sql='''
            INSERT INTO events SELECT eventDate, count()
            FROM monitor.warnings WHERE eventDate = '{{ ds }}'
        ''',
    )
```

--------------------------------

### ClickHouse Connection Configuration Examples

Source: https://context7.com/bryzgaloff/airflow-clickhouse-plugin/llms.txt

Examples of configuring ClickHouse connections via environment variables or connection URIs, including options for compression and secure connections.

```python
# Connection via environment variable
# AIRFLOW_CONN_CLICKHOUSE_DEFAULT=clickhouse://user:password@host:9000/database
```

```python
# Connection URI with compression
# clickhouse://login:password@host:port/?compression=lz4
```

```python
# Example extra for secure connection with compression:
# {"secure": true, "compression": "lz4"}
```

--------------------------------

### Install ClickHouse Plugin

Source: https://context7.com/bryzgaloff/airflow-clickhouse-plugin/llms.txt

Commands for installing the Airflow ClickHouse Plugin, with options for basic installation or installation with DB API 2.0 support.

```bash
# Install compression support:
# pip install clickhouse-cityhash lz4
```

```bash
# Installation commands:
# Basic: pip install airflow-clickhouse-plugin
# With DB API 2.0: pip install airflow-clickhouse-plugin[common.sql]
```

--------------------------------

### Install Compression Packages

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Install necessary packages for LZ4 compression support. This is required before configuring compression in the Airflow connection.

```bash
pip3 install clickhouse-cityhash lz4
```

--------------------------------

### ClickHouseOperator Example

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Demonstrates how to use ClickHouseOperator to execute multiple SQL statements, including INSERT, OPTIMIZE, and SELECT. The result of the last query is pushed to XCom.

```python
from airflow import DAG
from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

with DAG(
        dag_id='update_income_aggregate',
        start_date=days_ago(2),
) as dag:
    ClickHouseOperator(
        task_id='update_income_aggregate',
        database='default',
        sql=(
            '''
                INSERT INTO aggregate
                SELECT eventDt, sum(price * qty) AS income FROM sales
                WHERE eventDt = '{{ ds }}' GROUP BY eventDt
            ''', '''
                OPTIMIZE TABLE aggregate ON CLUSTER {{ var.value.cluster_name }}
                PARTITION toDate('{{ execution_date.format('%Y-%m-01') }}')
            ''', '''
                SELECT sum(income) FROM aggregate
                WHERE eventDt BETWEEN
                    '{{ execution_date.start_of('month').to_date_string() }}'
                    AND '{{ execution_date.end_of('month').to_date_string() }}'
            ''',
            # result of the last query is pushed to XCom
        ),
        # query_id is templated and allows to quickly identify query in ClickHouse logs
        query_id='{{ ti.dag_id }}-{{ ti.task_id }}-{{ ti.run_id }}-{{ ti.try_number }}',
        clickhouse_conn_id='clickhouse_test',
    ) >> PythonOperator(
        task_id='print_month_income',
        python_callable=lambda task_instance:
            # pulling XCom value and printing it
            print(task_instance.xcom_pull(task_ids='update_income_aggregate')),
    )
```

--------------------------------

### ClickHouseSensor Example

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Demonstrates using ClickHouseSensor to wait for a condition based on a SQL query result. The sensor checks if the count of warnings exceeds a threshold before proceeding.

```python
from airflow import DAG
from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor
from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
from airflow.utils.dates import days_ago


with DAG(
        dag_id='listen_warnings',
        start_date=days_ago(2),
) as dag:
    dag >> ClickHouseSensor(
        task_id='poke_events_count',
        database='monitor',
        sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'",
        is_success=lambda cnt: cnt > 10000,
    ) >> ClickHouseOperator(
        task_id='create_alert',
        database='alerts',
        sql='''
            INSERT INTO events SELECT eventDate, count()
            FROM monitor.warnings WHERE eventDate = '{{ ds }}'
        ''',
    )
```

--------------------------------

### Connection URI with Compression

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Example of an Airflow connection URI that includes compression settings. The 'compression=lz4' parameter is passed via the 'extra' field.

```bash
clickhouse://login:password@host:port/?compression=lz4
```

--------------------------------

### Run Integration Tests with Docker

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

These commands set up a local ClickHouse Docker container and run integration tests for the Airflow ClickHouse plugin. Ensure Docker is installed and running.

```bash
docker run -p 9000:9000 --ulimit nofile=262144:262144 -it clickhouse/clickhouse-server
```

```bash
PYTHONPATH=src AIRFLOW_CONN_CLICKHOUSE_DEFAULT=clickhouse://localhost python3 -m unittest discover -t tests -s integration
```

```bash
PYTHONPATH=src AIRFLOW_CONN_CLICKHOUSE_DEFAULT=clickhouse://localhost python3 -m unittest discover -s tests
```

```bash
docker exec -it $(docker run --rm -d clickhouse/clickhouse-server) bash
```

```bash
apt-get update
apt-get install -y python3 python3-pip git make
git clone https://github.com/whisklabs/airflow-clickhouse-plugin.git
cd airflow-clickhouse-plugin
python3 -m pip install -r requirements.txt
PYTHONPATH=src AIRFLOW_CONN_CLICKHOUSE_DEFAULT=clickhouse://localhost python3 -m unittest discover -s tests
```

--------------------------------

### ClickHouseHook Example: SQLite to ClickHouse Data Transfer

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Shows how to use ClickHouseHook to insert data from a SQLite table into a ClickHouse table. Ensure that INSERT queries use the `parameters` argument for values due to clickhouse-driver specifics.

```python
from airflow import DAG
from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago


def sqlite_to_clickhouse():
    sqlite_hook = SqliteHook()
    ch_hook = ClickHouseHook()
    records = sqlite_hook.get_records('SELECT * FROM some_sqlite_table')
    ch_hook.execute('INSERT INTO some_ch_table VALUES', records)


with DAG(
        dag_id='sqlite_to_clickhouse',
        start_date=days_ago(2),
) as dag:
    dag >> PythonOperator(
        task_id='sqlite_to_clickhouse',
        python_callable=sqlite_to_clickhouse,
    )
```

--------------------------------

### Import ClickHouseSensor

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Import the ClickHouseSensor class from the airflow_clickhouse_plugin.sensors.clickhouse module.

```python
from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor
```

--------------------------------

### Import ClickHouseHook

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Import the ClickHouseHook class from the airflow_clickhouse_plugin.hooks.clickhouse module.

```python
from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook
```

--------------------------------

### Airflow Connection to ClickHouse

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Instructions on how to configure an Airflow connection to a ClickHouse database.

```APIDOC
## Airflow Connection to ClickHouse

To create an Airflow connection to ClickHouse:

1.  In the Airflow UI, navigate to Admin -> Connections.
2.  Click the '+' button to add a new connection.
3.  Set the **Conn Id** (e.g., `clickhouse_default`).
4.  Set the **Conn Type** to **SQLite** or any other generic SQL database type, as there is no specific ClickHouse type.
5.  Configure the connection details:
    *   **Host**: ClickHouse host (default: `localhost`).
    *   **Port**: ClickHouse port (default: 9000).
    *   **Schema**: Database name (overridden by `database` argument in hooks/operators).
    *   **Login**: Username for ClickHouse.
    *   **Password**: Password for ClickHouse.
    *   **Extra**: For additional parameters, use a JSON object. For example, to enable secure connections, set `{"secure": true}`.

### Connection Schema Mapping

The following attributes from the Airflow Connection map to `clickhouse_driver.Client` arguments:

| Airflow Connection Attribute | `clickhouse_driver.Client.__init__` Argument |
|------------------------------|-------------------------------------------|
| `host`                       | `host`                                    |
| `port`                       | `port`                                    |
| `schema`                     | `database`                                |
| `login`                      | `user`                                    |
| `password`                   | `password`                                |
| `extra`                      | `**kwargs`                                |

### Extra Arguments Example (Compression)

To use compression (e.g., lz4):

1.  Install the required packages: `pip3 install clickhouse-cityhash lz4`.
2.  Set the `extra` field in the Airflow connection to `{"compression": "lz4"}`.

Alternatively, you can specify compression in the connection URI: `clickhouse://login:password@host:port/?compression=lz4`.
```

--------------------------------

### Interact with ClickHouse using ClickHouseHook

Source: https://context7.com/bryzgaloff/airflow-clickhouse-plugin/llms.txt

Utilize ClickHouseHook for direct database interaction within Python callables. It wraps `clickhouse_driver.Client.execute` and supports parameters, external tables, and query settings. Use `get_conn()` to access the underlying client for advanced operations. Ensure 'clickhouse_default' connection is configured.

```python
from airflow import DAG
from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago


def sqlite_to_clickhouse():
    sqlite_hook = SqliteHook()
    ch_hook = ClickHouseHook(clickhouse_conn_id='clickhouse_default')

    # Fetch records from SQLite
    records = sqlite_hook.get_records('SELECT * FROM some_sqlite_table')

    # Insert into ClickHouse (values must be passed via parameters)
    ch_hook.execute('INSERT INTO some_ch_table VALUES', records)

    # Execute SELECT query with parameters
    result = ch_hook.execute(
        'SELECT * FROM users WHERE created_at > %(date)s',
        params={'date': '2024-01-01'}
    )

    # Get underlying client for advanced operations
    client = ch_hook.get_conn()
    client.execute('SELECT 1')


with DAG(
    dag_id='sqlite_to_clickhouse',
    start_date=days_ago(2),
) as dag:
    PythonOperator(
        task_id='sqlite_to_clickhouse',
        python_callable=sqlite_to_clickhouse,
    )
```

--------------------------------

### Execute ClickHouse SQL Queries with ClickHouseOperator

Source: https://context7.com/bryzgaloff/airflow-clickhouse-plugin/llms.txt

Use ClickHouseOperator to run SQL queries in Airflow. Supports Jinja templating, multiple queries, and XCom push. Queries can be strings or loaded from .sql files. Ensure 'clickhouse_default' connection is configured.

```python
from airflow import DAG
from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

with DAG(
    dag_id='update_income_aggregate',
    start_date=days_ago(2),
) as dag:
    ClickHouseOperator(
        task_id='update_income_aggregate',
        database='default',
        clickhouse_conn_id='clickhouse_default',
        sql=(
            '''
                INSERT INTO aggregate
                SELECT eventDt, sum(price * qty) AS income FROM sales
                WHERE eventDt = '{{ ds }}' GROUP BY eventDt
            ''',
            '''
                OPTIMIZE TABLE aggregate ON CLUSTER {{ var.value.cluster_name }}
                PARTITION toDate('{{ execution_date.format('%Y-%m-01') }}')
            ''',
            '''
                SELECT sum(income) FROM aggregate
                WHERE eventDt BETWEEN
                    '{{ execution_date.start_of('month').to_date_string() }}'
                    AND '{{ execution_date.end_of('month').to_date_string() }}'
            ''',
        ),
        # Query ID for tracking in ClickHouse logs
        query_id='{{ ti.dag_id }}-{{ ti.task_id }}-{{ ti.run_id }}-{{ ti.try_number }}',
    ) >> PythonOperator(
        task_id='print_month_income',
        python_callable=lambda task_instance:
            print(task_instance.xcom_pull(task_ids='update_income_aggregate')),
    )
```

--------------------------------

### Execute SQL with ClickHouseSQLExecuteQueryOperator

Source: https://context7.com/bryzgaloff/airflow-clickhouse-plugin/llms.txt

Execute arbitrary SQL queries against ClickHouse using a DB API 2.0 compliant operator. This is useful for migrating pipelines and maintaining a consistent SQL interface across different databases.

```python
from airflow import DAG
from airflow_clickhouse_plugin.operators.clickhouse_dbapi import (
    ClickHouseSQLExecuteQueryOperator,
    ClickHouseSQLColumnCheckOperator,
    ClickHouseSQLTableCheckOperator,
)
from airflow.utils.dates import days_ago


with DAG(
    dag_id='clickhouse_dbapi_example',
    start_date=days_ago(2),
) as dag:
    # Execute SQL query
    execute_task = ClickHouseSQLExecuteQueryOperator(
        task_id='execute_query',
        database='default',
        sql='''
            INSERT INTO events SELECT eventDate, count()
            FROM monitor.warnings WHERE eventDate = '{{ ds }}'
        ''',
    )

    # Column-level data quality checks
    column_check = ClickHouseSQLColumnCheckOperator(
        task_id='column_check',
        table='events',
        column_mapping={
            'event_count': {
                'min': {'geq_to': 0},
                'max': {'leq_to': 1000000},
            },
        },
    )

    # Table-level data quality checks
    table_check = ClickHouseSQLTableCheckOperator(
        task_id='table_check',
        table='events',
        checks={
            'row_count_check': {'check_statement': 'COUNT(*) > 0'},
        },
    )

    execute_task >> column_check >> table_check
```

--------------------------------

### ClickHouseOperator Reference

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Reference for the ClickHouseOperator, detailing its parameters and functionality.

```APIDOC
## ClickHouseOperator reference

To import `ClickHouseOperator` use `from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator`.

Supported arguments:
* `sql` (templated, required): query (if argument is a single `str`) or multiple queries (iterable of `str`). Supports files with `.sql` extension.
* `clickhouse_conn_id`: Airflow connection id. Connection schema is described [below](#clickhouse-connection-schema). Default connection id is `clickhouse_default`.
* Arguments of [`clickhouse_driver.Client.execute` method][ch-driver-execute-summary]:
  * `parameters` (templated): passed `params` of the `execute` method. (Renamed to avoid name conflict with Airflow tasks' `params` argument.)
    * `dict` for `SELECT` queries.
    * `list`/`tuple`/generator for `INSERT` queries.
    * If multiple queries are provided via `sql` then the `parameters` are passed to _all_ of them.
  * `with_column_types` (not templated).
  * `external_tables` (templated).
  * `query_id` (templated).
  * `settings` (templated).
  * `types_check` (not templated).
  * `columnar` (not templated).
  * For the documentation of these arguments, refer to [`clickhouse_driver.Client.execute` API reference][ch-driver-execute-reference].
* `database` (templated): if present, overrides `schema` of Airflow connection.
* Other arguments (including a required `task_id`) are inherited from Airflow [BaseOperator][airflow-base-op].

Result of the _last_ query is pushed to XCom (disable using `do_xcom_push=False` argument).

In other words, the operator simply wraps [`ClickHouseHook.execute` method](#clickhousehook-reference).

See [example](#clickhouseoperator-example) below.
```

--------------------------------

### ClickHouseHook Reference

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Reference for the ClickHouseHook class, used for interacting with ClickHouse from Airflow.

```APIDOC
## ClickHouseHook

To import `ClickHouseHook` use `from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook`.

### Constructor (`__init__`)

Supported kwargs:
* `clickhouse_conn_id` (string): Airflow connection id for ClickHouse. Defaults to `clickhouse_default`.
* `database` (string): Overrides the `schema` from the Airflow connection. Defaults to connection's schema.

### Methods

* `execute(sql, **kwargs)`:
  Wraps `clickhouse_driver.Client.execute`. Accepts all arguments of `clickhouse_driver.Client.execute` except for `query` (uses `sql` instead). Returns the result of the last query.

* `get_conn()`:
  Returns an instance of `clickhouse_driver.Client`.

### Example Usage

```python
from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook

hook = ClickHouseHook(clickhouse_conn_id='my_clickhouse_conn', database='my_database')
result = hook.execute("SELECT count() FROM my_table")
print(result)
```
```

--------------------------------

### ClickHouseSensor Reference

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

Reference for the ClickHouseSensor class, used to monitor ClickHouse for specific conditions.

```APIDOC
## ClickHouseSensor

To import `ClickHouseSensor` use `from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor`.

This sensor wraps `ClickHouseHook.execute` and supports all arguments of `ClickHouseOperator`, plus:

### Constructor (`__init__`)

* `is_success` (callable): A function that accepts the return value of `ClickHouseHook.execute`. If it returns truthy, the sensor succeeds. Defaults to `bool`.
* `is_failure` (callable): A function that accepts the return value of `ClickHouseHook.execute`. If it returns truthy, the sensor raises an `AirflowException`. Defaults to `None` (no failure check).

### Example Usage

```python
from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor

sensor = ClickHouseSensor(
    task_id='wait_for_data',
    clickhouse_conn_id='my_clickhouse_conn',
    sql='SELECT count() FROM my_table WHERE status = \'processed\'',
    is_success=lambda result: result and result[0][0] > 0  # Succeeds if count > 0
)
```
```

--------------------------------

### Use ClickHouseSensor to Wait for Query Results

Source: https://context7.com/bryzgaloff/airflow-clickhouse-plugin/llms.txt

Use this sensor to pause an Airflow task until a ClickHouse query returns a truthy result. Customize success and failure conditions using lambda functions based on query output. Ensure the ClickHouse connection ID is correctly configured.

```python
from airflow import DAG
from airflow_clickhouse_plugin.sensors.clickhouse import ClickHouseSensor
from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
from airflow.utils.dates import days_ago


with DAG(
    dag_id='listen_warnings',
    start_date=days_ago(2),
) as dag:
    ClickHouseSensor(
        task_id='poke_events_count',
        database='monitor',
        clickhouse_conn_id='clickhouse_default',
        sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'",
        # Succeed when count exceeds threshold
        is_success=lambda cnt: cnt[0][0] > 10000,
        # Fail if data quality issue detected
        is_failure=lambda cnt: cnt[0][0] < 0,
        poke_interval=60,
        timeout=3600,
    ) >> ClickHouseOperator(
        task_id='create_alert',
        database='alerts',
        sql='''
            INSERT INTO events SELECT eventDate, count()
            FROM monitor.warnings WHERE eventDate = '{{ ds }}'
        ''',
    )
```

--------------------------------

### Use ClickHouseDbApiHook for DB API 2.0 Connections

Source: https://context7.com/bryzgaloff/airflow-clickhouse-plugin/llms.txt

Utilize the ClickHouseDbApiHook to establish DB API 2.0 compliant connections to ClickHouse. This hook allows you to interact with ClickHouse using standard Python DB API 2.0 methods, compatible with various tools and operators.

```python
from airflow import DAG
from airflow_clickhouse_plugin.hooks.clickhouse_dbapi import ClickHouseDbApiHook
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago


def query_with_dbapi_hook():
    hook = ClickHouseDbApiHook(
        clickhouse_conn_id='clickhouse_default',
        schema='default'
    )

    # Get DB API 2.0 compliant connection
    conn = hook.get_conn()
    cursor = conn.cursor()

    # Execute query using standard DB API interface
    cursor.execute("SELECT * FROM users WHERE active = 1")
    rows = cursor.fetchall()

    for row in rows:
        print(row)

    cursor.close()
    conn.close()


with DAG(
    dag_id='dbapi_hook_example',
    start_date=days_ago(2),
) as dag:
    PythonOperator(
        task_id='query_clickhouse',
        python_callable=query_with_dbapi_hook,
    )
```

--------------------------------

### Airflow DAG with ClickHouse Sensor and Operator

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

This DAG uses ClickHouseSqlSensor to check for a condition and ClickHouseSQLExecuteQueryOperator to insert data. Ensure your Airflow environment is configured with the ClickHouse connection.

```python
from airflow import DAG
from airflow_clickhouse_plugin.sensors.clickhouse_dbapi import ClickHouseSqlSensor
from airflow_clickhouse_plugin.operators.clickhouse_dbapi import ClickHouseSQLExecuteQueryOperator
from airflow.utils.dates import days_ago


with DAG(
        dag_id='listen_warnings',
        start_date=days_ago(2),
) as dag:
    dag >> ClickHouseSqlSensor(
        task_id='poke_events_count',
        hook_params=dict(schema='monitor'),
        sql="SELECT count() FROM warnings WHERE eventDate = '{{ ds }}'",
        success=lambda cnt: cnt > 10000,
        conn_id=None,  # required by common.sql SqlSensor; use None for default
    ) >> ClickHouseSQLExecuteQueryOperator(
        task_id='create_alert',
        database='alerts',
        sql='''
            INSERT INTO events SELECT eventDate, count() 
            FROM monitor.warnings WHERE eventDate = '{{ ds }}'
        ''',
    )
```

--------------------------------

### Import ClickHouseOperator

Source: https://github.com/bryzgaloff/airflow-clickhouse-plugin/blob/master/README.md

To use the ClickHouseOperator, import it from the airflow_clickhouse_plugin.operators.clickhouse module. This operator allows you to execute ClickHouse queries within Airflow DAGs.

```python
from airflow_clickhouse_plugin.operators.clickhouse import ClickHouseOperator
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.