Running Jobs

Running Jobs

run()

job.run(
    data,                    # pandas DataFrame, list of dicts, or Spark DataFrame
    backend="local",         # "local" | "multiprocessing" | "spark" (keyword-only from here)
    output_col="output",     # adds output, output_error, output_meta
    batch_size=32,           # rows per engine call
    workers=None,            # multiprocessing only
    partitions=None,         # spark only: repartition the input first
    dedupe=False,            # send identical inputs once
    checkpoint=None,         # directory for persisted results ...
    id_col=None,             # ... keyed by this unique column
    retry_failed=True,       # with a checkpoint: re-run rows that failed before
    checkpoint_every=None,   # flush size / Spark wave size
)

parameter	meaning
data	pandas DataFrame or list of dicts (local, multiprocessing) → same type back. Spark DataFrame (spark) → Spark DataFrame.
backend	where rows are processed. See Backends.
output_col	name of the value column. Errors go to <output_col>_error, metrics to <output_col>_meta. The three names must not exist in the input.
batch_size	rows handed to the engine per call (and per worker task in multiprocessing). Larger batches help GPU engines. API engines parallelize within a batch up to max_concurrency.
workers	number of worker processes. Default: one per GPU slot for local models, min(4, CPUs) for API models.
partitions	df.repartition(n) before processing. Controls how many Spark tasks run the model.
dedupe	rows with identical values in the job's input columns are processed once. See Checkpoints and Dedupe.
checkpoint	directory (local, /Volumes/..., dbfs:/...) where results are persisted while the run progresses. Requires id_col.
id_col	unique str/int id column for checkpoints.
retry_failed	with a checkpoint: rows stored with an error are processed again (default). False keeps them as they are.
checkpoint_every	local backends: flush every N rows (default 1 000). Spark: rows per wave (default 10 000).

Before any work starts, run() checks:

• the backend and its parameters;
• that every column the job reads exists;
• that the output columns don't exist yet;
• that an engine for the model and task is available.

Violations raise ConfigError.

The output keeps the input as it was: pandas index and row order, list order, Spark columns and types. Only the three output columns are added.

preview()

job.preview(data, n=5, output_col="output")

Runs the first n rows with the local backend in the current process. For a Spark DataFrame, the first n rows are collected to the driver and the result is a pandas DataFrame. Useful for checking prompts and outputs before a big run. The driver must be able to run the engine, e.g. reach the API or fit the model.

Lazy results on Spark

Without a checkpoint, run(backend="spark") returns a lazy DataFrame: inference happens when an action runs (write, count, collect, display, …), and again for every further action. Write the result once and read it back, or use a checkpoint, which makes the run eager and returns a cheap DataFrame over stored results.

out = job.run(df, backend="spark")
out.write.mode("overwrite").saveAsTable("main.ml.scored")   # one inference pass
scored = spark.table("main.ml.scored")

Choosing a backend

situation	backend
a few hundred rows, notebook exploration	local (or preview)
one machine with GPUs, data fits in pandas	multiprocessing
an API model, up to ~100k rows	local already runs max_concurrency requests at once
large tables, many GPUs or many API calls	spark