Beam / Dataflow Sandbox

A synthetic-only, safety-first demo of a real streaming architecture:

Local Python dashboard + scenario generator
  -> synthetic Pub/Sub events
  -> Apache Beam streaming pipeline
  -> Google Cloud DataflowRunner (later, gated)
  -> sanitized telemetry Pub/Sub topic
  -> sanitized aggregate metrics in BigQuery
  -> local dashboard showing actual pipeline telemetry and metric output

1. What this project is

• A real Apache Beam pipeline (parsing, validation, filtering, keyed aggregation, metrics, telemetry) that runs today under DirectRunner and is structured to run later, unmodified, under DataflowRunner.
• A small FastAPI dashboard + scenario publisher that drives that pipeline and displays its actual sanitized output.
• Four deterministic, fully synthetic scenarios that exercise validation, business filtering, aggregation, and reconciliation.
• A set of gated, safety-first scripts and documents for a future, explicitly-approved cloud deployment.
• Two packages: a reusable, generic beam_pipeline_toolkit/ framework with no business logic baked in, and a demo/ package that applies it to one concrete, Cyberbiz-relevant e-commerce shape. See §3a.

2. What this project is NOT

• This sandbox is synthetic-only.
• It is not connected to Cyberbiz -- not its repository, not its GCP project, not its users, not its data, not its credentials, not its reporting tables.
• It is not a production deployment. No cloud resource has been created or contacted by this project.
• It does not prove production parity. The x1-x6 metric mapping and DBT-compatible order filter are sandbox/demo semantics only.
• It is intended only to demonstrate actual Beam/Dataflow mechanics safely, end to end, with data nobody needs to worry about.

3. Architecture

+----------------------------------------------------------------------+
| Local Dashboard / Scenario Publisher                                 |
|                                                                        |
|  Browser UI -> choose synthetic scenario -> POST /api/runs            |
|                                                                        |
|  Python FastAPI backend                                               |
|    -> creates synthetic run_id                                        |
|    -> publishes safe synthetic events in cloud mode (gated, future)   |
|    -> writes local events in local mode (today)                       |
|    -> exposes sanitized run telemetry and metrics to the browser      |
+------------------------------+-----------------------------------------+
                               |
                               v
                 +----------------------------+
                 | Pub/Sub input topic        |
                 | sandbox-synthetic-events    |  (future, not created)
                 +--------------+-------------+
                                |
                                v
+----------------------------------------------------------------------+
| Apache Beam / Dataflow Streaming Pipeline (demo/pipeline/)             |
|                                                                        |
| Read synthetic events -> Parse/normalize -> Validate                  |
|   -> invalid aggregate telemetry branch                               |
|   -> split session/order/customer events                              |
|   -> DBT-compatible order filter                                      |
|   -> key by run_id + synthetic_shop_id + metric_date                  |
|   -> windowed daily/run aggregation                                   |
|   -> semantic x1-x6-style metric row                                  |
|   -> custom Beam counters (namespace: sandbox_beam_demo)               |
|   -> sanitized stage telemetry                                        |
+---------------+--------------------------------+----------------------+
                |                                 |
                v                                 v
 +---------------------------+     +-----------------------------------+
 | Pub/Sub telemetry topic   |     | BigQuery sandbox aggregate tables  |
 | sandbox-pipeline-telemetry|     | demo_metrics, demo_run_events      |
 | (future, not created)     |     | (future, not created)              |
 +-------------+-------------+     +-----------------+-------------------+
               |                                     |
               +------------------+------------------+
                                  v
                    +----------------------------+
                    | Dashboard backend          |
                    | actual sanitized status,   |
                    | metrics, and explanations  |
                    +----------------------------+

Two modes, same pipeline code:

Local mode (today):
  Synthetic scenario -> DirectRunner batch run -> local sanitized
  telemetry + metrics files -> dashboard visualizes the actual local
  run artifacts.

Cloud mode (future, gated, not enabled):
  Synthetic scenario publisher -> Pub/Sub -> real Dataflow streaming
  job -> real Beam transforms and counters -> sanitized telemetry
  Pub/Sub topic -> BigQuery aggregate output -> dashboard receives
  actual telemetry and polls aggregate results.

3a. Two packages: a reusable framework, and a concrete demo

This repository is deliberately split into two packages with different audiences:

Package	What it is	Knows about Cyberbiz?
beam_pipeline_toolkit/	A generic, standalone Apache Beam pipeline framework -- parse/validate/aggregate/sink wiring, a Dataflow control-plane API (submit/stop/read evidence), and a safety layer (leak guard + cloud-action gate) -- plus an optional e-commerce domain extension (beam_pipeline_toolkit/ecommerce/) built on top of it using the same public API any caller would use.	No. No company name, field name, table schema, or business rule is hardcoded anywhere in this package. It's proven generic by tests/toolkit/, which exercises the core with a deliberately unrelated "widget count per warehouse" pipeline as well as the e-commerce layer.
demo/	The concrete, presentation-ready sandbox described in the rest of this README: a synthetic e-commerce scenario generator, dashboard, and Beam pipeline shaped like a Cyberbiz-style order/session/customer reporting pipeline (sessions, orders, x1-x6 metrics, DBT-compatible order filtering).	Only in shape -- the pipeline stages and semantics mirror the kind of streaming reporting pipeline this domain needs, but every identifier, id range, and date is synthetic (see §2), and no Cyberbiz code, credentials, or data are touched.

Why split it this way: the parse -> validate -> aggregate -> sink shape, and the submit/stop/evidence mechanics, are not specific to any one business. Pushing that shape into beam_pipeline_toolkit/ means the same framework -- not just the same idea -- could back a differently-shaped pipeline (different fields, different validity rules, a different metric row) without touching the framework itself. demo/ is what that looks like applied to one concrete, Cyberbiz-relevant e-commerce shape: it is the reference consumer, not the framework. See beam_pipeline_toolkit/README.md for the framework's own quick-start, extension points, and the e-commerce domain layer's API.

4. Local installation

Requires Python 3.11+ (this project was built and verified against Python 3.12.6, the newest locally available interpreter compatible with Apache Beam's Python SDK; Beam 2.74.0 supports 3.10-3.14).

# From the project root (PowerShell)
python -m venv .venv
.venv\Scripts\pip install --upgrade pip
.venv\Scripts\pip install -r requirements.txt

# macOS/Linux
python3.11 -m venv .venv
.venv/bin/pip install --upgrade pip
.venv/bin/pip install -r requirements.txt

requirements.lock.txt records the exact frozen dependency set verified against this interpreter.

5. PyCharm interpreter setup

1. Open this folder as a PyCharm project (it already contains .idea/).
2. File > Settings > Project > Python Interpreter > Add Interpreter > Existing Environment.
3. Point it at .venv\Scripts\python.exe (Windows) or .venv/bin/python (macOS/Linux).
4. Mark the project root as a Sources Root if imports like from demo.shared import ... aren't resolving (Right-click project root > Mark Directory as > Sources Root).

6. Local DirectRunner demo

.venv\Scripts\python demo\scripts\run_local_demo.py --scenario healthy_shadow_run

Valid --scenario values: healthy_shadow_run, validation_rejection, order_filter_exclusions, reconciliation_mismatch.

This generates the scenario's deterministic synthetic events, runs the real Beam pipeline with DirectRunner, and writes sanitized artifacts to runtime/<run_id>/:

input_events.jsonl     # synthetic input fed to the pipeline
metrics.jsonl          # sanitized x1-x6 style aggregate metric row
telemetry.jsonl        # sanitized pipeline stage telemetry
invalid_summary.json   # sanitized invalid-record summary
run_manifest.json       # run_id, scenario, mode, runner, status, counts

No cloud API is called. No ADC, GCP project, Pub/Sub, BigQuery, or GCS is required.

7. Dashboard startup

.venv\Scripts\python demo\scripts\run_dashboard.py

.venv/bin/python demo/scripts/run_dashboard.py

Then open http://127.0.0.1:8000/ in a browser. The dashboard binds only to 127.0.0.1, never 0.0.0.0.

From the dashboard you can pick one of the four scenarios and click Run Local DirectRunner Demo -- this runs the real pipeline (same code path as demo/scripts/run_local_demo.py) and renders its actual sanitized output: run details, x1-x6 metrics, invalid/reconciliation summary, the sanitized event timeline, and an explanation rail built from that run's real telemetry (not a browser-timer simulation).

8. Scenario descriptions

Scenario	Shop	Date	What it demonstrates
healthy_shadow_run	90001	2026-07-01	Fully clean run: 2400 sessions, 43 eligible orders @ 280.50, 6 customers, FULL_MATCH.
validation_rejection	90002	2026-07-02	A duplicated synthetic session-day key is detected and isolated as DUPLICATE_SESSION_DAY; run completes with COMPLETED_WITH_VALIDATION_WARNING.
order_filter_exclusions	90003	2026-07-03	Synthetic orders excluded for cancelled status, unsupported return status, and a non-eligible user marker; orders_dbt_compatible / orders_excluded counters demonstrate the sandbox order filter.
reconciliation_mismatch	90004	2026-07-04	The pipeline processes its input correctly, but the dashboard compares it to a separately defined synthetic reporting contract that is deliberately off by one order -- clearly labeled SIMULATED MISMATCH, not a real production discrepancy.

Full expected values are in demo/config/scenario_definitions.yaml.

9. Safety model

• Synthetic data only. Synthetic shop ids are limited to 90001-90004; synthetic dates are fixed sandbox dates (2026-07-01..2026-07-04).
• No forbidden fields, ever. demo/shared/safety.py defines a recursive validator that rejects any object (scenario data, API response, telemetry event, metric row, or persisted artifact) containing a key resembling customer_id, email, name*, phone, address, payload, raw, token, secret, credential, service_account, traceback, stack, or log. (*name/scenario are allowed as the sandbox's own non-personal identifier fields.)
• No raw records ever leave validation/filtering. Invalid records and excluded orders are only ever surfaced as sanitized aggregate counts, never as raw rows, in local files, the API, or (future) Pub/Sub/BigQuery.
• Cloud is triple-gated. Any code path that would construct a Pub/Sub, BigQuery, or Dataflow client requires ALL of: --enable-cloud-mode, --execute-cloud-pipeline (or --execute-dataflow-job), and --confirm-synthetic-sandbox-only I_CONFIRM_SYNTHETIC_SANDBOX_ONLY (see demo/shared/safety.py::assert_cloud_gates_open). None of these gates are set in this project as delivered.
• Cloud config has no defaults. Project id, region, bucket, topics, subscriptions, dataset, and tables must all come from environment configuration -- never from a browser request, and never hardcoded.
• The browser never talks to Google Cloud directly. Only to this backend's sanitized JSON API.

10. Future GCP setup checklist

See demo/infra/README_CLOUD_SETUP.md for the full walkthrough. Summary:

1. Create/approve a separate sandbox GCP project (own billing/budget alert).
2. Enable Pub/Sub, BigQuery, Dataflow, and Cloud Storage APIs manually.
3. Create the sandbox bucket (see demo/infra/dataflow_submission_plan.md).
4. Create Pub/Sub resources (see demo/infra/pubsub_resource_plan.md).
5. Apply the BigQuery schema (see below).
6. Set up least-privilege IAM (see demo/infra/iam_role_plan.md).
7. Populate a local .env from .env.example.
8. Run python demo/scripts/preflight_cloud.py (read-only check).

11. Future BigQuery schema application

Review demo/infra/bigquery_schema.sql. It is not executed automatically. Once the sandbox dataset should exist, replace <SANDBOX_PROJECT_ID> and apply it manually via the BigQuery console or bq query -- not via any script in this repository.

12. Future Pub/Sub setup checklist

See demo/infra/pubsub_resource_plan.md for placeholder topic/subscription names and their roles. Resources are created manually, later, by the user -- never automatically by this codebase.

13. Future Dataflow submission checklist

See demo/infra/dataflow_submission_plan.md. In summary, once GCP resources exist and .env is populated:

.venv\Scripts\python demo\scripts\submit_dataflow.py `
  --enable-cloud-mode --execute-dataflow-job `
  --confirm-synthetic-sandbox-only I_CONFIRM_SYNTHETIC_SANDBOX_ONLY

Without all three flags (and complete .env configuration), the script only prints the dry-run command and refuses to do anything else.

14. Stopping the future streaming Dataflow job

.venv\Scripts\python demo\scripts\stop_dataflow.py `
  --job-id <JOB_ID> --confirm-stop-sandbox-job

Without both flags, it only prints the dry-run gcloud dataflow jobs cancel command.

15. Cost-control and cleanup checklist

See demo/infra/cost_and_cleanup_plan.md and demo/scripts/cleanup_instructions.md. In short: separate billing project, budget alert, conservative worker limits, stop the streaming job after every demo, and periodically delete sandbox BigQuery/Pub/Sub/GCS artifacts.

16. Troubleshooting boundaries

This project will not, under any circumstances triggered by normal use:

• run gcloud, bq, gsutil, Terraform, or Docker;
• create, modify, or delete a cloud resource;
• read your local credentials, ADC files, or environment secrets;
• submit a Dataflow job without all three explicit cloud gates;
• send raw synthetic events, raw invalid records, or any forbidden field to the browser, to a local file, or (in the future) to Pub/Sub/BigQuery.

If python demo/scripts/validate_project.py or pytest fail, that is a bug in this sandbox's own code, not a sign that cloud setup is needed -- local mode never needs cloud configuration to work.

---

Repository layout

beam-dataflow-sandbox/                  (this project, built at the repo root)
├── beam_pipeline_toolkit/  Reusable, generic Apache Beam pipeline framework +
│                           Dataflow control-plane API + safety layer (not
│                           demo-specific; see beam_pipeline_toolkit/README.md)
├── demo/                   Everything specific to this presentation demo:
│   ├── app/                FastAPI dashboard + scenario publisher backend
│   ├── pipeline/           Beam pipeline (DirectRunner today, DataflowRunner later)
│   ├── shared/             Synthetic event schema, generators, and the safety validator
│   ├── scripts/            Local demo/dashboard runners + gated cloud scripts
│   ├── infra/              Cloud setup plans (BigQuery/Pub/Sub/IAM/Dataflow/cost) -- not executed
│   ├── config/             Example sandbox config + scenario reference (documentation only)
│   └── docs/               Live Dataflow Demo dashboard tab runbook (disabled by default)
├── tests/
│   ├── demo/               pytest suite for demo/ (schema, validation, filter, aggregation,
│   │                       API, cloud gates)
│   └── toolkit/            pytest suite for beam_pipeline_toolkit/
├── setup.py                Shared --setup_file used by DataflowRunner to stage both
│                           beam_pipeline_toolkit/ and demo/ packages onto workers
├── .env / .env.example     Shared env-config story for both packages
└── runtime/                Generated local run artifacts (gitignored except README/.gitkeep)

See demo/docs/live_dataflow_demo_runbook.md for the Live Dataflow Demo dashboard tab (disabled by default).