Iceberg setup

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## 2026-07-30
 
 1. Gated the UI behind Supabase email/password auth (issue #124): a login page, a protected route group that redirects unauthenticated visitors to sign-in and returns them to their intended destination, and a sign-out control showing the signed-in user's email. Access is invite-only, with users added directly in Supabase. Also removed job-polling status flicker. [PR #140](https://github.com/METResearchGroup/lab_data_integrations_interface/pull/140)
+2. Bluesky Jetstream ingestion now commits to Iceberg tables in the Glue catalog (`bluesky_raw`) instead of writing Parquet to disk, retrying transient commit failures and dead-lettering batches to `s3://lab-data-integrations-interface/dead_letter/` when they cannot land. Replaced the `data_platform` Terraform stack with `terraform/bluesky_ingestion_jetstream/`, which destroyed the previous Glue database, both Athena workgroups, and the pipeline-runs DynamoDB table that `backend/routes/posts.py` still references. [PR #141](https://github.com/METResearchGroup/lab_data_integrations_interface/pull/141)
 
 ## 2026-07-29
 

bluesky_ingestion_jetstream/aws/__init__.py

@@ -0,0 +1 @@
+"""AWS-facing pieces of the Jetstream pipeline: Glue catalog and Iceberg tables."""

bluesky_ingestion_jetstream/aws/bootstrap.py

@@ -0,0 +1,97 @@
+"""One-shot creation of the four Iceberg tables. Run by hand, not by the ingester.
+
+ python -m bluesky_ingestion_jetstream.aws.bootstrap
+
+Idempotent: tables that already exist are reported and left alone, so a partial
+failure can be resolved by re-running. Dropping a table is not offered here --
+it discards data, and doing it by hand is the point.
+
+The Glue database itself is Terraform's
+(`terraform/bluesky_ingestion_jetstream/main.tf`); the
+tables are not, because Iceberg rewrites a table's schema, partition spec, and
+snapshot pointer on every commit, which an `aws_glue_catalog_table` resource
+would read as drift and revert on the next apply.
+"""
+
+from pyiceberg.catalog.glue import GlueCatalog
+from pyiceberg.exceptions import NoSuchNamespaceError, TableAlreadyExistsError
+from pyiceberg.table import Table
+from pyiceberg.transforms import DayTransform
+
+from bluesky_ingestion_jetstream.aws.catalog import build_catalog
+from bluesky_ingestion_jetstream.aws.constants import (
+ GLUE_DATABASE,
+ PARTITION_FIELD_NAME,
+ PARTITION_SOURCE_COLUMN,
+ TABLE_LOCATIONS,
+ TABLE_PROPERTIES,
+)
+from bluesky_ingestion_jetstream.schemas.arrow_schemas import RECORD_TYPE_TO_SCHEMA
+
+
+def require_namespace(catalog: GlueCatalog) -> None:
+ """Fail if Terraform has not created the Glue database yet."""
+
+ try:
+ catalog.list_tables(GLUE_DATABASE)
+ except NoSuchNamespaceError as error:
+ raise RuntimeError(
+ f"Glue database {GLUE_DATABASE!r} does not exist. It is managed by "
+ "Terraform -- run `terraform apply` in "
+ "terraform/bluesky_ingestion_jetstream first."
+ ) from error
+
+
+def create_table(catalog: GlueCatalog, record_type: str) -> Table:
+ """Create one partitioned table from its Arrow schema.
+
+ The Arrow schema is passed through untouched so PyIceberg assigns the field
+ IDs itself. Hand-written IDs are renumbered on create, and because Iceberg
+ resolves columns by ID rather than name, any mismatch reads back as NULL for
+ every affected column rather than failing. Not writing IDs at all removes
+ that failure mode instead of guarding against it.
+ """
+
+ table = catalog.create_table(
+ identifier=(GLUE_DATABASE, record_type),
+ schema=RECORD_TYPE_TO_SCHEMA[record_type],
+ location=TABLE_LOCATIONS[record_type],
+ properties=TABLE_PROPERTIES,
+ )
+
+ # Partition after creation rather than passing a `PartitionSpec`, because
+ # `add_field` takes the column name and resolves the source ID itself.
+ table.update_spec().add_field(
+ PARTITION_SOURCE_COLUMN, DayTransform(), PARTITION_FIELD_NAME
+ ).commit()
+
+ return table
+
+
+def bootstrap(catalog: GlueCatalog | None = None) -> dict[str, Table]:
+ """Create every missing table. Returns record type -> table for all four."""
+
+ catalog = catalog or build_catalog()
+ require_namespace(catalog)
+
+ tables: dict[str, Table] = {}
+ for record_type in RECORD_TYPE_TO_SCHEMA:
+ try:
+ tables[record_type] = create_table(catalog, record_type)
+ print(f"created {GLUE_DATABASE}.{record_type} -> {TABLE_LOCATIONS[record_type]}")
+ except TableAlreadyExistsError:
+ tables[record_type] = catalog.load_table((GLUE_DATABASE, record_type))
+ print(f"exists {GLUE_DATABASE}.{record_type}")
+
+ return tables
+
+
+def main() -> None:
+ """CLI entry point."""
+
+ for record_type, table in bootstrap().items():
+ print(f"{record_type}: {table.spec()}")
+
+
+if __name__ == "__main__":
+ main()

bluesky_ingestion_jetstream/aws/catalog.py

@@ -0,0 +1,98 @@
+"""Connect to the Glue catalog and load the four Iceberg tables.
+
+Runs on every process start, unlike `bootstrap.py`. Nothing here creates or
+alters a table: an ingester that can issue DDL is one typo in a database name
+away from quietly writing a full day of data into a second, empty set of tables
+that looks fine until someone queries the real ones. Missing tables raise.
+"""
+
+import boto3
+from botocore.config import Config
+from pyiceberg.catalog.glue import GlueCatalog
+from pyiceberg.exceptions import NoSuchTableError
+from pyiceberg.table import Table
+
+from bluesky_ingestion_jetstream.aws.constants import (
+ AWS_REGION,
+ GLUE_CONNECT_TIMEOUT_SECONDS,
+ GLUE_DATABASE,
+ GLUE_MAX_ATTEMPTS,
+ GLUE_READ_TIMEOUT_SECONDS,
+ S3_BUCKET,
+ S3_CONNECT_TIMEOUT_SECONDS,
+ S3_PREFIX,
+ S3_REQUEST_TIMEOUT_SECONDS,
+)
+from bluesky_ingestion_jetstream.constants import RECORD_TYPES
+
+
+class MissingTablesError(RuntimeError):
+ """Raised when the catalog is missing tables `bootstrap.py` should have created."""
+
+
+def build_glue_client():
+ """A Glue client with a bounded worst case.
+
+ Built here rather than left to PyIceberg because its default is `standard`
+ retry mode with ten attempts over a 60s read timeout, which puts an
+ open-ended retry loop underneath every commit. The commit is retried at a
+ higher level where the failure can be dead-lettered, so this layer only needs
+ to cover a single dropped packet, not an outage.
+ """
+
+ return boto3.client(
+ "glue",
+ region_name=AWS_REGION,
+ config=Config(
+ retries={"max_attempts": GLUE_MAX_ATTEMPTS, "mode": "standard"},
+ connect_timeout=GLUE_CONNECT_TIMEOUT_SECONDS,
+ read_timeout=GLUE_READ_TIMEOUT_SECONDS,
+ ),
+ )
+
+
+def build_catalog() -> GlueCatalog:
+ """Construct the Glue-backed catalog.
+
+ Left on PyIceberg's default PyArrowFileIO. The Iceberg experiment pinned
+ FsspecFileIO so its S3 meter could see every request through aiobotocore;
+ production has nothing to meter and PyArrow's client is faster.
+ """
+
+ return GlueCatalog(
+ name="bluesky",
+ client=build_glue_client(),
+ **{
+ "warehouse": f"s3://{S3_BUCKET}/{S3_PREFIX}",
+ "glue.region": AWS_REGION,
+ "s3.region": AWS_REGION,
+ "s3.connect-timeout": S3_CONNECT_TIMEOUT_SECONDS,
+ "s3.request-timeout": S3_REQUEST_TIMEOUT_SECONDS,
+ },
+ )
+
+
+def load_tables(catalog: GlueCatalog) -> dict[str, Table]:
+ """Load every record type's table, or raise naming all the ones missing.
+
+ Called once at startup rather than per flush, because each load is a Glue
+ `GetTable` call. Every missing table is collected before raising, so a fresh
+ environment reports all four in one go instead of one per re-run.
+ """
+
+ tables: dict[str, Table] = {}
+ missing: list[str] = []
+
+ for record_type in RECORD_TYPES:
+ try:
+ tables[record_type] = catalog.load_table((GLUE_DATABASE, record_type))
+ except NoSuchTableError:
+ missing.append(record_type)
+
+ if missing:
+ raise MissingTablesError(
+ f"Glue database {GLUE_DATABASE!r} is missing table(s): {', '.join(missing)}. "
+ "Run `python -m bluesky_ingestion_jetstream.aws.bootstrap` to create them."
+ )
+
+ return tables

bluesky_ingestion_jetstream/aws/constants.py

@@ -0,0 +1,75 @@
+"""AWS identifiers and Iceberg table configuration for this pipeline."""
+
+from bluesky_ingestion_jetstream.constants import RECORD_TYPES
+
+AWS_REGION = "us-east-2"
+S3_BUCKET = "lab-data-integrations-interface"
+S3_PREFIX = "bluesky/raw"
+
+# Created by Terraform (`terraform/bluesky_ingestion_jetstream/main.tf`).
+GLUE_DATABASE = "bluesky_raw"
+
+# One table per record type. Glue names cannot contain `/`, so the location is
+# passed explicitly at creation.
+TABLE_LOCATIONS = {
+ record_type: f"s3://{S3_BUCKET}/{S3_PREFIX}/{record_type}" for record_type in RECORD_TYPES
+}
+
+# Applied at table creation only; edits here do not reach existing tables.
+TABLE_PROPERTIES = {
+ "format-version": "2",
+ "write.parquet.compression-codec": "zstd",
+ "write.target-file-size-bytes": str(256 * 1024 * 1024),
+ # Iceberg's default salts a hash into the data path; unnecessary at this
+ # volume, and it costs a browsable `created_at_day=.../` layout.
+ "write.object-storage.enabled": "false",
+ "write.metadata.delete-after-commit.enabled": "true",
+ "write.metadata.previous-versions-max": "100",
+ # Inert while ingestion is append-only. Set now for the duplicates backfill
+ # will introduce; PyIceberg cannot write delete files, so that merge needs
+ # Athena or Spark.
+ "write.delete.mode": "merge-on-read",
+ "write.update.mode": "merge-on-read",
+ "write.merge.mode": "merge-on-read",
+}
+
+# Iceberg's default name for a `day()` transform, as it appears on disk.
+PARTITION_SOURCE_COLUMN = "created_at"
+PARTITION_FIELD_NAME = "created_at_day"
+
+# ---------------------------------------------------------------------------
+# Client bounds
+#
+# Overrides PyIceberg's Glue defaults
+# ---------------------------------------------------------------------------
+
+GLUE_MAX_ATTEMPTS = 2
+GLUE_CONNECT_TIMEOUT_SECONDS = 3.0
+GLUE_READ_TIMEOUT_SECONDS = 10.0
+
+# Passed to the PyArrow S3 filesystem (pyiceberg/io/pyarrow.py:445-448).
+S3_CONNECT_TIMEOUT_SECONDS = 3.0
+S3_REQUEST_TIMEOUT_SECONDS = 15.0
+
+# ---------------------------------------------------------------------------
+# Commit retry
+#
+# Three attempts, two sleeps
+# ---------------------------------------------------------------------------
+
+COMMIT_MAX_ATTEMPTS = 3
+COMMIT_INITIAL_DELAY_SECONDS = 1.0
+COMMIT_MAX_DELAY_SECONDS = 8.0
+
+# Stamped on the snapshot, so a retry can tell a failed commit from a lost reply.
+SNAPSHOT_FLUSH_ID_TAG = "flush_id"
+
+# ---------------------------------------------------------------------------
+# Dead letter
+#
+# Outside `S3_PREFIX`: orphan cleanup deletes unreferenced files under the
+# warehouse root, and these are unreferenced by definition.
+# ---------------------------------------------------------------------------
+
+DEAD_LETTER_PREFIX = "dead_letter/bluesky/raw"
+DEAD_LETTER_ROOT = f"{S3_BUCKET}/{DEAD_LETTER_PREFIX}"