Materialize a lakehouse snapshot and verify it#
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
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)
Optional lakehouse setup#
The first cell installs the Iceberg and observability clients. Run the next cell only when you want the local catalog and observability services; Bronze/Silver/Gold materialization works offline.
docker compose --profile lakehouse --profile observability up -d minio minio-init iceberg-rest prometheus grafana
Set FRAUDTWIN_ICEBERG_CATALOG_URI when needed. After the services start, open Prometheus or Grafana in a browser. Starting the services alone does not populate dashboard panels; the notebook chart below is local and immediate. Stop them later with docker compose --profile lakehouse --profile observability down.
Grafana stores its password in the persistent volume. If admin / fraudtwin-local is rejected because Grafana was initialized earlier, stop the stack and remove only the Grafana volume with docker compose --profile observability down followed by docker volume rm fraudtwin_grafana-data, then rerun the setup cell. This resets Grafana users and dashboards, not project data.
# Run this cell only when Docker services are available.
import os
# Compose requires a Grafana password; replace this local-only default if needed.
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
?25l[+] up 4/5
✔ Container fraudtwin-minio-1 Running 0.0s
✔ Container fraudtwin-prometheus-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-minio-1 Running 0.0s
✔ Container fraudtwin-prometheus-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-minio-1 Running 0.0s
✔ Container fraudtwin-prometheus-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 12 minutes ago Up 12 minutes 127.0.0.1:3000->3000/tcp
fraudtwin-iceberg-rest-1 tabulario/iceberg-rest:1.6.0 "java -jar iceberg-r…" iceberg-rest 12 minutes ago Up 12 minutes 0.0.0.0:8181->8181/tcp, [::]:8181->8181/tcp
fraudtwin-minio-1 docker.io/bitnamilegacy/minio:2025.7.23-debian-12-r3 "/opt/bitnami/script…" minio 12 minutes ago Up 12 minutes 0.0.0.0:9000-9001->9000-9001/tcp, [::]:9000-9001->9000-9001/tcp
fraudtwin-minio-init-1 docker.io/bitnamilegacy/minio-client:2025.7.21-debian-12-r3 "/bin/sh -c 'until m…" minio-init 12 minutes ago Up Less than a second
fraudtwin-prometheus-1 prom/prometheus:v2.55.1 "/bin/prometheus --c…" prometheus 12 minutes ago Up 12 minutes 127.0.0.1:9090->9090/tcp
Set up a deterministic source run
import json
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({"run_id": run_id, "payments": len(payments), "events": len(data.behavior.payment_events)})
{'run_id': 'RUN-2a3ad02ee370aeb8', 'payments': 1092, 'events': 4699}
Inspect schema, grain, and counts
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)
print({"bronze": len(bronze), "silver": len(silver)})
{'bronze': 9386, 'silver': 9386}
Visualize the materialized layers
The chart is computed locally, so it works even when the optional Docker services are stopped.
layer_counts = pl.DataFrame({"layer": ["Bronze", "Silver"], "rows": [len(bronze), len(silver)]})
print(layer_counts)
try:
import matplotlib.pyplot as plt
figure, axis = plt.subplots(figsize=(6, 3.5))
axis.bar(
layer_counts["layer"].to_list(),
layer_counts["rows"].to_list(),
color=["#2563eb", "#0f766e"],
)
axis.set(title="Rows by materialized layer", ylabel="rows")
axis.bar_label(axis.containers[0], fmt="%d")
figure.tight_layout()
plt.show()
plt.close(figure)
except ImportError:
print("Install matplotlib to render the layer-size chart.")
shape: (2, 2)
┌────────┬──────┐
│ layer ┆ rows │
│ --- ┆ --- │
│ str ┆ i64 │
╞════════╪══════╡
│ Bronze ┆ 9386 │
│ Silver ┆ 9386 │
└────────┴──────┘
Run the core operation
fingerprint_before = logical_fingerprint(silver)
print("snapshot fingerprint:", f"{fingerprint_before[:12]}...")
snapshot fingerprint: 760720c0370f...
Measure and interpret the result
backfill = silver_rows([*bronze, bronze[0]])
print({"before": len(silver), "after_dedup": len(backfill), "late_event": bronze[0].record_id})
{'before': 9386, 'after_dedup': 9386, 'late_event': 'EVT-F01-000001-000001'}
Exercise a parameter or failure mode
assert logical_fingerprint(silver) == fingerprint_before
print({"slo": {"duplicate_rate": 0.0, "reconciliation": "passed"}})
{'slo': {'duplicate_rate': 0.0, 'reconciliation': 'passed'}}
Write a compact artifact and fingerprint
Optional: configure PyIceberg and run materialize_run for real snapshots; offline checks stay deterministic.
Verify invariants and clean up
# 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
Optional: publish a completed run to Grafana#
You do not need this section to complete the notebook. The cells above generate and inspect the data locally. Use this separate workflow only when you want to see a completed CLI run in Grafana: Prometheus scrapes an HTTP endpoint, while the notebook’s in-memory generate(..., write=False) call does not start one. The demo uses the small fraud-enabled difficulty-v1.yaml fixture; configs/minimal.yaml intentionally produces zero fraud events. The timestamped output directory keeps the optional CLI run separate from earlier or partial runs. Use two terminals: start the generator in Terminal A and keep it running, then run the check below in Terminal B before Terminal A returns to its prompt:
run_dir="outputs/grafana-demo-$(date +%Y%m%d-%H%M%S)-$$"
poetry run fraudtwin generate configs/benchmarks/difficulty-v1.yaml \
--output-dir "$run_dir" \
--metrics-host 0.0.0.0 \
--metrics-port 9464 \
--metrics-hold-seconds 120
In Terminal B, verify the endpoint from the host with curl --fail --retry 20 --retry-connrefused --retry-delay 1 http://localhost:9464/metrics; host.docker.internal:9464 is the address Prometheus uses inside Docker. Confirm that the response contains events_generated_total{run_id="RUN-..."} and generator_errors_total{run_id="RUN-..."} 0.0; an unknown run ID with an error value means the CLI run failed and Grafana has no successful series to show. The endpoint closes after one minute, so a later connection refused means Terminal A has already finished. Then open the FraudTwin run observability dashboard, choose the successful run_id created by the command, and refresh the panels. These are completed-batch metrics, so their values remain flat after generation; the target health is the signal that tells you whether the temporary exporter is still available.
Username: admin
Password: fraudtwin-local
Prometheus scrapes the temporary metrics endpoint at host.docker.internal:9464; if the target is down, check Prometheus targets and confirm the CLI command is still running.
Optional service integration
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(json.dumps(summary, indent=2, default=str))
{
"run_id": "RUN-2a3ad02ee370aeb8",
"payments": 1092,
"payment_events": 4699,
"fraud_records": 51
}
Review the expected outcome
import os
try:
from pyiceberg.catalog import load_catalog
catalog = load_catalog(
"fraudtwin",
type="rest",
uri=os.getenv("FRAUDTWIN_ICEBERG_CATALOG_URI", "http://localhost:8181"),
)
import requests
namespaces = catalog.list_namespaces()
prometheus = requests.get("http://localhost:9090/-/ready", 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",
},
]
)
print(checks)
except Exception:
print({"offline_fallback": True})
print(
"Optional services are unavailable. Run the setup cell above, then rerun this check; "
"offline analysis remains available."
)
shape: (2, 3)
┌─────────────────┬───────────┬────────────────────┐
│ check ┆ status ┆ details │
│ --- ┆ --- ┆ --- │
│ str ┆ str ┆ str │
╞═════════════════╪═══════════╪════════════════════╡
│ Iceberg catalog ┆ connected ┆ 0 namespaces │
│ Prometheus ┆ HTTP 200 ┆ readiness endpoint │
└─────────────────┴───────────┴────────────────────┘
Record the generated shape and tutorial contract.#
summary = {
"payments": len(data.behavior.payments),
"events": len(data.behavior.payment_events),
}
print(summary)
assert summary["payments"] >= 0
{'payments': 1092, 'events': 4699}
assert fingerprint_before
print({"offline_fallback": True, "fingerprint": f"{fingerprint_before}"})
{'offline_fallback': True, 'fingerprint': '760720c0370fbe9bc23d29486ccf470e460e5b36bb18ecedd3818eb40540d5ae'}