Dataproc Serverless¶

executor.name: dataproc_serverless submits an asynchronous PySpark batch using the active gcloud login. It requires a pre-existing GCS staging bucket, worker service account, and permissions to use Dataproc Serverless and write the configured output prefix. Submission never creates IAM bindings, buckets, clusters, or images.

Principal	Required existing access
Submitter (the active `gcloud` login)	Permission to submit and inspect the Dataproc batch, and to write the staging URI. Artifact Registry Reader on the maintained private repository is needed only to list or resolve image tags, which submitting against the default pin never does.
Worker service account	Dataproc worker access, read/write on the staging and output prefixes, Artifact Registry Reader on the private default-image repository, and `roles/aiplatform.user` for remote annotation. Which identity this is differs by route: see Which identity runs the batch.
External-image operator	Only when overriding the default pin: a pinned image URI that the worker service account can pull.

Which identity runs the batch¶

Route	Identity	Configured by
Generation	The account you name; required	`executor.dataproc.service_account`
Annotation	The project’s Dataproc runtime service account	Nothing. The annotation `dataproc` block has no service-account field.

Grant the worker access above to whichever of the two applies to the route you are submitting. If your project has no dedicated worker account, its Compute Engine default account works once it carries that access. List the accounts you can use:

gcloud iam service-accounts list --project=<project>

When the annotation configuration sets inference.api_key_secret, that identity additionally needs roles/secretmanager.secretAccessor on the secret rather than on the project:

gcloud secrets add-iam-policy-binding SECRET \
  --member="serviceAccount:SA" \
  --role="roles/secretmanager.secretAccessor"

Use the Managed Service for Apache Spark IAM roles guide, custom-container guide, and Artifact Registry access-control guide to provision that access before submitting.

Remote generation requires storage.backend: signal_dataset (or a compatible native dataset-store plugin) and a gs:// output URI. Workers return strict, attempt-keyed shard receipts. The coordinator submits all receipts to root-last publication, which selects winning attempts and creates the ordinal index.

Immutable publication boundary¶

Signal Dataset owns immutable shard writes and root-last publication. RFGen owns Spark task scheduling, attempt receipts, expected-work completeness, and dataset/run identifiers. See Storage layouts.

Runnable configuration and lifecycle¶

Start from the complete maintained baseline, then replace its executor, storage, and run blocks with the following values (keep the baseline’s scene, channel, emitter, placement, and label blocks):

rfgen init narrowband-baseline ./config

executor:
  name: dataproc_serverless
  parallelism: 1
  dataproc:
    project: rf-foundation-models
    region: us-central1
    staging_uri: gs://my-precreated-staging/rfgen
    service_account: rfgen-worker@my-project.iam.gserviceaccount.com
    # image_uri: us-central1-docker.pkg.dev/my-project/rfgen/rfgen@sha256:...
storage:
  backend: signal_dataset
  path: gs://my-precreated-output/training-set
run:
  run_id: training-set-v1
  num_samples: 100000
  shard_size: 1000

Validate and submit it with:

rfgen validate --config-dir ./config
rfgen generate --config-dir ./config

Generation returns after submission and prints the batch id plus exact status and cancel commands. The corresponding manual forms are:

gcloud dataproc batches describe BATCH_ID --region=REGION --project=PROJECT
gcloud dataproc batches cancel BATCH_ID --region=REGION --project=PROJECT --quiet

wait(handle) and cancel(handle) are the executor API equivalents. The remote handle is opaque and belongs to the executor that submitted it.

Unified observation submissions¶

A configuration with projections uses the same ScenePlan, projection preflight, aggregate adapter, and exactly-one-record-per-sample contract as local execution. It additionally sets executor.dataproc.observation_contract_version: 1.

Before creating a batch, RFGen requires a digest-pinned image present in the qualified observation-runtime map, resolves all projection descriptors, checks store capabilities, validates staged plugin distributions, and rejects an external ScenePlan world. Projection-enabled Dataproc currently accepts only no world or a sionna://builtin/... world; other world bundles are not staged to workers.

The qualification map binds an exact image digest to the runtime build identity measured inside that image. The image workflow compares a real communications observation’s local and container encode/decode semantics before updating the map. A new source commit can therefore contain the feature while its map is still empty: this is a fail-closed release state. Publish and qualify the image, then use the workflow-generated pin before advertising remote availability.

The staged control objects are create-only and generation-verified. The driver and workers independently verify the observation contract, image/runtime identity, ordered package manifest, projection descriptors, and realized archives before plugin discovery or record work. A mismatch stops the job instead of accepting a newer configuration with older behavior. Retry attempts retain the same logical identities; publication selects one winning attempt per work ID.

See Generate one unified observation dataset for the local to remote configuration change.

Runtime image¶

Both remote routes default to a maintained image, pinned by digest in rfgen.remote_image.DEFAULT_DATAPROC_IMAGE. A configuration that names no image submits against that pin: annotation leaves dataproc.image_digest unset, generation leaves executor.dataproc.image_uri unset. Selecting or building an image is an override, described under Overriding the pin.

The pin is a digest, not a tag, so a recorded run can prove afterwards what it ran. A tag can be moved; a digest cannot. Dataproc Serverless documents --container-image in tag form only, and the digest form is verified here by use rather than by its documentation: the qualification runs cited under Validation boundary submitted digest-pinned images.

The worker service account needs Artifact Registry Reader on that repository to pull the image. The repository is private and lives in the rf-foundation-models project, so for a worker in any other project this is a cross-project grant that a maintainer of rf-foundation-models has to make.

When a rebuild is required¶

Rarely, and specifically not because your checkout is newer than the image. Shipping plugin packages is not a reason either: it works against the default pin.

For annotation, a submission carries the whole annotation configuration, and the image ignores configuration keys it does not declare, so adding a field does not by itself break an image that predates it.

That protects new keys, not new values inside a key the image already declares. An image validates the contents of every field it knows about against its own registries, so a value the submitting checkout added is rejected rather than ignored. Both of these are submission-time failures in the driver, against an image that is merely older than the checkout:

Value error, declared_evidence quantity fields have no registered extractor:
['duty_cycle']; supported: ['frequency_offset_hz', 'occupied_bandwidth_hz',
'sinr_db', 'snr_db', 'time_interval_s']

Value error, declared_evidence configuration applies only to the
declared-evidence template

The second is a renamed template: an image predating the rename accepts neither the new caption.declared.v1 nor the name it replaced. Because a template id and a quantity-field name are both values, no version handshake catches either one, and the batch fails after it has been billed. When you change a template id or add a quantity field, rebuild the image and pin the new digest.

Two further things force a rebuild. A change to what the remote driver needs from the configuration bumps rfgen.annotation.config.REMOTE_CONTRACT_VERSION, and an image implementing an older contract then refuses the submission and names both versions:

RuntimeError: invalid_bootstrap: the submission could not be read:
submission_requires_remote_contract_v2: this image implements v1. Submit
against a newer pinned annotation image.

A change to the bootstrap descriptor or to the staged PySpark wrapper is refused earlier, before the submission is read, so it reports itself differently:

RuntimeError: invalid_bootstrap: the bootstrap descriptor does not match the
shape this image implements; the submitting checkout is likely newer than the
pinned image

The wrapper check is symmetric: a newer image also refuses an older checkout’s staged application, with the staged PySpark application is not the expected wrapper.

The signal_dataset snapshot route stages no submission and no bootstrap descriptor, so it has no version handshake at all: an image predating the route simply lacks rfgen.annotation.snapshot_bootstrap, and the batch dies on an ImportError in the driver rather than on a named refusal.

The maintained pin contains the route and has been run with it, so the route’s presence is not a reason to rebuild. Its contract still can be: see When a rebuild is required, which lists the changes that fail inside the driver with no handshake to catch them. See Generate, then annotate and the availability limit in rfgen.annotation.

Legacy generation behaves differently on purpose. It stages its whole driver from the submitting checkout, where annotation stages only a two-line wrapper, but the configuration models and every generation primitive still come from the image, so its configuration is coupled to the image just as annotation’s is. A driver that calls into an image API that has since changed fails as an ordinary ImportError or AttributeError, not as a named refusal. What differs for configuration is the payload: a generation submission carries only the settings the operator actually wrote, so a setting the image does not implement cannot be quietly skipped without producing a dataset nobody asked for. The batch stops instead, naming what it could not honor:

ValueError: generation_config_not_implemented_by_this_image: channel.chain.0.foo.
Submit against a newer pinned image, or remove those settings.

So a legacy generation configuration that stays within what the pinned image implements submits unchanged, and one that reaches past it says so.

Projection-enabled generation uses the explicit observation contract and qualified image/runtime handshake above. It refuses an unqualified digest before billing.

The remedy in each of these cases is a newer pinned image, which normally means updating rfgen. Build only when you are changing rfgen itself and want to run remote work against your own change before it merges.

A field the image cannot read¶

RuntimeError: invalid_bootstrap: the submission could not be read:
invalid_submission

The submission carried a value the image’s own model rejects. Unknown keys are ignored, so this is not a field the image has never heard of: it is a field it knows with a value or type it does not accept. Read the batch’s driver output, which names the field and the reason. Then use a newer image, or set the field to something the image you are pointing at accepts.

Images built before the contract version existed¶

An image published before remote_contract_version was added to the submission fails on any submission written by a current checkout, because it predates both that field and the tolerance that would ignore it. It reports the failure with no reason attached at all:

RuntimeError: invalid_bootstrap

Every image the repository has ever pinned as the default postdates that field, so you reach this only by naming an image yourself. Point at the default pin, or at any image published after 2026-08-09. Updating rfgen does not help, because the pin your checkout already carries is newer than the image you named.

Overriding the pin¶

Published tags are dated, <rfgen-version>-<YYYYMMDD>, with a -r2, -r3 suffix for a second or later build on the same day. List the newest few:

gcloud artifacts docker images list \
  us-central1-docker.pkg.dev/rf-foundation-models/rfgen/rfgen \
  --include-tags --sort-by=~CREATE_TIME --limit=5 \
  --project=rf-foundation-models

Do not select the tag matching your installed rfgen.__version__. The package version moves only on release, so many source trees share one version string, and that tag names whichever of them was built last.

Resolve the tag you chose to the immutable reference a submission records:

gcloud artifacts docker images describe \
  us-central1-docker.pkg.dev/rf-foundation-models/rfgen/rfgen:${SELECTED_TAG} \
  --project=rf-foundation-models \
  --format='value(image_summary.fully_qualified_digest)'

Set the returned ...rfgen@sha256:... value on the route you are submitting: dataproc.image_digest for annotation, executor.dataproc.image_uri for generation.

Do not try to judge compatibility in advance from timestamps. An image older than your last commit is usually still fine, because most changes touch neither the remote contract nor the bootstrap shape, and a rule based on commit dates rejects working images. The batch is the check, and it names what it refused.

Building your own¶

Build by hand to run remote work against a change that has not merged yet. Where image publishing is enabled for the repository, a merge to main that touches the package or the image build publishes a new image and commits the new digest into rfgen.remote_image.DEFAULT_DATAPROC_IMAGE; where it is not, that constant is whatever a person last committed. Pick the next unused suffix for today, since republishing an existing tag is forbidden.

gcloud builds submit --config=docker/dataproc-serverless/cloudbuild.yaml \
  --substitutions=_IMAGE_TAG=<rfgen-version>-<YYYYMMDD>[-rN],_RFGEN_VERSION=<rfgen-version> .

_IMAGE_TAG and _RFGEN_VERSION are independent substitutions. The tag is free-form and carries the date; _RFGEN_VERSION must equal pyproject.toml’s version, which the Docker build asserts. Pass _RFGEN_VERSION explicitly, because cloudbuild.yaml carries a literal fallback that is only correct until the next version bump. The tag identifies the build; the resolved digest is what a submission records.

What the image contains¶

The image is built from docker/dataproc-serverless/Dockerfile, which installs rfgen with its maintained remote dependencies and runtime utilities. It does not contain every use-case package; ship the ones a configuration needs with the submission instead, as described under Shipping plugin packages. Sionna matters: every shipped rfgen init template except multi-emitter-baseline and radar-response selects a Sionna propagation backend, so an image without that extra fails a generation batch with BackendUnavailableError even though the configuration is correct. The radar-response template additionally requires a separately supplied RadarSimPy artifact and is not supported by this image. Dataproc Serverless supplies Spark and PySpark, so the custom container must not package its own Spark distribution.

The driver resolves the configured BaseDatasetStore. Workers write immutable Signal Dataset shard attempts and return receipts; the coordinator publishes the dataset root only after all expected work IDs arrive.

Dataproc runs the container as UID/GID 1099 (spark): the image uses PYSPARK_PYTHON to select its rfgen virtual environment instead of the service-mounted Conda environment, and any runtime cache path must be writable by that identity. Its CPU PyTorch installation avoids bundling unused CUDA libraries.

Shipping plugin packages¶

One maintained image serves multiple use cases. A configuration that selects a node from a use-case package the image does not contain ships that package with the submission instead of requiring a rebuild:

executor:
  name: dataproc_serverless
  dataproc:
    # ... project, region, staging_uri, service_account ...
    extra_packages:
      - dist/rfgen_signal_atlas_cuas-0.2.0-py3-none-any.whl

Build the wheel from the use-case package first:

uv build --wheel use_cases/signal-atlas/cuas-v1 -o dist

Match the version to the package: the wheel name follows its pyproject.toml.

Each entry must be a submitter-local .whl or .zip; nothing else is supported. A remote (gs:// or any other scheme) reference is rejected, because RFGen cannot verify its bytes before billing and has no mechanism to verify them afterward either. Build the artifact where the submission runs.

A reference’s shape is checked when the configuration is read, so an unusable suffix or scheme fails rfgen validate before anything is staged: each entry must be non-blank and end in .whl or .zip. Entries are also deduplicated by the name they would be staged under, not by the string you wrote: because a .whl is renamed to .zip when it is staged (see below), dist/foo-1.0-py3-none-any.whl and other/foo-1.0-py3-none-any.zip collide, and the configuration is refused rather than silently keeping one.

Whether a local file actually exists, is a real zip archive, and declares at least one rfgen.* entry-point group in its dist-info/entry_points.txt is a separate, later check: it runs at submission time, immediately before staging and before the batch is created, because rfgen validate may run before the artifact is even built and cannot assume the filesystem it is handed is the one that will submit. A package that fails any of these checks is refused with zero gcloud calls, before the active-login check and before anything is billed.

Verified packages are uploaded create-only into a packages/ subdirectory of the submission’s staging root, and every staged URI is passed to gcloud dataproc batches submit pyspark --py-files, Spark’s own mechanism for placing files on sys.path for the driver and every worker. Spark’s --py-files loads only .py, .egg, and .zip files by suffix; a .whl on that flag is silently ignored. A wheel is already a valid zip archive, so a local .whl is renamed to .zip when it is staged (the bytes, including the dist-info/entry_points.txt that makes the plugin discoverable, are unchanged) purely so Spark agrees to load it.

Dependencies are not resolved: a shipped package’s own third-party requirements must already be present in the pinned image. extra_packages travels with the rest of the submitted configuration to the driver, which revalidates the same shape rules there; the driver does not act on the field itself, because by the time it runs the packages are already staged and already on sys.path.

Shipping a package to the worker does not relieve the submitting machine of needing it too. rfgen generate compiles the whole configured graph locally — resolving and binding every node, including one this mechanism ships — before it resolves or submits any executor, so a use-case package named in the configuration has to be installed in the submitter’s own environment (for example with uv pip install -e use_cases/<variant> or by installing the same built wheel) or the submission never gets far enough to reach extra_packages at all. extra_packages closes only the worker-side half of that gap.

A version mismatch between the rfgen core installed in the pinned image and the core a shipped package actually requires is not caught by anything described on this page. If the package declares its own capability check — see, for example, the pattern in use_cases/signal-atlas/pnt-v1/rfgen_signal_atlas_pnt/__init__.py — that check runs again on the worker at import time and fails the same way it would locally, but only once a task actually imports the package, which is mid-batch rather than at submission. It surfaces inside the same per-shard error evidence any other node failure produces, not as a distinct refusal.

The image must still be new enough for the generation path the submission uses, which is a separate matter. The driver is staged from your checkout while every generation primitive comes from the image, so an image predating a primitive the driver calls fails at driver start:

ImportError: cannot import name 'generate_native_records' from
'rfgen.generation.single_record'

That is what a stale pin looks like: mid-batch, not at submission. The image’s own build-time smoke check imports the remote driver’s entry points so the mismatch surfaces when the image is built instead.

Validation boundary¶

Generation¶

The following is the guarded Dataproc generation E2E. It is excluded from ordinary CI and remains available-only rather than production-qualified:

RFGEN_ANNOTATION_E2E=YES \
RFGEN_ANNOTATION_E2E_REGION=us-central1 \
RFGEN_ANNOTATION_E2E_STAGING_URI=gs://my-staging/rfgen \
RFGEN_ANNOTATION_E2E_OUTPUT_URI=gs://my-output/rfgen-e2e \
RFGEN_ANNOTATION_E2E_SERVICE_ACCOUNT=rfgen-worker@my-project.iam.gserviceaccount.com \
pytest -m dataproc_e2e tests/manual/test_dataproc_serverless.py

GOOGLE_CLOUD_PROJECT optionally overrides the default project for that test.

Annotation¶

The annotation live-E2E is separate and manually gated. Bounded qualifications are recorded under Evidence, which lists what each run covers. None qualifies hosted text semantically or establishes fleet reliability. The Dataproc annotation route has a bounded, manually qualified workflow that requires an immutable image and a worker-readable dependency prefix; the Vertex Batch workflow over WebDataset shards that stood beside it went with that store, and so did the live test that used RFGEN_ANNOTATION_E2E_DEPS_BUCKET and RFGEN_ANNOTATION_E2E_IMAGE_DIGEST — nothing in this build reads either variable. See the unified annotation manual qualification.

Only existing local Golden Paths are production-qualified.