Week 4 · Load to the warehouse › Day 25

Day 25. Postgres, DuckDB and ClickHouse side by side

1. Why this matters

Your later mission is low-cost stacks for medium companies. That means knowing the open-source warehouses by hand, not by blog post. Today you load the same tables into three of them and measure.

2. Postgres

docker run -d --name pg -e POSTGRES_PASSWORD=pg -p 5432:5432 postgres:16

import psycopg, io
with psycopg.connect("postgresql://postgres:pg@localhost:5432/postgres") as con, con.cursor() as cur:
    cur.execute("create table if not exists stg_ads_load (like fct_ads_daily including defaults)")
    cur.execute("truncate stg_ads_load")
    buf = io.StringIO(); df.write_csv(buf); buf.seek(0)
    with cur.copy("copy stg_ads_load from stdin with (format csv, header true)") as cp:
        cp.write(buf.read())                                          # COPY: the fast path
    cur.execute("""
      insert into fct_ads_daily select * from stg_ads_load
      on conflict (ad_date, campaign_id) do update set
        spend = excluded.spend, clicks = excluded.clicks, conversions = excluded.conversions, updated_at = now()
      where (fct_ads_daily.spend, fct_ads_daily.clicks, fct_ads_daily.conversions)
            is distinct from (excluded.spend, excluded.clicks, excluded.conversions)""")

3. DuckDB

con = duckdb.connect("data/wh.duckdb")
con.sql("""create table if not exists fct_ads_daily (
  ad_date date, campaign_id varchar, spend decimal(18,2), clicks integer, impressions bigint,
  conversions integer, conversion_value decimal(18,2), updated_at timestamp,
  primary key (ad_date, campaign_id))""")
con.sql("""insert or replace into fct_ads_daily          -- upsert: needs the primary key above
  select ad_date, campaign_id, spend, clicks, impressions, conversions, conversion_value, now()
  from 'data/staging/ads/*.parquet'""")

DuckDB can also attach Postgres (attach 'postgresql://…' as pg (type postgres)) and query or write it directly, which is useful for heavy scans over a Postgres warehouse.

4. ClickHouse

docker run -d --name ch -p 8123:8123 -p 9000:9000 -e CLICKHOUSE_PASSWORD=ch clickhouse/clickhouse-server

create table fct_ads_daily (
  ad_date Date, campaign_id LowCardinality(String), spend Decimal(18,2), clicks UInt32,
  impressions UInt64, conversions UInt32, conversion_value Decimal(18,2), updated_at DateTime64(3)
) engine = ReplacingMergeTree(updated_at)
order by (campaign_id, ad_date);

ReplacingMergeTree keeps the row with the latest updated_at per sort key, eventually (during background merges). Queries that need exact answers now use FINAL or argMax. That's the ClickHouse trade-off: blazing scans and inserts, unusual update semantics.

import clickhouse_connect
ch = clickhouse_connect.get_client(host="localhost", username="default", password="ch")
ch.insert_df("fct_ads_daily", df.to_pandas())
ch.query("select campaign_id, sum(spend) from fct_ads_daily final group by campaign_id")

5. One interface, many warehouses

class Warehouse(dg.ConfigurableResource):
    def upsert(self, table: str, df: pl.DataFrame, keys: list[str]) -> dict: ...
    def replace_partition(self, table: str, df: pl.DataFrame, col: str, value) -> dict: ...
    def query(self, sql: str) -> pl.DataFrame: ...

class AzureSqlWarehouse(Warehouse): ...   # Day 23–24 code
class PostgresWarehouse(Warehouse): ...   # COPY + ON CONFLICT
class DuckDBWarehouse(Warehouse): ...

The asset calls warehouse.upsert(...); the environment chooses which.