diff --git a/dgf/src/sampling/offline_distributed/BUILD b/dgf/src/sampling/offline_distributed/BUILD
index aceb8a9..eeebfe7 100644
--- a/dgf/src/sampling/offline_distributed/BUILD
+++ b/dgf/src/sampling/offline_distributed/BUILD
@@ -17,6 +17,7 @@ py_library(
     name = "offline_distributed_gcp",
     srcs = ["offline_distributed_gcp.py"],
     deps = [
+        # dataclasses_json dep,
         "//dgf/src/data:schema",
         "//dgf/src/io:schema",
         "//dgf/src/sampling:config",
diff --git a/dgf/src/sampling/offline_distributed/offline_distributed_gcp.py b/dgf/src/sampling/offline_distributed/offline_distributed_gcp.py
index f1b8389..3c24855 100644
--- a/dgf/src/sampling/offline_distributed/offline_distributed_gcp.py
+++ b/dgf/src/sampling/offline_distributed/offline_distributed_gcp.py
@@ -16,12 +16,15 @@
 
 from __future__ import annotations
 
+import dataclasses
 import datetime
+import json
 import os
 import subprocess
 import time
 from typing import Any
 
+import dataclasses_json
 from dgf.src.data import schema as schema_lib
 from dgf.src.io import schema as io_schema
 from dgf.src.sampling import config as config_lib
@@ -122,6 +125,8 @@ def _validate_seed_args(
     num_seeds: int | None,
     num_samples_per_seed: int,
     input_seed_container: str,
+    filter_seed_node: str | None = None,
+    input_seeds: str | None = None,
 ) -> None:
   """Validates the arguments controlling the selection of the seeds.
 
@@ -129,6 +134,8 @@ def _validate_seed_args(
     num_seeds: Number of seeds to select among the available ones, or None.
     num_samples_per_seed: Number of samples to generate for each seed.
     input_seed_container: Container of the input seeds.
+    filter_seed_node: Optional filter on the seed nodes.
+    input_seeds: Optional path to input seeds.
 
   Raises:
     ValueError: If the arguments are inconsistent.
@@ -152,6 +159,12 @@ def _validate_seed_args(
         f"Unsupported input_seed_container {input_seed_container!r}. Supported"
         f" containers are {sorted(_SUPPORTED_INPUT_SEED_CONTAINERS)}."
     )
+  if filter_seed_node is not None and input_seeds is not None:
+    raise ValueError(
+        "filter_seed_node and input_seeds are exclusive: filter_seed_node only"
+        " applies to the seed nodes taken from the root nodeset of the sampling"
+        " plan."
+    )
 
 
 def _get_job_urls(
@@ -211,6 +224,29 @@ def _write_sampling_plan(
   return sampling_config_path
 
 
+@dataclasses_json.dataclass_json
+@dataclasses.dataclass
+class OfflineDistributedSamplingConfig:
+  """Extra arguments for the offline distributed sampler.
+
+  This config is passed to the sampler as a single JSON encoded `--dgf_config`
+  argument. Each field overrides the default of the sampler flag of the same
+  name, and is ignored if that flag is set on the command line. A `None` field
+  does not override anything.
+
+  See third_party/py/dgf/src/bin/google/offline_distributed_sampling.go.
+  """
+
+  output_container: str | None = None
+  shard_prefix: str | None = None
+  num_samples_per_seed: int | None = None
+  input_seeds: str | None = None
+  input_seed_container: str | None = None
+  filter_seed_node: str | None = None
+  debug_sampling: bool | None = None
+  random_seed: int | None = None
+
+
 def _create_custom_job(
     input_path: str,
     output_path: str,
@@ -223,18 +259,45 @@ def _create_custom_job(
     input_seeds: str | None,
     input_seed_container: str,
     random_seed: int,
+    num_output_shards: int,
+    output_container: str,
+    shard_prefix: str,
+    filter_seed_node: str | None,
+    debug_sampling: bool,
     staging_location: str,
     temp_location: str,
     display_name: str,
 ) -> aiplatform.CustomJob:
   """Builds and instantiates the Vertex AI CustomJob."""
+  # Only the arguments allowlisted for the sampler image in
+  # java/com/google/cloud/ai/platform/boq/shared/configuration/config/template/custom_training_base.pi
+  # can be passed to a Vertex AI CustomJob: the sampler options are grouped into
+  # a single JSON encoded "--dgf_config" argument.
+  config = OfflineDistributedSamplingConfig(
+      output_container=output_container,
+      shard_prefix=shard_prefix,
+      num_samples_per_seed=num_samples_per_seed,
+      input_seeds=input_seeds,
+      input_seed_container=(
+          input_seed_container if input_seeds is not None else None
+      ),
+      filter_seed_node=filter_seed_node,
+      debug_sampling=debug_sampling,
+      random_seed=random_seed,
+  )
+  # Unset fields are not exported: they do not override the sampler defaults.
+  dgf_config = json.dumps({
+      key: value
+      for key, value in dataclasses.asdict(config).items()
+      if value is not None
+  })
   args = [
       f"--input_graph={input_path}",
       f"--output_samples={output_path}",
       f"--sampling_config={sampling_config_path}",
       f"--num_seeds={num_seeds if num_seeds is not None else 0}",
-      f"--num_samples_per_seed={num_samples_per_seed}",
-      f"--random_seed={random_seed}",
+      f"--num_output_shards={num_output_shards}",
+      f"--dgf_config={dgf_config}",
       "--runner=dataflow",
       f"--project={project}",
       f"--region={region}",
@@ -248,9 +311,6 @@ def _create_custom_job(
       f"--sdk_container_image={_DEFAULT_IMAGE_URI}",
       f"--worker_binary={_WORKER_BINARY}",
   ]
-  if input_seeds is not None:
-    args.append(f"--input_seeds={input_seeds}")
-    args.append(f"--input_seed_container={input_seed_container}")
 
   worker_pool_specs = [{
       "machine_spec": {
@@ -365,6 +425,11 @@ def offline_distributed_sampler_gcp(
     num_samples_per_seed: int = 1,
     input_seeds: str | None = None,
     input_seed_container: str = "TFRECORD",
+    filter_seed_node: str | None = None,
+    num_output_shards: int = 20,
+    output_container: str = "TFRECORD",
+    shard_prefix: str = "samples",
+    debug_sampling: bool = False,
     random_seed: int = 42,
     temp_location: str | None = None,
     staging_location: str | None = None,
@@ -427,6 +492,14 @@ def offline_distributed_sampler_gcp(
       generated. If None, the seeds are all the nodes of the seed nodeset of the
       sampling plan.
     input_seed_container: Container of `input_seeds`: "TFRECORD" or "RECORDIO".
+    filter_seed_node: Optional filter on the seed nodes of the root nodeset of
+      the sampling plan, with the syntax '<feature name>=<required value>' (e.g.
+      '#split=train'). Only the nodes whose feature is equal to the required
+      value are used as seeds. Exclusive with `input_seeds`.
+    num_output_shards: Number of shards to write.
+    output_container: Format of output samples: "TFRECORD" or "RECORDIO".
+    shard_prefix: Prefix for output shards (e.g., 'samples' or 'Shard').
+    debug_sampling: Enable deterministic debug sampling mode.
     random_seed: Seed of the random number generator.
     temp_location: Optional GCS temporary directory for Dataflow.
     staging_location: Optional GCS staging directory for Dataflow and Vertex AI.
@@ -441,7 +514,17 @@ def offline_distributed_sampler_gcp(
       be resolved.
   """
   input_path, output_path = _validate_paths(input_path, output_path)
-  _validate_seed_args(num_seeds, num_samples_per_seed, input_seed_container)
+  _validate_seed_args(
+      num_seeds=num_seeds,
+      num_samples_per_seed=num_samples_per_seed,
+      input_seed_container=input_seed_container,
+      filter_seed_node=filter_seed_node,
+      input_seeds=input_seeds,
+  )
+  if num_output_shards < 1:
+    raise ValueError(
+        f"num_output_shards cannot be less than one, got {num_output_shards}."
+    )
 
   if project is None:
     project = _get_default_gcp_project()
@@ -479,6 +562,11 @@ def offline_distributed_sampler_gcp(
       input_seeds=input_seeds,
       input_seed_container=input_seed_container,
       random_seed=random_seed,
+      num_output_shards=num_output_shards,
+      output_container=output_container,
+      shard_prefix=shard_prefix,
+      filter_seed_node=filter_seed_node,
+      debug_sampling=debug_sampling,
       staging_location=staging_location,
       temp_location=temp_location,
       display_name=display_name,
diff --git a/dgf/src/sampling/offline_distributed/offline_distributed_gcp_test.py b/dgf/src/sampling/offline_distributed/offline_distributed_gcp_test.py
index 08a88c0..646dd92 100644
--- a/dgf/src/sampling/offline_distributed/offline_distributed_gcp_test.py
+++ b/dgf/src/sampling/offline_distributed/offline_distributed_gcp_test.py
@@ -14,7 +14,9 @@
 
 """Unit tests for offline_distributed_gcp."""
 
+from collections.abc import Sequence
 import json
+from typing import Any
 from unittest import mock
 from absl.testing import absltest
 from dgf.src.sampling import config as config_lib
@@ -23,6 +25,23 @@
 from dgf.src.util import log
 from dgf.src.util.weak_dep.weak_dep_aiplatform import aiplatform
 
+_DGF_CONFIG_PREFIX = "--dgf_config="
+
+
+def _get_dgf_config(container_args: Sequence[str]) -> dict[str, Any]:
+  """Returns the JSON decoded "--dgf_config" argument of the sampler."""
+  configs = [
+      arg.removeprefix(_DGF_CONFIG_PREFIX)
+      for arg in container_args
+      if arg.startswith(_DGF_CONFIG_PREFIX)
+  ]
+  if len(configs) != 1:
+    raise ValueError(
+        f"Expected exactly one {_DGF_CONFIG_PREFIX} argument, got"
+        f" {container_args}."
+    )
+  return json.loads(configs[0])
+
 
 class OfflineDistributedGcpTest(absltest.TestCase):
 
@@ -82,15 +101,22 @@ def test_offline_distributed_sampler_gcp_blocking_success(
         container_args,
     )
     self.assertIn("--num_seeds=500", container_args)
-    self.assertIn("--num_samples_per_seed=1", container_args)
-    self.assertIn("--random_seed=42", container_args)
     self.assertIn("--num_workers=3", container_args)
     self.assertIn("--max_num_workers=3", container_args)
     self.assertIn("--runner=dataflow", container_args)
-    # Without input seeds, the sampler seeds on the nodes of the seed nodeset.
-    self.assertNoCommonElements(
-        ["--input_seeds", "--input_seed_container"],
-        [arg.split("=")[0] for arg in container_args],
+    self.assertIn("--num_output_shards=20", container_args)
+    # The sampler options are passed in a single JSON encoded argument. Without
+    # input seeds, the sampler seeds on the nodes of the seed nodeset. Unset
+    # fields are not exported.
+    self.assertEqual(
+        _get_dgf_config(container_args),
+        {
+            "output_container": "TFRECORD",
+            "shard_prefix": "samples",
+            "num_samples_per_seed": 1,
+            "debug_sampling": False,
+            "random_seed": 42,
+        },
     )
 
     mock_job.submit.assert_called_once()
@@ -322,11 +348,10 @@ def test_input_seeds(self, mock_custom_job_cls, mock_open_write):
 
     _, kwargs = mock_custom_job_cls.call_args
     container_args = kwargs["worker_pool_specs"][0]["container_spec"]["args"]
-    self.assertIn(
-        "--input_seeds=gs://my_bucket/seeds@10.recordio", container_args
-    )
-    self.assertIn("--input_seed_container=RECORDIO", container_args)
-    self.assertIn("--random_seed=7", container_args)
+    config = _get_dgf_config(container_args)
+    self.assertEqual(config["input_seeds"], "gs://my_bucket/seeds@10.recordio")
+    self.assertEqual(config["input_seed_container"], "RECORDIO")
+    self.assertEqual(config["random_seed"], 7)
 
   @mock.patch.object(offline_distributed_gcp.filesystem, "open_write")
   @mock.patch.object(aiplatform, "CustomJob")
@@ -346,9 +371,58 @@ def test_num_samples_per_seed(self, mock_custom_job_cls, mock_open_write):
 
     _, kwargs = mock_custom_job_cls.call_args
     container_args = kwargs["worker_pool_specs"][0]["container_spec"]["args"]
-    self.assertIn("--num_samples_per_seed=3", container_args)
+    self.assertEqual(_get_dgf_config(container_args)["num_samples_per_seed"], 3)
     self.assertIn("--num_seeds=0", container_args)
 
+  @mock.patch.object(offline_distributed_gcp.filesystem, "open_write")
+  @mock.patch.object(aiplatform, "CustomJob")
+  def test_only_allowed_args_are_passed(
+      self, mock_custom_job_cls, mock_open_write
+  ):
+    mock_open_write.return_value.__enter__.return_value = mock.MagicMock()
+    mock_custom_job_cls.return_value = mock.MagicMock()
+
+    offline_distributed_gcp.offline_distributed_sampler_gcp(
+        input_path="gs://my_bucket/graph",
+        output_path="gs://my_bucket/samples",
+        plan=self.simple_plan,
+        schema=self.mock_schema,
+        project="test-proj",
+        input_seeds="gs://my_bucket/seeds@10.recordio",
+        num_samples_per_seed=3,
+        blocking=False,
+    )
+
+    _, kwargs = mock_custom_job_cls.call_args
+    container_args = kwargs["worker_pool_specs"][0]["container_spec"]["args"]
+    # Vertex AI only accepts the argument keys allowlisted for the sampler image
+    # in
+    # java/com/google/cloud/ai/platform/boq/shared/configuration/config/template/custom_training_base.pi:
+    # the other sampler options must go through "--dgf_config".
+    self.assertContainsSubset(
+        [arg.split("=")[0] for arg in container_args],
+        [
+            "--input_graph",
+            "--output_samples",
+            "--sampling_config",
+            "--num_seeds",
+            "--num_output_shards",
+            "--dgf_config",
+            "--runner",
+            "--project",
+            "--region",
+            "--worker_machine_type",
+            "--num_workers",
+            "--max_num_workers",
+            "--autoscaling_algorithm",
+            "--staging_location",
+            "--temp_location",
+            "--environment_type",
+            "--sdk_container_image",
+            "--worker_binary",
+        ],
+    )
+
   def test_num_seeds_and_num_samples_per_seed_are_exclusive(self):
     with self.assertRaisesRegex(ValueError, "are exclusive"):
       offline_distributed_gcp.offline_distributed_sampler_gcp(
@@ -385,6 +459,98 @@ def test_invalid_num_seeds_raises_error(self):
           num_seeds=-1,
       )
 
+  @mock.patch.object(offline_distributed_gcp.filesystem, "open_write")
+  @mock.patch.object(aiplatform, "CustomJob")
+  def test_output_options(self, mock_custom_job_cls, mock_open_write):
+    mock_open_write.return_value.__enter__.return_value = mock.MagicMock()
+    mock_custom_job_cls.return_value = mock.MagicMock()
+
+    offline_distributed_gcp.offline_distributed_sampler_gcp(
+        input_path="gs://my_bucket/graph",
+        output_path="gs://my_bucket/samples",
+        plan=self.simple_plan,
+        schema=self.mock_schema,
+        project="test-proj",
+        output_container="RECORDIO",
+        shard_prefix="Shard",
+        num_output_shards=10,
+        blocking=False,
+    )
+
+    _, kwargs = mock_custom_job_cls.call_args
+    container_args = kwargs["worker_pool_specs"][0]["container_spec"]["args"]
+    self.assertIn("--num_output_shards=10", container_args)
+    config = _get_dgf_config(container_args)
+    self.assertEqual(config["output_container"], "RECORDIO")
+    self.assertEqual(config["shard_prefix"], "Shard")
+
+  @mock.patch.object(offline_distributed_gcp.filesystem, "open_write")
+  @mock.patch.object(aiplatform, "CustomJob")
+  def test_filter_seed_node(self, mock_custom_job_cls, mock_open_write):
+    mock_open_write.return_value.__enter__.return_value = mock.MagicMock()
+    mock_custom_job_cls.return_value = mock.MagicMock()
+
+    offline_distributed_gcp.offline_distributed_sampler_gcp(
+        input_path="gs://my_bucket/graph",
+        output_path="gs://my_bucket/samples",
+        plan=self.simple_plan,
+        schema=self.mock_schema,
+        project="test-proj",
+        filter_seed_node="#split=train",
+        blocking=False,
+    )
+
+    _, kwargs = mock_custom_job_cls.call_args
+    container_args = kwargs["worker_pool_specs"][0]["container_spec"]["args"]
+    config = _get_dgf_config(container_args)
+    self.assertEqual(config["filter_seed_node"], "#split=train")
+
+  @mock.patch.object(offline_distributed_gcp.filesystem, "open_write")
+  @mock.patch.object(aiplatform, "CustomJob")
+  def test_debug_sampling(self, mock_custom_job_cls, mock_open_write):
+    mock_open_write.return_value.__enter__.return_value = mock.MagicMock()
+    mock_custom_job_cls.return_value = mock.MagicMock()
+
+    offline_distributed_gcp.offline_distributed_sampler_gcp(
+        input_path="gs://my_bucket/graph",
+        output_path="gs://my_bucket/samples",
+        plan=self.simple_plan,
+        schema=self.mock_schema,
+        project="test-proj",
+        debug_sampling=True,
+        blocking=False,
+    )
+
+    _, kwargs = mock_custom_job_cls.call_args
+    container_args = kwargs["worker_pool_specs"][0]["container_spec"]["args"]
+    config = _get_dgf_config(container_args)
+    self.assertTrue(config["debug_sampling"])
+
+  def test_filter_seed_node_and_input_seeds_are_exclusive(self):
+    with self.assertRaisesRegex(ValueError, "are exclusive"):
+      offline_distributed_gcp.offline_distributed_sampler_gcp(
+          input_path="gs://my_bucket/graph",
+          output_path="gs://my_bucket/samples",
+          plan=self.simple_plan,
+          schema=self.mock_schema,
+          project="test-proj",
+          input_seeds="gs://my_bucket/seeds",
+          filter_seed_node="#split=train",
+      )
+
+  def test_invalid_num_output_shards_raises_error(self):
+    with self.assertRaisesRegex(
+        ValueError, "num_output_shards cannot be less than one"
+    ):
+      offline_distributed_gcp.offline_distributed_sampler_gcp(
+          input_path="gs://my_bucket/graph",
+          output_path="gs://my_bucket/samples",
+          plan=self.simple_plan,
+          schema=self.mock_schema,
+          project="test-proj",
+          num_output_shards=0,
+      )
+
   def test_invalid_input_seed_container_raises_error(self):
     with self.assertRaisesRegex(
         ValueError, "Unsupported input_seed_container 'SSTABLE'"
diff --git a/examples/BUILD b/examples/BUILD
index a19360f..d3c62df 100644
--- a/examples/BUILD
+++ b/examples/BUILD
@@ -130,5 +130,7 @@ py_binary(
         # absl:app dep,
         # absl/flags dep,
         "//dgf",
+        # Weak dependency of the offline distributed sampler on GCP.
+        # google/cloud/aiplatform dep,  # keep
     ],
 )
diff --git a/examples/create_graph_samples_offline_distributed_gcp.py b/examples/create_graph_samples_offline_distributed_gcp.py
index c7858a0..0908aa2 100644
--- a/examples/create_graph_samples_offline_distributed_gcp.py
+++ b/examples/create_graph_samples_offline_distributed_gcp.py
@@ -16,20 +16,21 @@
 
 The sampling pipeline runs on Google Cloud using Vertex AI and Dataflow.
 
-Usage example:
+External usage example:
 
 ```shell
-blaze run -c opt //third_party/py/dgf/examples:create_graph_samples_offline_distributed_gcp -- \
-  --input_graph=gs://gf-experiment-gbm-test/fetch_repo/ogb_mag \
-  --output_samples=gs://gf-experiment-gbm-test/examples/ogb_mag_samples \
-  --project=graphflow-experiments-49784 \
+# This example requires the `google-cloud-aiplatform` pip package:
+pip install google-cloud-aiplatform
+
+python create_graph_samples_offline_distributed_gcp.py --input_graph=gs://gf-experiment-gbm-test/fetch_repo/ogb_mag \
+  --output_samples=gs://gf-experiment-gbm-test/examples/ogb_mag_samples_v2 \
   --seed_nodeset=paper \
   --num_hops=2 \
   --hop_width=10 \
   --num_workers=5 \
-  --num_seeds=1000 \
-  --alsologtostderr
+  --num_seeds=1000
 ```
+
 """
 
 from collections.abc import Sequence