ClickHouse: Python (clickhouse-connect)

Loads the demo dataset and prints revenue per country over the ClickHouse HTTP interface. The program is ordinary clickhouse-connect code.

The only change

# CLICKHOUSE_HOST
 clickhouse.internal
 warehouse.internal
# CLICKHOUSE_PORT
 8443
 8443 (ClickHouse HTTP over TLS)
# CLICKHOUSE_USER / CLICKHOUSE_PASSWORD
 ClickHouse user
 MachWarehouse user (same name if migrated)

Run

python3 -m venv .venv && . .venv/bin/activate
pip install -r requirements.txt
CLICKHOUSE_HOST=warehouse.internal CLICKHOUSE_PASSWORD=... python main.py

Library: clickhouse-connect 1.11.0 (Python 3.10+).

Source

The program as it is kept and tested in the Mach Database Warehouse repository; highlighted lines read the connection setting.

main.py

"""Load the demo events and print revenue per country with clickhouse-connect.

Plain ClickHouse client code: nothing here knows about MachWarehouse.
"""

import csv
import datetime as dt
import decimal
import os
import pathlib

import clickhouse_connect

DATASET = pathlib.Path(__file__).resolve().parents[2] / "dataset"


def rows():
    with open(DATASET / "events.csv", newline="") as f:
        for r in csv.DictReader(f):
            yield [
                dt.date.fromisoformat(r["event_date"]),
                dt.datetime.fromisoformat(r["event_time"]),
                int(r["user_id"]),
                r["country"],
                r["device"],
                decimal.Decimal(r["revenue"]),
            ]


def main() -> None:
    client = clickhouse_connect.get_client(
        host=os.environ.get("CLICKHOUSE_HOST", "localhost"),
        port=int(os.environ.get("CLICKHOUSE_PORT", "8443")),
        username=os.environ.get("CLICKHOUSE_USER", "default"),
        password=os.environ["CLICKHOUSE_PASSWORD"],
        database=os.environ.get("CLICKHOUSE_DATABASE", "default"),
        secure=os.environ.get("CLICKHOUSE_SECURE", "true") == "true",
    )
    client.command((DATASET / "clickhouse.sql").read_text())
    client.insert(
        "events",
        list(rows()),
        column_names=["event_date", "event_time", "user_id", "country", "device", "revenue"],
    )
    result = client.query(
        "SELECT country, count() AS events, sum(revenue) AS revenue "
        "FROM events GROUP BY country ORDER BY revenue DESC"
    )
    for country, events, revenue in result.result_rows:
        print(f"{country}\t{events}\t{revenue}")
    # Columnar results straight into pandas, as with ClickHouse:
    # df = client.query_df("SELECT * FROM events")


if __name__ == "__main__":
    main()

requirements.txt

clickhouse-connect==1.11.0