Run on Spark¶

Submit a sharded generation job to GCP Managed Spark serverless (formerly Dataproc Serverless).

Warning

Pre-implementation. Targets v0.5+.

When to use Spark¶

Scale	Recommended runner
< 10 K samples	`local` (single process)
10 K – 1 M samples	`local` today is still single-process; use Spark for cloud / shared infra
1 M – 100 M samples	Spark serverless
100 M+ samples	Spark serverless with custom executor pools

Spark gives you autoscaling, per-shard idempotency, GCS-native I/O, and integrated monitoring, at the cost of needing GCP infra.

Prerequisites¶

pip install "rfgen[spark]"

# GCP setup
gcloud auth application-default login
gcloud config set project $GCP_PROJECT
gcloud services enable dataproc.googleapis.com

You’ll need:

A GCS bucket for output (e.g., gs://rf-fm-datasets-synth)
A service account with Dataproc Worker, Storage Admin, and Vertex AI User roles
Managed Spark serverless API enabled

The orchestration config¶

# configs/orchestration/spark_serverless.yaml
backend: spark_serverless
parallelism: 32                     # requested PySpark partition count
spark:
  project_id: ${oc.env:GCP_PROJECT}
  region: us-central1
  service_account: rfgen-runner@${oc.env:GCP_PROJECT}.iam.gserviceaccount.com
  machine_type: n2-standard-8       # 8 vCPU, 32 GB RAM
  max_executors: 32
  image_uri: gcr.io/${oc.env:GCP_PROJECT}/rfgen-spark:0.5.0

The image must contain rfgen plus all extras you need (TorchSig, Sionna, etc.). The framework ships a Dockerfile (docker/Dockerfile.spark) and a make spark-image target that builds + pushes it.

Submit¶

rfgen generate \
    +preset=wideband_detection_baseline \
    storage.path=gs://rf-fm-datasets-synth/wideband-baseline/v0.5.0 \
    orchestration=spark_serverless \
    orchestration.spark.project_id=$GCP_PROJECT

Output:

Submitting Spark batch job: rfgen-wideband-20260601-1432
  region: us-central1
  template: managed-spark-rfgen-v0.5.0
  executors: autoscale 8–32 × n2-standard-8
  estimated wall-clock: 3.5–5.5 hours

Job ID: 1234abcd-...
Console: https://console.cloud.google.com/dataproc/batches/us-central1/1234abcd...

Streaming job logs ... (Ctrl-C to detach; job continues)

Sharding and idempotency¶

Each Spark task processes one shard. Shard ID is sha256(materialized_config, shard_index)[:16]. The output path includes the shard ID, so re-running with the same config produces identical paths, which lets the framework skip-if-exists.

Property	Behavior
Shard naming	`shards/shard-{idx:06d}-{shard_id}.tar`
Re-run	Skips already-written shards (idempotent)
Partial failure	Successful sample records may persist, but the shard remains incomplete and its errors are reported; the rest of the run continues
Resume	Re-submit the same command; only missing shards regenerate

Per-shard seeding¶

shard_seed = hash(global_seed, shard_id)

Every emitter, channel, and labeler in a shard derives its RNG state from shard_seed. Two re-runs with the same global seed and the same shard ID produce byte-identical IQ.

Monitor¶

# Job state
gcloud dataproc batches describe 1234abcd-... --region us-central1

# Logs
gcloud dataproc batches logs 1234abcd-... --region us-central1 --tail

# rfgen-side progress
rfgen inspect gs://rf-fm-datasets-synth/wideband-baseline/v0.5.0 summary
# Sample count:    347_000 / 1_000_000   (34.7%)
# Shards complete: 347 / 1000
# ...

Resource sizing¶

Preset	Recommended config
`narrowband_classifier_baseline_md` (100K)	`local`. Spark adds overhead at this size.
`wideband_detection_baseline_md` (100K)	Spark, `max_executors=8`, ~30 min
`wideband_detection_baseline_lg` (1M)	Spark, `max_executors=32`, ~4 hours
`dense_urban_2_4ghz_lg` with RT	Spark + GPU pool, `max_executors=8` GPU + `max_executors=32` CPU
Anything 10M+	Spark with custom executor pool config

GPU executors (for Sionna RT)¶

Sionna RT has a CPU fallback, but production ray tracing should use GPUs. Configure a GPU executor pool:

orchestration:
  spark:
    executor_pool:
      cpu:
        machine_type: n2-standard-8
        max_executors: 32
      gpu:
        machine_type: g2-standard-8     # 1 × NVIDIA L4
        max_executors: 8
        accelerator: nvidia-l4

scene:
  channel:
    name: sionna_rt              # framework routes to GPU pool automatically

The framework partitions work: RT-channel shards go to the GPU pool, comms-only shards stay on CPU. Mix is set by the per-emitter weights and the channel choice.

Cost¶

Rough order-of-magnitude (us-central1, 2026 prices):

Preset / size	CPU-hours	GPU-hours	Cost
`wideband_baseline_md` (100K, no RT)	~30	0	~$15
`wideband_baseline_lg` (1M, no RT)	~280	0	~$140
`dense_urban_2_4ghz_lg` (1M, RT)	~280	~120	~$420

Spot VMs cut this 60–80%; the framework defaults to spot with on-demand fallback.

Provider-neutral managed-Spark control (Layer 19)¶

The shipped rfgen.managed_spark module is a strict control-plane contract, not a Dataproc submission client. Production deployments provide a provider-specific ManagedSparkAdapter; this release supplies only the deterministic in-memory reference adapter for contract tests. Do not infer from this page that the example CLI, provider maps, or cloud job submission are implemented by Layer 19.

Construct ManagedSparkJobSpecV1 with a UTC deadline, wheel URI and SHA-256, entry point, and immutable configuration maps. Keep credentials in the provider identity/adapter boundary, not in request fields, arguments, or log records. Reuse an idempotency key only for the identical request: an identical retry returns the same handle, while a different request fails closed with an idempotency conflict.

UNKNOWN is an ambiguous provider outcome. Reconcile it before wait or cancel; if it remains unknown, stop and investigate rather than submit a duplicate job. Cancellation has to converge within five minutes of its UTC deadline. Log-page tokens are opaque and bound to their handle, so retain the handle and returned token together rather than synthesizing an offset.