PureStream — clean a dataset, and explain every rejected row

Point it at an Apify Dataset. It cleans the rows the same way every time and writes the ones it accepts.

A check that rejects a row sets that row aside in its own dataset, the quarantine, with the reason. The six jobs below do not reject rows. Rows a dedupe or a filter removes are counted as dropped, and they are not stored. A summary keeps its result rows; its status line says how many were combined, and the report's dropped count is that same number.

It can stack datasets (union) and add columns from a second dataset, matched on a key you name (join).

Job runs one job from the form. A config file runs the operations you list, in that order.

To have an agent write the file, start here: Using this Actor from an agent.

One job per run

1. Source dataset. Leave it empty to run a three-row sample. A job that reads a second dataset still needs a source dataset: the sample is only one dataset.
2. Job. Pick one, or leave it empty to pass the rows through unchanged.
3. Fill that job's section. Fields in the other job sections are not used.
4. Run it.

Job	What it does	What you fill in
Dedupe by fields	One row per distinct value	Columns that mark a copy (at least one) and Copy to keep
Append datasets	Stack the source with more datasets	Additional datasets (at least one)
Join on keys	Add columns from one other dataset	Additional datasets (exactly one), Key columns (the shared keys), Join kind, Can a key repeat?
Filter rows	Keep the rows that match	Column to test (exactly one), Comparison, Value to compare with
Keep or drop columns	Keep the named columns, or drop them	Keep or drop and Columns to keep or drop (at least one)
Summarise	One row per group	Columns that set the group (empty means one row for the whole dataset), Calculation, Result column

A join is refused when Can a key repeat? is empty. The run checks that choice. Both sides may repeat a key is taken as stated. Each key must be a column of both datasets. If a key is a number in one and text in the other, turn on Compare keys as text. A column other than the key that both datasets have is renamed on the second dataset with Suffix for a column both datasets hold: url becomes url_right unless you change the suffix.

A filter is refused without a value. A null matches no comparison. A number is a plain number, with no separators. A true/false column takes true or false. Text is compared as text, so "10" comes before "9". A date or date-time column is not filtered from this form; state the format in a config file.

count of rows takes no column. Any other calculation needs Column to calculate on.

Every column is one you named. These six jobs do not reject rows. Rows a dedupe or a filter removes are counted as dropped, and they are not stored. A summary keeps its result rows; the status line says how many were combined, and rows.dropped is that same count. To check a value, parse it, or run more than one step, use a config file.

Config file

Paste JSON into Config file, or give an https URL in Config file URL. Use one, not both. When either is set, Job is ignored. Source dataset is still required, and the file's source.dataset_id must name that same dataset. Every other dataset the file reads must be picked under Additional datasets. A dataset named only in the file cannot be read. The file's source has to be an Apify dataset.

A config file names its own quarantine. Quarantine dataset on the form applies to a job, which does not reject rows.

{
    "version": "1",
    "source": { "format": "apify_dataset", "dataset_id": "YOUR_DATASET_ID" },
    "operations": [
        { "id": "trim-name", "type": "trim", "column": "name" },
        {
            "id": "check-email",
            "type": "validate_email",
            "column": "email",
            "on_error": "reject_row"
        }
    ],
    "sink": { "format": "apify_dataset", "dataset_name": null },
    "quarantine": {
        "format": "apify_dataset",
        "dataset_name": "YOUR_QUARANTINE_NAME"
    }
}

Reusing a dataset name appends this run's rejected rows onto whatever that dataset already holds.

The operation reference and the JSON Schema are the format.

An operation that can reject rows needs a quarantine destination. deduplicate, filter, limit, drop_nulls, group_by, union, join, and explode change how many rows there are, so none of them may sit between two operations that can reject rows. Put them after the last operation that can reject.

Before a full run

Dry run writes nothing. It reads every input and reports the cleaning, the columns, and every configuration error. It does not apply the spending limit or the size check. The real run does.

Preview rows (1 to 1,000) runs the real operations on that many items and writes nothing. The columns in that sample can differ from the whole dataset. Do not set it together with dry run.

Results

Cleaned rows	this run's default dataset
Rejected rows	the dataset named in the config file's quarantine. The report names it outputs.quarantined.dataset. That dataset is created only when a row was rejected. A job does not reject rows, so it does not create this dataset; rows it removes are counted as rows.dropped and are not stored. A summary's result rows are the cleaned dataset; rows.dropped is how many were combined.
Run report	the key-value store, under OUTPUT

OUTPUT.reconciles is false when the counts do not add up. OUTPUT.outcome is completed, completed_with_rejects, or unreconciled. A run that did not finish is failed.

With Redact values in error messages on, which is the default, a rejection names the field and not the value. The rejected row still carries the value in its column. __errors.original_value is null.