Add Iceberg write-amplification experiment

experimentation/iceberg/.gitignore

@@ -0,0 +1,9 @@
+# Isolated interpreter for this experiment (see requirements.txt).
+.venv/
+__pycache__/
+
+# Jetstream captures are ~21 MiB per 10 minutes and are reproducible by
+# re-running capture.py. The report and results JSON are committed; the
+# per-call CSV is not, since it holds one row per AWS request.
+data/captures/
+data/results/*-calls.csv

experimentation/iceberg/README.md

@@ -0,0 +1,130 @@
+# Iceberg write-amplification experiment
+
+Prices the four S3-facing operations in an Iceberg ingest pipeline fed by the
+Bluesky Jetstream firehose:
+
+1. **Write records to S3** — the raw-Parquet control, no table format.
+2. **Update Iceberg metadata** — the same rows through `table.append()`.
+3. **Compaction** — scan, collapse each AT-URI to its latest state, drop delete
+ tombstones, rewrite one file per partition.
+4. **Metadata cleaning** — expire snapshots, then sweep orphaned objects.
+
+Each runs inside a metered phase, so the output is an exact per-operation ledger
+of S3 and Glue calls, latency, and cost — not an estimate.
+
+## How it measures
+
+PyIceberg's default `PyArrowFileIO` drives a C++ S3 client that Python cannot
+intercept. The catalog therefore pins `py-io-impl=pyiceberg.io.fsspec.FsspecFileIO`,
+routing every request through s3fs → aiobotocore → botocore, where
+`s3_meter.Meter` counts it.
+
+`Meter.install()` wraps `botocore.session.Session.__init__`, so *every* session —
+boto3's for raw writes, aiobotocore's inside s3fs, and the Glue client PyIceberg
+builds for catalog commits — carries the handlers. It must run before any client
+is constructed; `run_experiment.py` does this at import time.
+
+Two things worth knowing about the numbers:
+
+- **Glue commits are not S3 calls.** Swapping `metadata_location` is a Glue
+ `UpdateTable`, billed at $1/100k rather than $5/100k. Metering only S3 would
+ miss the commit path entirely, so Glue is counted as its own tier.
+- **Request bytes come from the body stream, not a header.** botocore has no
+ `Content-Length` on the request dict at `before-call`, and large uploads switch
+ to `aws-chunked` encoding which carries none at all. `_body_size` measures the
+ `BytesIO` directly and restores its position.
+
+## Setup
+
+pyiceberg pins `rich<15` and the root project requires `rich>=15`, so this
+experiment cannot share the root venv:
+
+```bash
+uv venv --python 3.11 experimentation/iceberg/.venv
+uv pip install --python experimentation/iceberg/.venv -r experimentation/iceberg/requirements.txt
+```
+
+## Running
+
+Capture and replay are separate on purpose. A 10-minute firehose capture is never
+reproducible, so it is recorded once and replayed as many times as needed — that
+way every write-path variant is measured against byte-identical input.
+
+```bash
+# 1. Capture (writes data/captures/jetstream-<stamp>.jsonl.gz)
+./experimentation/iceberg/.venv/bin/python -m experimentation.iceberg.capture --seconds 600
+
+# 2. Replay through both write paths
+./experimentation/iceberg/.venv/bin/python -m experimentation.iceberg.run_experiment \
+ --capture experimentation/iceberg/data/captures/jetstream-<stamp>.jsonl.gz
+```
+
+Useful flags: `--flush-seconds` (default 60), `--max-batches`, `--skip-raw`,
+`--run-id`.
+
+Outputs land in `data/results/<run_id>-{report.md,results.json,calls.csv}`.
+`calls.csv` is every individual API call, for slicing outside the report.
+
+## Layout
+
+```
+s3://lab-data-integrations-interface/experiments/iceberg/<run_id>/
+ raw/ # baseline Parquet, no table format
+ warehouse/ # Iceberg tables
+```
+
+Glue tables are `<record_type>_<run_id>` in the `iceberg_experiments` database,
+so repeat runs never collide.
+
+## Data model
+
+Four tables — `posts`, `likes`, `reposts`, `follows` — each partitioned by
+`days(created_at)`. Separate tables mean 4x the metadata commits per flush, which
+is itself one of the findings.
+
+Bluesky `createdAt` is client-supplied. Anything more than 24h from the broker's
+`time_us` falls back to ingest time, and the report splits the fallbacks into
+their two very different causes:
+
+- **`delete` events** carry no record body, so they have no `createdAt` at all.
+ Structural, not a data problem — and the large majority of fallbacks.
+- **Skewed timestamps** parse cleanly but sit far from the broker clock. In the
+ measured capture these were overwhelmingly *one* archive-import bot stamping
+ genuine historical dates (2011, 2013, 2016…). Those dates are arguably
+ correct; the rule rewrites them so a single bot cannot open a daily partition
+ per historical date it touches. That is a deliberate correctness-for-file-count
+ trade, not a data-cleaning step.
+
+## Duplicates vs. lifecycle collapses
+
+These are counted separately because they are constantly conflated:
+
+- A **redelivered duplicate** is the identical event twice — same `uri` *and*
+ same `cid`. A stable 10-minute capture contained **zero** of these.
+- A **lifecycle collapse** is several distinct events about one record (create
+ then delete, create then update), each with its own `cid`. Collapsing these
+ materialises current state; it is not deduplication.
+
+Compaction keeps the latest row per URI and then **drops `delete` tombstones**.
+Note this only cancels a create that is in the same table — a delete of a record
+written before the table existed has nothing to reconcile against and is simply
+discarded. A tombstone also lands in the partition of its *ingest* day, not the
+partition of the record it deletes, so cross-partition deletes need equality
+deletes or merge-on-read to work properly.
+
+## Cleanup
+
+```bash
+aws s3 rm s3://lab-data-integrations-interface/experiments/iceberg/<run_id>/ --recursive
+python -c "from experimentation.iceberg import catalog; \
+ catalog.drop_tables(catalog.build_catalog('<run_id>'), '<run_id>')"
+```
+
+## Tests
+
+```bash
+./experimentation/iceberg/.venv/bin/python -m pytest experimentation/iceberg/tests
+```
+
+The suite is offline — no AWS, no network. A `conftest.py` skips collection when
+the root interpreter picks it up, since that venv has no pyiceberg.

experimentation/iceberg/capture.py

@@ -0,0 +1,122 @@
+"""Phase 1 -- capture the Bluesky Jetstream firehose to a local file.
+
+Deliberately decoupled from the S3/Iceberg write path. A 10-minute capture is
+expensive to re-collect and never reproducible, so it is recorded once and then
+replayed as many times as needed. That keeps every write-path variant (flush
+interval, partition spec, compaction strategy) measured against byte-identical
+input.
+
+Usage:
+ python -m experimentation.iceberg.capture --seconds 600
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import gzip
+import json
+import time
+from collections import Counter
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlencode
+
+import websockets
+
+from experimentation.iceberg import constants
+
+CAPTURE_DIR = Path(__file__).parent / "data" / "captures"
+
+
+def build_endpoint() -> str:
+ """Jetstream subscribe URL filtered to the four collections we care about."""
+ query = urlencode([("wantedCollections", nsid) for nsid in constants.COLLECTIONS])
+ return f"{constants.JETSTREAM_ENDPOINT}?{query}"
+
+
+def _tally(message: str, counts: Counter[str]) -> None:
+ """Update per-record-type counters from one raw frame."""
+ counts["total"] += 1
+ try:
+ event = json.loads(message)
+ except json.JSONDecodeError:
+ counts["unparseable"] += 1
+ return
+ commit = event.get("commit")
+ if not isinstance(commit, dict):
+ return
+ record_type = constants.COLLECTIONS.get(commit.get("collection", ""))
+ if record_type:
+ counts[record_type] += 1
+
+
+async def _drain(socket: Any, handle: Any, counts: Counter[str], deadline: float) -> None:
+ """Write frames verbatim to ``handle`` until ``deadline``, tallying as we go."""
+ started = time.monotonic()
+ while time.monotonic() < deadline:
+ remaining = deadline - time.monotonic()
+ try:
+ message = await asyncio.wait_for(socket.recv(), timeout=remaining)
+ except TimeoutError:
+ return
+
+ text = message if isinstance(message, str) else message.decode("utf-8")
+ handle.write(text)
+ handle.write("\n")
+ _tally(text, counts)
+
+ if counts["total"] % 20_000 == 0:
+ rate = counts["total"] / max(time.monotonic() - started, 1e-9)
+ print(f" {counts['total']:,} events {rate:,.0f}/s")
+
+
+async def capture(seconds: int, output_path: Path) -> dict[str, int]:
+ """Stream Jetstream for ``seconds`` and write raw events as gzipped JSONL.
+
+ Returns a per-record-type count. Events are written exactly as received so
+ the replay stage owns all parsing -- a schema change should never require
+ re-capturing.
+ """
+ output_path.parent.mkdir(parents=True, exist_ok=True)
+ endpoint = build_endpoint()
+ counts: Counter[str] = Counter()
+ started = time.monotonic()
+
+ print(f"connecting to {endpoint}")
+ print(f"capturing for {seconds}s -> {output_path}")
+
+ # max_size=None: some posts with large embeds exceed the 1MiB default frame cap.
+ async with websockets.connect(endpoint, max_size=None) as socket:
+ with gzip.open(output_path, "wt", encoding="utf-8") as handle:
+ await _drain(socket, handle, counts, started + seconds)
+
+ counts["elapsed_seconds"] = int(time.monotonic() - started)
+ return dict(counts)
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description="Capture Bluesky Jetstream to a local file.")
+ parser.add_argument("--seconds", type=int, default=constants.DEFAULT_CAPTURE_SECONDS)
+ parser.add_argument("--output", type=Path, default=None)
+ args = parser.parse_args()
+
+ stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
+ output_path = args.output or CAPTURE_DIR / f"jetstream-{stamp}.jsonl.gz"
+
+ counts = asyncio.run(capture(args.seconds, output_path))
+
+ size_mb = output_path.stat().st_size / 1024 / 1024
+ print(f"\ncaptured {counts.get('total', 0):,} events in {counts.get('elapsed_seconds', 0)}s")
+ for record_type in constants.RECORD_TYPES:
+ print(f" {record_type:<10} {counts.get(record_type, 0):>10,}")
+ print(f"compressed size: {size_mb:.1f} MiB")
+ print(f"wrote {output_path}")
+
+ metadata_path = output_path.with_suffix(".meta.json")
+ metadata_path.write_text(json.dumps(counts, indent=2))
+
+
+if __name__ == "__main__":
+ main()

experimentation/iceberg/catalog.py

@@ -0,0 +1,102 @@
+"""Glue catalog wiring for the experiment.
+
+Two things here are load-bearing:
+
+1. ``py-io-impl`` is pinned to ``FsspecFileIO``. PyIceberg defaults to
+ ``PyArrowFileIO``, whose S3 client lives in C++ and is invisible to the
+ meter. Fsspec routes through aiobotocore, so every request is countable.
+2. Tables are named ``<record_type>_<run_id>`` inside a single dedicated Glue
+ database, so repeat runs never collide and cleanup is one prefix delete.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pyiceberg.catalog import Catalog
+from pyiceberg.catalog.glue import GlueCatalog
+from pyiceberg.exceptions import NamespaceAlreadyExistsError, NoSuchTableError
+
+from experimentation.iceberg import constants, schemas
+
+
+def warehouse_uri(run_id: str) -> str:
+ return f"s3://{constants.S3_BUCKET}/{constants.S3_EXPERIMENT_PREFIX}/{run_id}/warehouse"
+
+
+def raw_uri(run_id: str) -> str:
+ return f"s3://{constants.S3_BUCKET}/{constants.S3_EXPERIMENT_PREFIX}/{run_id}/raw"
+
+
+def table_name(record_type: str, run_id: str) -> str:
+ return f"{record_type}_{run_id}"
+
+
+def build_catalog(run_id: str) -> Catalog:
+ """Construct the Glue-backed catalog for this run."""
+ return GlueCatalog(
+ name="iceberg_experiment",
+ **{
+ "warehouse": warehouse_uri(run_id),
+ "glue.region": constants.AWS_REGION,
+ "s3.region": constants.AWS_REGION,
+ # Required for the meter to see anything -- see module docstring.
+ "py-io-impl": "pyiceberg.io.fsspec.FsspecFileIO",
+ },
+ )
+
+
+def ensure_namespace(catalog: Catalog) -> None:
+ try:
+ catalog.create_namespace(constants.GLUE_DATABASE)
+ except NamespaceAlreadyExistsError:
+ pass
+
+
+def create_tables(catalog: Catalog, run_id: str) -> dict[str, Any]:
+ """Create one partitioned table per record type. Returns record_type -> Table."""
+ ensure_namespace(catalog)
+ tables: dict[str, Any] = {}
+
+ for record_type in constants.RECORD_TYPES:
+ identifier = (constants.GLUE_DATABASE, table_name(record_type, run_id))
+ tables[record_type] = catalog.create_table(
+ identifier=identifier,
+ schema=schemas.SCHEMAS[record_type],
+ partition_spec=schemas.PARTITION_SPEC,
+ location=f"{warehouse_uri(run_id)}/{record_type}",
+ properties={
+ "format-version": "2",
+ "write.parquet.compression-codec": "zstd",
+ # Leave stale metadata.json files in place so the expiry phase
+ # has real work to measure.
+ "write.metadata.delete-after-commit.enabled": "false",
+ },
+ )
+ return tables
+
+
+def load_tables(catalog: Catalog, run_id: str) -> dict[str, Any]:
+ """Load existing tables for a run, skipping any that were never created."""
+ tables: dict[str, Any] = {}
+ for record_type in constants.RECORD_TYPES:
+ try:
+ tables[record_type] = catalog.load_table(
+ (constants.GLUE_DATABASE, table_name(record_type, run_id))
+ )
+ except NoSuchTableError:
+ continue
+ return tables
+
+
+def drop_tables(catalog: Catalog, run_id: str) -> list[str]:
+ """Drop this run's Glue tables. Does not remove the S3 objects behind them."""
+ dropped = []
+ for record_type in constants.RECORD_TYPES:
+ name = table_name(record_type, run_id)
+ try:
+ catalog.drop_table((constants.GLUE_DATABASE, name))
+ dropped.append(name)
+ except NoSuchTableError:
+ continue
+ return dropped