Verify Iceberg projections and catalog health

Goal. Work through a bounded, reproducible example and inspect the evidence before connecting an external service.

Prerequisites. Base FraudTwin install. Optional extras and Docker commands are clearly marked.

Produces. Tables, fingerprints, manifests, and verification output.

Source size. The default cells generate approximately 1,000 logical payments; increase duration and population together for a 10,000-payment run.

Offline path. All marked offline cells run without Docker or network services. Service cells are optional and explicitly marked in notebook metadata.

Cleanup. Outputs are written under a temporary directory; remove any local run directory if you changed the output location.

!pip install pyiceberg[s3fs]
!pip install prometheus-client requests

# Launch the services for this tutorial. Re-running this cell is safe.
import os

os.environ.setdefault("GRAFANA_ADMIN_PASSWORD", "fraudtwin-local")
compose_profiles = "--profile lakehouse --profile observability"
!docker compose {compose_profiles} up -d minio minio-init iceberg-rest prometheus grafana
!docker compose --profile lakehouse --profile observability ps

Requirement already satisfied: pyiceberg[s3fs] in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (0.12.0)
Requirement already satisfied: mmh3>=4.0.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (5.3.0)
Requirement already satisfied: requests>=2.20.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (2.34.2)
Requirement already satisfied: click>=7.1.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (8.5.0)
Requirement already satisfied: rich>=10.11.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (15.0.0)
Requirement already satisfied: strictyaml>=1.7.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (1.7.3)
Requirement already satisfied: pydantic!=2.12.0,!=2.12.1,!=2.4.0,!=2.4.1,<3.0,>=2.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (2.13.5)
Requirement already satisfied: fsspec>=2023.1.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (2026.9.0)
Requirement already satisfied: pyparsing>=3.1.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (3.3.3)
Requirement already satisfied: tenacity>=8.2.3 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (9.1.4)
Requirement already satisfied: pyroaring>=1.0.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (1.1.0)
Requirement already satisfied: cachetools>=5.5 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (7.2.0)
Requirement already satisfied: zstandard>=0.13.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (0.25.0)
Requirement already satisfied: s3fs>=2023.1.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pyiceberg[s3fs]) (2026.9.0)
Requirement already satisfied: annotated-types>=0.6.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pydantic!=2.12.0,!=2.12.1,!=2.4.0,!=2.4.1,<3.0,>=2.0->pyiceberg[s3fs]) (0.8.0)
Requirement already satisfied: pydantic-core==2.46.5 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pydantic!=2.12.0,!=2.12.1,!=2.4.0,!=2.4.1,<3.0,>=2.0->pyiceberg[s3fs]) (2.46.5)
Requirement already satisfied: typing-extensions>=4.14.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pydantic!=2.12.0,!=2.12.1,!=2.4.0,!=2.4.1,<3.0,>=2.0->pyiceberg[s3fs]) (4.16.0)
Requirement already satisfied: typing-inspection>=0.4.2 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from pydantic!=2.12.0,!=2.12.1,!=2.4.0,!=2.4.1,<3.0,>=2.0->pyiceberg[s3fs]) (0.4.4)
Requirement already satisfied: charset_normalizer<4,>=2 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from requests>=2.20.0->pyiceberg[s3fs]) (3.5.1)
Requirement already satisfied: idna<4,>=2.5 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from requests>=2.20.0->pyiceberg[s3fs]) (3.19)
Requirement already satisfied: urllib3<3,>=1.26 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from requests>=2.20.0->pyiceberg[s3fs]) (2.7.0)
Requirement already satisfied: certifi>=2023.5.7 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from requests>=2.20.0->pyiceberg[s3fs]) (2026.7.22)
Requirement already satisfied: markdown-it-py>=2.2.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from rich>=10.11.0->pyiceberg[s3fs]) (3.0.0)
Requirement already satisfied: pygments<3.0.0,>=2.13.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from rich>=10.11.0->pyiceberg[s3fs]) (2.21.0)
Requirement already satisfied: mdurl~=0.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from markdown-it-py>=2.2.0->rich>=10.11.0->pyiceberg[s3fs]) (0.1.2)
Requirement already satisfied: aiobotocore<4.0.0,>=2.19.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from s3fs>=2023.1.0->pyiceberg[s3fs]) (3.9.1)
Requirement already satisfied: aiohttp!=4.0.0a0,!=4.0.0a1,>=3.9.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from s3fs>=2023.1.0->pyiceberg[s3fs]) (3.14.3)
Requirement already satisfied: aioitertools<1.0.0,>=0.5.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiobotocore<4.0.0,>=2.19.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (0.13.0)
Requirement already satisfied: botocore<1.43.76,>=1.43.66 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiobotocore<4.0.0,>=2.19.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (1.43.75)
Requirement already satisfied: jmespath<2.0.0,>=0.7.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiobotocore<4.0.0,>=2.19.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (1.1.0)
Requirement already satisfied: multidict<7.0.0,>=6.0.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiobotocore<4.0.0,>=2.19.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (6.9.1)
Requirement already satisfied: python-dateutil<3.0.0,>=2.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiobotocore<4.0.0,>=2.19.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (2.9.0.post0)
Requirement already satisfied: wrapt<3.0.0,>=1.10.10 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiobotocore<4.0.0,>=2.19.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (2.4.1)
Requirement already satisfied: aiohappyeyeballs>=2.5.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiohttp!=4.0.0a0,!=4.0.0a1,>=3.9.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (2.7.1)
Requirement already satisfied: aiosignal>=1.4.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiohttp!=4.0.0a0,!=4.0.0a1,>=3.9.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (1.4.0)
Requirement already satisfied: attrs>=17.3.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiohttp!=4.0.0a0,!=4.0.0a1,>=3.9.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (26.1.0)
Requirement already satisfied: frozenlist>=1.1.1 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiohttp!=4.0.0a0,!=4.0.0a1,>=3.9.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (1.8.0)
Requirement already satisfied: propcache>=0.2.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiohttp!=4.0.0a0,!=4.0.0a1,>=3.9.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (0.5.4)
Requirement already satisfied: yarl<2.0,>=1.17.0 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from aiohttp!=4.0.0a0,!=4.0.0a1,>=3.9.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (1.25.1)
Requirement already satisfied: six>=1.5 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from python-dateutil<3.0.0,>=2.1->aiobotocore<4.0.0,>=2.19.0->s3fs>=2023.1.0->pyiceberg[s3fs]) (1.17.0)
Requirement already satisfied: prometheus-client in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (0.26.0)
Requirement already satisfied: requests in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (2.34.2)
Requirement already satisfied: charset_normalizer<4,>=2 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from requests) (3.5.1)
Requirement already satisfied: idna<4,>=2.5 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from requests) (3.19)
Requirement already satisfied: urllib3<3,>=1.26 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from requests) (2.7.0)
Requirement already satisfied: certifi>=2023.5.7 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from requests) (2026.7.22)
?25l[+] up 4/5
 ✔ Container fraudtwin-prometheus-1   Running                              0.0s
 ✔ Container fraudtwin-minio-1        Running                              0.0s
 ✔ Container fraudtwin-grafana-1      Running                              0.0s
 ✔ Container fraudtwin-iceberg-rest-1 Running                              0.0s
 ⠋ Container fraudtwin-minio-init-1   Starting                             0.0s
?25h?25l[+] up 4/5
 ✔ Container fraudtwin-prometheus-1   Running                              0.0s
 ✔ Container fraudtwin-minio-1        Running                              0.0s
 ✔ Container fraudtwin-grafana-1      Running                              0.0s
 ✔ Container fraudtwin-iceberg-rest-1 Running                              0.0s
 ⠙ Container fraudtwin-minio-init-1   Starting                             0.1s
?25h?25l[+] up 5/5
 ✔ Container fraudtwin-prometheus-1   Running                              0.0s
 ✔ Container fraudtwin-minio-1        Running                              0.0s
 ✔ Container fraudtwin-grafana-1      Running                              0.0s
 ✔ Container fraudtwin-iceberg-rest-1 Running                              0.0s
 ✔ Container fraudtwin-minio-init-1   Started                              0.2s
?25hNAME                           IMAGE                                                  COMMAND                  SERVICE           CREATED         STATUS         PORTS
fraudtwin-grafana-1           grafana/grafana:11.2.0                                 "/run.sh"                grafana           5 minutes ago   Up 5 minutes   127.0.0.1:3000->3000/tcp
fraudtwin-iceberg-rest-1      tabulario/iceberg-rest:1.6.0                           "java -jar iceberg-r…"   iceberg-rest      5 minutes ago   Up 5 minutes   0.0.0.0:8181->8181/tcp, [::]:8181->8181/tcp
fraudtwin-kafka-1             confluentinc/cp-kafka:7.7.1                            "/etc/confluent/dock…"   kafka             5 minutes ago   Up 5 minutes   0.0.0.0:9092->9092/tcp, [::]:9092->9092/tcp
fraudtwin-minio-1             docker.io/bitnamilegacy/minio:2025.7.23-debian-12-r3   "/opt/bitnami/script…"   minio             5 minutes ago   Up 5 minutes   0.0.0.0:9000-9001->9000-9001/tcp, [::]:9000-9001->9000-9001/tcp
fraudtwin-prometheus-1        prom/prometheus:v2.55.1                                "/bin/prometheus --c…"   prometheus        5 minutes ago   Up 5 minutes   127.0.0.1:9090->9090/tcp
fraudtwin-schema-registry-1   confluentinc/cp-schema-registry:7.7.1                  "/etc/confluent/dock…"   schema-registry   5 minutes ago   Up 5 minutes   0.0.0.0:8081->8081/tcp, [::]:8081->8081/tcp

Optional Iceberg and observability setup

The offline cells below build deterministic Bronze/Silver projections. Run the optional integration setup cell below when you want this independent tutorial to launch the catalog, object store, Prometheus, and Grafana for service-backed checks or dashboard publishing.

docker compose --profile lakehouse --profile observability up -d minio minio-init iceberg-rest prometheus grafana

After the services start, the integration cell checks the Iceberg catalog, Prometheus, and Grafana health endpoints. The completed-run Grafana workflow is maintained in the lakehouse observability tutorial; use that notebook when you want dashboard panels and run-level metrics. Stop the services later with docker compose --profile lakehouse --profile observability down.

Grafana uses username admin and password fraudtwin-local for a fresh local volume. If that password was changed earlier, use the current one. Set FRAUDTWIN_ICEBERG_CATALOG_URI when using another REST catalog.

Set up a deterministic source run

from pathlib import Path

import polars as pl

from fraudtwin.config import load_config
from fraudtwin.generation import generate

root = next(
    (p for p in (Path.cwd(), *Path.cwd().parents) if (p / "configs" / "minimal.yaml").exists()),
    Path.cwd(),
)
base = load_config(root / "configs" / "minimal.yaml")
# Scale the population so the bounded example produces about 1,000 payments.
population = base.population.model_copy(
    update={
        "customers": 200,
        "accounts": 300,
        "cards": 240,
        "devices": 240,
        "pix_keys": 160,
        "merchants": 60,
    }
)
simulation = base.simulation.model_copy(update={"duration_days": 10})
fraud = base.fraud.model_copy(update={"enabled": True, "target_rate": 0.05})
config = base.model_copy(
    update={"population": population, "simulation": simulation, "fraud": fraud}
)
data = generate(config, write=False)
run_id = data.run_id
payments = pl.DataFrame([item.model_dump(mode="json") for item in data.behavior.payments])
print("Generated source run")
print(f"  run id: {run_id}")
print(f"  payments: {len(payments):,}")
print(f"  events: {len(data.behavior.payment_events):,}")

Generated source run
  run id: RUN-2a3ad02ee370aeb8
  payments: 1,092
  events: 4,699

Build the local Silver projection

from fraudtwin.lakehouse import build_bronze_records, logical_fingerprint, silver_rows

bronze = build_bronze_records(data.entities, data.behavior, data.manifest)
silver = silver_rows(bronze)
snapshot_a = logical_fingerprint(silver)
print("Silver projection")
print(f"  rows: {len(silver):,}")
print(f"  logical fingerprint: {snapshot_a[:12]}...")

Silver projection
  rows: 9,386
  logical fingerprint: 760720c0370f...

Compare two logical versions

snapshot_b = logical_fingerprint(silver[:-1])
display(
    pl.DataFrame(
        [
            {"version": "A", "rows": len(silver), "fingerprint": f"{snapshot_a[:12]}..."},
            {"version": "B", "rows": len(silver[:-1]), "fingerprint": f"{snapshot_b[:12]}..."},
        ]
    )
)
print(f"content changed: {snapshot_a != snapshot_b}")

shape: (2, 3)

version	rows	fingerprint
str	i64	str
"A"	9386	"760720c0370f..."
"B"	9385	"97e4783c49a7..."

content changed: True

Test a late-event backfill

backfill = silver_rows([*bronze, bronze[0]])
print("Late-event backfill")
print(f"  rows after deduplication: {len(backfill):,}")
print(f"  duplicate ignored: {len(backfill) == len(silver)}")

Late-event backfill
  rows after deduplication: 9,386
  duplicate ignored: True

Review the offline observability contract

slo = {"lag_seconds": 0, "duplicate_rate": 0.0, "invalid_records": 0, "reconciliation": "passed"}
print("Offline SLO contract (illustrative)")
display(pl.DataFrame([slo]))

Offline SLO contract (illustrative)

shape: (1, 4)

lag_seconds	duplicate_rate	invalid_records	reconciliation
i64	f64	i64	str
0	0.0	0	"passed"

Write a compact artifact and fingerprint

assert slo["reconciliation"] == "passed"
print(
    "Logical fingerprints change when a projection changes; real Iceberg snapshot IDs require "
    "materialize_run."
)

Logical fingerprints change when a projection changes; real Iceberg snapshot IDs require materialize_run.

Verify invariants and clean up

import os

try:
    import requests
    from pyiceberg.catalog import load_catalog

    catalog = load_catalog(
        "fraudtwin",
        type="rest",
        uri=os.getenv("FRAUDTWIN_ICEBERG_CATALOG_URI", "http://localhost:8181"),
    )
    namespaces = catalog.list_namespaces()
    prometheus = requests.get("http://localhost:9090/-/ready", timeout=3)
    metric = requests.get("http://localhost:9090/api/v1/query", params={"query": "up"}, timeout=3)
    grafana = requests.get("http://localhost:3000/api/health", timeout=3)
    checks = pl.DataFrame(
        [
            {
                "check": "Iceberg catalog",
                "status": "connected",
                "details": f"{len(namespaces)} namespaces",
            },
            {
                "check": "Prometheus",
                "status": f"HTTP {prometheus.status_code}",
                "details": "readiness endpoint",
            },
            {
                "check": "Prometheus query",
                "status": f"HTTP {metric.status_code}",
                "details": "up query",
            },
            {
                "check": "Grafana",
                "status": f"HTTP {grafana.status_code}",
                "details": "health endpoint",
            },
        ]
    )
    display(checks)
    print("Open Grafana: http://localhost:3000")
    print("Open Prometheus: http://localhost:9090")
except Exception as exc:
    print("Optional services are unavailable; offline snapshot analysis remains valid.")
    print("  offline_fallback: yes")
    print(f"  reason: {type(exc).__name__}")

shape: (4, 3)

check	status	details
str	str	str
"Iceberg catalog"	"connected"	"0 namespaces"
"Prometheus"	"HTTP 200"	"readiness endpoint"
"Prometheus query"	"HTTP 200"	"up query"
"Grafana"	"HTTP 200"	"health endpoint"

Open Grafana: http://localhost:3000
Open Prometheus: http://localhost:9090

Related: Grafana observability

This notebook checks lakehouse inputs and logical projections. For the completed CLI run, Prometheus exporter, and Grafana dashboard workflow, continue with the lakehouse observability tutorial. Keeping that workflow in one tutorial avoids producing two different dashboard procedures.

# A compact inspection is more useful than printing an entire run.
sample_columns = [
    c for c in ("payment_id", "amount", "initiated_at", "payer_account_id") if c in payments.columns
]
sample_rows = payments.select(sample_columns).head(8).to_dicts()
print(f"Sample payments ({len(sample_rows)} of {payments.height} rows):")
for row in sample_rows:
    print(
        f"  - {row.get('payment_id')}: amount={row.get('amount')}, "
        f"initiated_at={row.get('initiated_at')}, payer={row.get('payer_account_id')}"
    )
nulls = {name: count for name, count in payments.null_count().to_dicts()[0].items() if count}
print("\nData quality summary:")
print(f"  rows: {payments.height}")
print(f"  columns: {payments.width}")
if not nulls:
    print("  nulls: none")
else:
    print("  columns with nulls:")
    for name, count in sorted(nulls.items()):
        print(f"    - {name}: {count}")

Sample payments (8 of 1092 rows):
  - PAY-00000001: amount=70.7, initiated_at=2026-01-03T16:25:00Z, payer=ACC-000123
  - PAY-00000002: amount=25.52, initiated_at=2026-01-05T11:37:00Z, payer=ACC-000174
  - PAY-00000003: amount=18.37, initiated_at=2026-01-05T11:21:00Z, payer=ACC-000174
  - PAY-00000004: amount=5.54, initiated_at=2026-01-02T22:30:00Z, payer=ACC-000003
  - PAY-00000005: amount=13.71, initiated_at=2026-01-02T09:41:00Z, payer=ACC-000029
  - PAY-00000006: amount=70.06, initiated_at=2026-01-04T10:40:00Z, payer=ACC-000179
  - PAY-00000007: amount=42.73, initiated_at=2026-01-02T18:04:00Z, payer=ACC-000247
  - PAY-00000008: amount=36.42, initiated_at=2026-01-05T09:27:00Z, payer=ACC-000255

Data quality summary:
  rows: 1092
  columns: 15
  columns with nulls:
    - card_id: 565
    - merchant_id: 565
    - payee_account_id: 86
    - payee_institution_id: 86
    - payee_pix_key_id: 857
    - payer_institution_id: 86
    - payer_pix_key_id: 857

Review the expected outcome

summary = {
    "run_id": run_id,
    "payments": len(data.behavior.payments),
    "payment_events": len(data.behavior.payment_events),
    "fraud_records": len(data.behavior.fraud_records),
}
assert summary["payments"] == len(payments)
assert summary["payments"] > 0
print("Generated dataset")
print(f"  run id: {summary['run_id']}")
print(f"  payments: {summary['payments']:,}")
print(f"  payment events: {summary['payment_events']:,}")
print(f"  fraud records: {summary['fraud_records']:,}")

Generated dataset
  run id: RUN-2a3ad02ee370aeb8
  payments: 1,092
  payment events: 4,699
  fraud records: 51

Next recommended step

Offline projections above remain valid when Iceberg and Prometheus are unavailable.

assert snapshot_a
print("Offline snapshot fallback")
print("  offline_fallback: yes")
print(f"  fingerprint: {snapshot_a}")

Offline snapshot fallback
  offline_fallback: yes
  fingerprint: 760720c0370fbe9bc23d29486ccf470e460e5b36bb18ecedd3818eb40540d5ae

assert snapshot_a != snapshot_b
assert len(backfill) == len(silver)
print("Offline snapshot invariants passed")