Generating Synthetic Training Data with Dask on Hugging Face Jobs

Training models means spending a lot of time thinking about data. At Hanno Labs, synthetic data has become one of the ways we shape a model toward the behavior we want, especially when the examples we need would be difficult or expensive to collect by hand.

Generating that data is a distributed workload of its own. Language models need GPUs, while the work around them, reading source material, filtering it, building prompts, validating outputs, and writing datasets, is mostly CPU-bound. Much of our source data already lives in Hugging Face datasets and buckets, so Hugging Face Jobs is a natural place to run the pipeline. The compute is available on demand and sits close to the data.

We already liked Dask for expressing this kind of work. What we did not have was a clean way to turn a set of Hugging Face Jobs into a Dask cluster. We built hfdask to fill that gap.

In this article, we will build that complete CPU → GPU → CPU pipeline. CPU workers prepare source documents from a mounted Hugging Face dataset, a GPU worker generates grounded question-and-answer examples with a worker-local vLLM engine, and CPU workers validate and write the accepted records to a Hugging Face bucket. The Dask program remains ordinary Python. hfdask handles the temporary cluster around it.

Table of Contents

• What is hfdask?
• Why Synthetic Data on Hugging Face Jobs?
• Pipeline Components
• Source Dataset
• Generation Model
• Prompt and Output Schema
• Cluster Configuration
• Validation
• Complete Pipeline
• Running the Pipeline
• Scaling Generation
• What hfdask Handles
• Additional Resources

What is hfdask?

hfdask turns a YAML cluster definition into a temporary Dask cluster on Hugging Face Jobs. It starts one coordinator Job that hosts the Dask scheduler and runs your script, then starts the worker Jobs that contribute CPU and GPU workers.

The application remains an ordinary Dask program. hfdask supplies the temporary Hugging Face infrastructure around it.

The machines connect through an authenticated, encrypted Iroh mesh. Dask services bind to loopback rather than exposing a public scheduler. Each machine starts one single-threaded worker for every complete CPU core available to it. Visible NVIDIA GPUs are assigned exclusively to the first worker processes, one GPU per process, and those workers advertise Dask's standard GPU resource.

The workload itself does not need an hfdask-specific wrapper. It gets the existing Dask client with distributed.get_client(), creates an ordinary task graph, and uses normal Dask worker affinity and resources for placement. hfdask owns the surrounding lifecycle: source shipping, locked dependency installation, Job submission, cluster formation, cleanup, and recovery metadata.

Why Synthetic Data on Hugging Face Jobs?

Synthetic data generation is often presented as a loop around a model API. Production pipelines quickly become more complicated:

Running every stage in one GPU process wastes expensive accelerator time on CPU work. Splitting the pipeline into separate scripts avoids that waste, but introduces another orchestration system and forces every intermediate boundary into storage.

Dask lets us keep the whole computation in one graph. Dependencies between stages stay explicit, CPU work can run concurrently, and generation tasks can request GPU workers with resources={"GPU": 1}. Hugging Face Jobs supplies the heterogeneous machines and native access to Hub resources. hfdask connects the two for the lifetime of one program.

This pattern is not limited to question-and-answer generation. The same shape appears in synthetic instruction generation, classification rationales, document augmentation, model distillation, preference-pair construction, and offline evaluation.

Pipeline Components

The synthetic question-and-answer pipeline in this walkthrough has six components:

1. Source dataset: Documents from which grounded examples will be generated.
2. Generation model: The language model that creates each example.
3. Prompt and output schema: The contract the model must follow.
4. Cluster configuration: The CPU and GPU Jobs assigned to each stage.
5. Validation: Checks that turn model output into auditable records.
6. Dask graph: The complete CPU → GPU → CPU computation.

Let's take a closer look at each component.

Source Dataset

For this walkthrough, we use the test split of fancyzhx/ag_news. Each row contains a news article and one of four topic labels. We use the article text as source material and preserve its row identifier and original label as provenance.

The dataset is mounted directly into every Job at a pinned revision:

mounts:
  - source: hf://datasets/fancyzhx/ag_news
    revision: eb185aade064a813bc0b7f42de02595523103ca4
    target: /dataset

The pipeline reads the mounted Parquet file with pandas, takes a deterministic balanced sample, and divides it into batches. This is CPU work:

from pathlib import Path

import pandas as pd
from distributed import get_worker

DATASET = Path("/dataset/data/test-00000-of-00001.parquet")
CATEGORIES = ("World", "Sports", "Business", "Sci/Tech")


def prepare_source(path: Path) -> list[pd.DataFrame]:
    # Read only the two columns we need from the mounted Parquet file.
    rows = pd.read_parquet(path, columns=["text", "label"])
    # Keep the original row index as a stable source identifier for provenance.
    rows.insert(0, "source_id", rows.index.astype("int64"))
    # Take a deterministic, balanced sample so every category is represented equally.
    rows = rows.groupby("label", sort=True).sample(n=32, random_state=23)
    # Map the integer labels to their human-readable category names.
    rows["label"] = rows["label"].map(dict(enumerate(CATEGORIES))).astype("str")
    # Restore a predictable row order after sampling.
    rows = rows.sort_values("source_id").reset_index(drop=True)
    # Emit a structured event so the run is observable from the Dask scheduler.
    get_worker().log_event("synthetic-data", {"phase": "prepared", "rows": len(rows)})
    # Split into fixed-size partitions, one Dask task per partition.
    return [rows.iloc[start : start + 16] for start in range(0, len(rows), 16)]

A small sample keeps the first run understandable and inexpensive. A production run would read many Parquet fragments and create enough partitions to keep every GPU worker fed.

Generation Model

For this walkthrough we use Qwen/Qwen3-0.6B, the model from the repository's GPU example. It is small enough to load on a single l4x1 Job, which keeps the demo cheap and fast. It is not a recommendation for real synthetic data generation, where a larger generator will usually produce better examples. The model is pinned to a specific Hub revision and mounted at /model:

mounts:
  - source: hf://models/Qwen/Qwen3-0.6B
    revision: c1899de289a04d12100db370d81485cdf75e47ca
    target: /model

Loading a model for every Dask task would dominate the run. Instead, a Dask WorkerPlugin initializes one vLLM engine when each GPU worker starts:

import time

from distributed import Worker, WorkerPlugin
from vllm import LLM, SamplingParams

MODEL = "/model"


class GeneratorSetup(WorkerPlugin):
    def setup(self, worker: Worker) -> None:
        if not worker.state.total_resources.get("GPU", 0):
            return

        started = time.monotonic()
        worker.log_event("synthetic-data", {"phase": "model_loading"})
        worker.generator = LLM(
            model=MODEL,
            tensor_parallel_size=1,
            max_model_len=2048,
            max_num_seqs=16,
            gpu_memory_utilization=0.5,
            enforce_eager=True,
            trust_remote_code=False,
            seed=23,
        )
        worker.sampling_params = SamplingParams(
            temperature=0.2,
            top_p=0.9,
            max_tokens=256,
            seed=23,
        )
        worker.log_event(
            "synthetic-data",
            {
                "phase": "model_loaded",
                "elapsed": time.monotonic() - started,
            },
        )

GPU discovery and process placement happen before the plugin runs. CPU-only workers skip model initialization, while every GPU worker keeps its engine available for repeated generation tasks.

Prompt and Output Schema

We ask the model to generate one answerable question from each source article. The response must contain a question, a concise answer, and an exact supporting quote:

SYSTEM_PROMPT = """You create grounded question-and-answer training examples.
Return exactly one JSON object with these string fields:
- question: a clear question answerable from the source article
- answer: a concise answer supported by the article
- supporting_quote: an exact, contiguous quote from the article that supports the answer

Do not use facts that are absent from the article. Do not wrap the JSON in Markdown.
"""


def prompt_for(text: str) -> str:
    return f"{SYSTEM_PROMPT}\n\nSOURCE ARTICLE:\n{text}"

We parse the output into a strict Pydantic model:

from pydantic import BaseModel, ConfigDict, field_validator


class GeneratedExample(BaseModel):
    model_config = ConfigDict(extra="forbid")

    question: str
    answer: str
    supporting_quote: str

    @field_validator("question", "answer", "supporting_quote")
    @classmethod
    def require_text(cls, value: str) -> str:
        value = value.strip()
        if not value:
            raise ValueError("value must not be empty")
        return value

Strict parsing is intentional. If the model adds unknown fields, omits a field, or returns another shape, the record belongs in the rejected output rather than being silently repaired into an apparently valid training example.

Cluster Configuration

Create a Git-backed uv project and install the workload libraries as normal dependencies. hfdask belongs in a deployment group because it is infrastructure used to run the project:

mkdir hfdask-synthetic-data
cd hfdask-synthetic-data
git init
uv init --bare --python 3.12
uv python pin 3.12
uv add dask distributed pandas pyarrow pydantic \
  "vllm==0.29.0; sys_platform == 'linux' and platform_machine == 'x86_64'"
uv add --group deploy "hfdask==0.1.1"

The cluster uses a CPU coordinator and one l4x1 GPU worker Job:

namespace: your-namespace

coordinator:
  flavor: cpu-basic
  worker: true

workers:
  flavor: l4x1
  count: 1

environment:
  image: vllm/vllm-openai:v0.29.0@sha256:082ca6f035279109041ffd3fe0695cb568b29bc580b35c4f297a66a08b216c1b
  groups: [deploy]

mounts:
  - source: hf://models/Qwen/Qwen3-0.6B
    revision: c1899de289a04d12100db370d81485cdf75e47ca
    target: /model
  - source: hf://datasets/fancyzhx/ag_news
    revision: eb185aade064a813bc0b7f42de02595523103ca4
    target: /dataset
  - source: hf://buckets/your-namespace/synthetic-data/ag-news-qa-run-001
    target: /output
    read_only: false

timeout: 30m

network:
  public_relays: true

The coordinator hosts the script and Dask scheduler. With worker: true, it reserves one complete CPU core for the scheduler and starts CPU workers on the remaining cores. The l4x1 Job starts one GPU-assigned worker and CPU-only workers on its remaining complete cores.

The image is pinned by digest and provides CUDA tooling, Python, and uv. hfdask runs its own bootstrap rather than the image's serving entrypoint. The project lockfile installs vLLM 0.29.0 on Linux, so the environment and image stay on the same vLLM release.

network.public_relays: true is explicit consent to public discovery and relay fallback. The data remains encrypted in transit, but discovery and relay services can observe connection metadata.

Before running, create a private output bucket and seed a fresh prefix for this run. Reusing an output prefix makes retries and provenance harder to reason about.

Validation

Model output that parses successfully is not automatically good training data. Our baseline validator applies checks that are deterministic and easy to audit:

1. The output must be valid JSON.
2. It must match the exact Pydantic schema.
3. Every text field must be non-empty.
4. The question and answer must stay within configured length bounds.
5. The supporting quote must appear exactly in the source article.
6. Duplicate normalized questions are removed before writing.

import json

import pandas as pd
from pydantic import ValidationError


def validate_partition(rows: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
    accepted: list[dict] = []
    rejected: list[dict] = []

    for row in rows.to_dict(orient="records"):
        try:
            parsed = GeneratedExample.model_validate(json.loads(row["raw_generation"]))
            if len(parsed.question) > 240:
                raise ValueError("question_too_long")
            if len(parsed.answer) > 1_000:
                raise ValueError("answer_too_long")
            if parsed.supporting_quote not in row["text"]:
                raise ValueError("supporting_quote_not_found")
        except (json.JSONDecodeError, ValidationError, ValueError) as error:
            rejected.append({**row, "rejection_reason": str(error)})
            continue

        accepted.append(
            {
                "source_id": row["source_id"],
                "source_text": row["text"],
                "source_label": row["label"],
                **parsed.model_dump(),
                "model_repo": "Qwen/Qwen3-0.6B",
                "model_revision": "c1899de289a04d12100db370d81485cdf75e47ca",
            }
        )

    return pd.DataFrame(accepted), pd.DataFrame(rejected)

These checks establish a minimum contract, not a final quality bar. Before using the result for training, we would also inspect a statistically meaningful sample and evaluate whether the questions are diverse, relevant, unambiguous, and useful for the target task. More demanding pipelines can add model-based judging, cross-model verification, personally identifiable information filtering, contamination checks, and domain-specific rules as separate Dask stages.

Rejected records are valuable evidence. We preserve the raw generation and rejection reason so we can improve the prompt and validation policy instead of guessing why yield changed.

Complete Pipeline

The generation task receives one pandas partition and reuses the vLLM engine attached to its GPU worker:

import time

import pandas as pd
from distributed import get_worker


def generate_partition(rows: pd.DataFrame) -> pd.DataFrame:
    worker = get_worker()
    engine = worker.generator
    tokenizer = engine.get_tokenizer()

    prompts = [
        tokenizer.apply_chat_template(
            [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": f"SOURCE ARTICLE:\n{text}"},
            ],
            enable_thinking=False,
            add_generation_prompt=True,
            tokenize=False,
        )
        for text in rows["text"]
    ]

    started = time.monotonic()
    outputs = engine.generate(
        prompts,
        worker.sampling_params,
        use_tqdm=False,
    )

    result = rows.copy()
    result["raw_generation"] = [
        output.outputs[0].text.strip() if output.outputs else ""
        for output in outputs
    ]
    worker.log_event(
        "synthetic-data",
        {
            "phase": "generation_complete",
            "rows": len(result),
            "elapsed": time.monotonic() - started,
        },
    )
    return result

The final CPU stage deduplicates accepted questions and writes accepted rows, rejected rows, and a machine-readable summary:

import json
from pathlib import Path

import pandas as pd
from distributed import get_worker

OUTPUT = Path("/output")


def write_dataset(
    accepted_parts: list[pd.DataFrame],
    rejected_parts: list[pd.DataFrame],
    output: Path,
) -> dict[str, int | float]:
    output.mkdir(parents=True, exist_ok=True)

    accepted = pd.concat(accepted_parts, ignore_index=True)
    rejected = pd.concat(rejected_parts, ignore_index=True)
    attempted = len(accepted) + len(rejected)

    if not accepted.empty:
        accepted["normalized_question"] = (
            accepted["question"].str.lower().str.replace(r"\s+", " ", regex=True).str.strip()
        )
        before_deduplication = len(accepted)
        accepted = accepted.drop_duplicates("normalized_question").drop(
            columns="normalized_question"
        )
    else:
        before_deduplication = 0

    duplicates = before_deduplication - len(accepted)
    accepted.to_parquet(output / "accepted.parquet", index=False)
    rejected.to_parquet(output / "rejected.parquet", index=False)

    summary = {
        "attempted": attempted,
        "accepted": len(accepted),
        "rejected": len(rejected),
        "duplicates": duplicates,
        "validation_yield": len(accepted) / attempted if attempted else 0.0,
    }
    (output / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
    get_worker().log_event("synthetic-data", {"phase": "saved", **summary})
    return summary

The main function assembles the graph and places each stage on suitable workers:

import json

import dask
from distributed import get_client


def main() -> None:
    client = get_client()
    workers = client.scheduler_info()["workers"]
    cpu_workers = [
        address
        for address, info in workers.items()
        if not info["resources"].get("GPU", 0)
    ]
    if not cpu_workers or len(cpu_workers) == len(workers):
        raise ValueError("This pipeline requires both CPU-only and GPU workers")

    client.register_plugin(GeneratorSetup(), name="synthetic-generator")

    with dask.annotate(workers=cpu_workers, allow_other_workers=False):
        source_parts = list(dask.delayed(prepare_source, nout=8)(DATASET))

    with dask.annotate(resources={"GPU": 1}):
        generated_parts = [
            dask.delayed(generate_partition)(part)
            for part in source_parts
        ]

    with dask.annotate(workers=cpu_workers, allow_other_workers=False):
        validated_parts = [
            dask.delayed(validate_partition, nout=2)(part)
            for part in generated_parts
        ]
        accepted_parts = [part[0] for part in validated_parts]
        rejected_parts = [part[1] for part in validated_parts]
        summary = dask.delayed(write_dataset)(
            accepted_parts,
            rejected_parts,
            OUTPUT,
        )

    print(json.dumps(summary.compute(optimize_graph=False)), flush=True)


if __name__ == "__main__":
    main()

optimize_graph=False matters here because Dask task fusion could otherwise collapse separately annotated CPU and GPU stages. We want those placement boundaries to remain explicit.

Running the Pipeline

Authenticate the submitting machine, set your HF namespace in cluster.yaml, review the paid hardware and timeout, and run:

uv run --group deploy hf auth login
uv run --group deploy hfdask run \
  --cluster cluster.yaml \
  generate_dataset.py

After that command, hfdask performs the cluster lifecycle:

1. It validates the cluster configuration locally.
2. It snapshots the eligible files in the current Git working tree.
3. It uploads the source archive under a unique prefix in the namespace's private jobs-artifacts bucket.
4. It submits the coordinator and worker Jobs.
5. Each Job verifies and extracts the source, then runs uv sync --locked --no-dev --group deploy.
6. The machines discover their fixed cluster peers and establish encrypted Iroh tunnels.
7. The coordinator starts the scheduler, waits for the expected workers, and executes generate_dataset.py.
8. The workload writes accepted and rejected rows to the mounted output bucket.
9. When the script finishes, hfdask cancels every known Job and verifies that each reached a terminal state.

The submitting machine does not need Linux-only workload dependencies such as vLLM. They remain normal project dependencies with platform markers and are installed from the same lockfile inside the Linux Jobs.

An isolated uvx hfdask invocation cannot replace the deployment group. It gives the submitting process a CLI, but does not put hfdask's remote runner in the project's lockfile for installation on the Jobs.

Scaling Generation

The graph does not change as the run grows. To go faster:

• Feed the GPUs. Size partitions so vLLM can batch, and create more partitions than GPU workers so no accelerator waits on CPU work.
• Add machines, not model parallelism. Raising workers.count adds independent GPU workers, each with its own model replica. Spreading one model across several GPUs needs a different image and launch strategy.
• Keep CPU stages on CPU workers. Preparation, validation, deduplication, and writes should not occupy the process that owns the accelerator.
• Measure accepted examples per dollar. Raw throughput is misleading if a faster configuration produces more malformed or unsupported rows.
• Write recoverable shards. Job disks and scheduler state are ephemeral, so long runs should write deterministic, independently recoverable shards and skip committed ones on retry.

What hfdask Handles

hfdask deliberately owns deployment mechanics rather than synthetic data semantics.

That boundary is important. hfdask makes the same Dask program portable to temporary HF Jobs infrastructure. It does not decide what constitutes a good synthetic example.

Conclusion

We built hfdask because the infrastructure we wanted was already close at hand. Hugging Face Jobs had the CPUs, GPUs, models, datasets, and storage. Dask had the programming model for coordinating heterogeneous work. The missing piece was a reliable cluster lifecycle between them.

Synthetic training data generation makes that combination concrete. Source preparation belongs on CPUs. Batched model generation belongs on GPUs. Validation, deduplication, and durable writes belong back on CPUs. With hfdask, those stages stay in one ordinary Dask graph and run across temporary Hugging Face Jobs from one YAML definition.

The library is open source, available on PyPI, and ready to try:

uv add --group deploy hfdask

Additional Resources

• hfdask on GitHub
• hfdask on PyPI
• hfdask CPU quickstart
• hfdask heterogeneous inference example
• Hugging Face Jobs documentation
• Dask task annotations
• vLLM documentation