L4 demo: the same data, stored columnar

L4 demo: the same data, stored columnar#

In L3 we put the Intel Berkeley Lab readings in a relational database and made a selective query fast with an index. This notebook keeps the exact same data and changes only where it lives: into columnar Parquet, queried by DuckDB. We ask one analytical question three ways and compare, then use DuckDB to query the Parquet with no import step.

The data is fetched from a mirror on first run.

Data: Intel Lab Data, ~2.3M readings from 54 motes, carried over from L3.

Run this first on Colab#

Colab starts from its own preinstalled environment rather than this course’s uv environment, so run the cell below before anything else. It installs what this notebook needs and Colab does not already have. Outside Colab it does nothing, so you can run it or skip it.

# Run this first on Colab. Anywhere else this cell does nothing.
#
# Only genuinely missing packages are installed, so Colab's own versions of
# everything it already ships are left alone.
import importlib.util
import subprocess
import sys

REQUIREMENTS = {
    "duckdb": "duckdb",
    "pandas": "pandas",
}


def _missing(module):
    try:
        return importlib.util.find_spec(module) is None
    except ModuleNotFoundError:  # the parent package is absent
        return True


if "google.colab" in sys.modules:
    need = sorted({pip for mod, pip in REQUIREMENTS.items() if _missing(mod)})
    if need:
        print("installing:", " ".join(need))
        subprocess.run([sys.executable, "-m", "pip", "install", "-q", *need], check=True)
    print("Colab setup done." if need else "Colab: nothing to install.")

1. Load the readings#

Same loader as L3: a whitespace-separated export with no header, rows not in time order, and short or malformed rows we skip. We keep the six columns we care about.

import io
import sqlite3
import time
import urllib.request
import zipfile
from pathlib import Path

import duckdb
import pandas as pd

DATA = Path('data/data.txt')
URL = 'https://raw.githubusercontent.com/linsea423/Intel_Lab_Data/master/data.zip'
COLS = ['date', 'time', 'epoch', 'moteid',
        'temperature', 'humidity', 'light', 'voltage']


def load(path: Path = DATA) -> pd.DataFrame:
    path.parent.mkdir(parents=True, exist_ok=True)
    if not path.exists():
        print(f'fetching {URL}')
        with urllib.request.urlopen(URL) as r:
            payload = r.read()
        with zipfile.ZipFile(io.BytesIO(payload)) as z:
            path.write_bytes(z.read('data.txt'))
    # sep=' ' and not r'\s+'. The delimiter is one space, and a dropped
    # channel leaves its field empty; collapsing runs of whitespace shifts
    # every following value one column left on the 93,879 rows that have
    # one, silently. L3's notebook shows the damage side by side.
    df = pd.read_csv(path, sep=' ', names=COLS, header=None, engine='c')
    df['ts'] = pd.to_datetime(df['date'] + ' ' + df['time'],
                              format='mixed', errors='coerce')
    df = df.dropna(subset=['ts', 'moteid'])
    df['moteid'] = df['moteid'].astype(int)
    return df[['moteid', 'ts', 'temperature', 'humidity', 'light', 'voltage']]


readings = load()
print(f'{len(readings):,} readings, {readings.moteid.nunique()} motes, '
      f'{readings.ts.min().date()} to {readings.ts.max().date()}')
readings.head(3)

2. Write it to Parquet, and compare size#

Parquet stores the table column by column, each column compressed on its own. Write the readings once as a single Parquet file and once as CSV, and compare the bytes on disk. One type of similar values per column is what compresses so well.

readings.to_parquet('data/readings.parquet', engine='pyarrow', compression='snappy')
readings.to_csv('data/readings.csv', index=False)

parq_mb = Path('data/readings.parquet').stat().st_size / 1e6
csv_mb = Path('data/readings.csv').stat().st_size / 1e6
print(f'CSV {csv_mb:.0f} MB   vs   Parquet {parq_mb:.0f} MB   ({csv_mb / parq_mb:.1f}x smaller)')

3. The same question, three ways#

Average temperature per mote over the whole table: a wide scan of one column across every row, exactly the query an index cannot help. We time each engine warmed (run once, then take the best of a few), so we are comparing steady-state work, not first-call overhead.

def bench(fn, n=5):
    fn()  # warm up: parse, open files, fill caches
    best = float('inf')
    for _ in range(n):
        s = time.perf_counter()
        fn()
        best = min(best, (time.perf_counter() - s) * 1000)
    return best


# pandas: the data is already in memory; compute in Python
ms = bench(lambda: readings.groupby('moteid')['temperature'].mean())
print(f'pandas (in memory)      {ms:6.0f} ms')

# SQLite: a row store queried with SQL. It reads every column of every row.
con = sqlite3.connect(':memory:')
con.execute('CREATE TABLE readings (moteid INT, temperature REAL, '
            'humidity REAL, light REAL, voltage REAL)')
con.executemany(
    'INSERT INTO readings VALUES (?,?,?,?,?)',
    readings[['moteid', 'temperature', 'humidity', 'light', 'voltage']]
    .itertuples(index=False, name=None))
con.commit()
Q = 'SELECT moteid, avg(temperature) FROM readings GROUP BY moteid'
print(f'SQLite (row store)      {bench(lambda: con.execute(Q).fetchall()):6.0f} ms')

# DuckDB over Parquet: a column store. It reads only moteid and temperature,
# vectorized, straight from the file with no import step.
duck_q = ("SELECT moteid, avg(temperature) "
          "FROM 'data/readings.parquet' GROUP BY moteid")
print(f'DuckDB + Parquet        {bench(lambda: duckdb.sql(duck_q).fetchall()):6.0f} ms')
duckdb.sql(duck_q).df().head(3)

All three return the same answer. pandas is fast because the data already sits in memory; the fair storage-to-storage comparison is the row store against the column store, and DuckDB over Parquet wins the analytical scan by a wide margin because it touches two columns of six and processes them in vectorized batches. The notes’ figure isolates this under controlled warmup; your exact numbers will vary with hardware, but the direction is the lesson.

4. DuckDB’s zero-import reach#

DuckDB queried the Parquet with no CREATE TABLE and no load. Two more moves it makes for free. Partitioning: write a directory tree keyed by date, and a filter on the date opens only the matching folders. COPY TO: stream a query result straight out to a new Parquet file.

# write a date-partitioned copy: readings_parquet/date=2004-03-01/...
part = readings.assign(date=readings['ts'].dt.date.astype(str))
part.to_parquet('readings_parquet', partition_cols=['date'],
                engine='pyarrow', compression='snappy')

# hive_partitioning exposes the folder's `date` as a column; the filter prunes folders
duckdb.sql("""
    SELECT count(*) AS n, round(avg(temperature), 2) AS avg_temp
    FROM read_parquet('readings_parquet/**/*.parquet', hive_partitioning = true)
    WHERE date = '2004-03-01'
""").show()

# COPY a query result straight to a new Parquet file, no dataframe round-trip
duckdb.sql("""
    COPY (SELECT moteid, avg(temperature) AS avg_temp
          FROM 'data/readings.parquet' GROUP BY moteid)
    TO 'data/mote_avg.parquet' (FORMAT parquet)
""")
print('wrote data/mote_avg.parquet')

Takeaway#

The row store from L3 (OLTP) is built for writes and point lookups; the column store here (OLAP) is built for scanning a few columns over the whole history, and it is much faster and smaller for exactly that. Real platforms run both and move data across the seam on purpose. Assignment A2 has you load the same dataset into PostgreSQL and into Parquet/DuckDB and compare, so the second half starts here.