# PureStream — Turn Messy Data into Trusted Insights with Clarity (`ahmedmohamed/purestream`) Actor

Consistently clean, validate, and reshape an Apify Dataset. You can join, stack, or summarize datasets. Parse numbers and dates, validate email fields, and flatten nested JSON. Accepted rows will be moved to a new dataset, while rejected rows will go to quarantine with the reason for rejection noted

- **URL**: https://apify.com/ahmedmohamed/purestream.md
- **Developed by:** [Ahmed Alzoghaly](https://apify.com/ahmedmohamed) (community)
- **Stats:** 2 total users, 1 monthly users, 100.0% runs succeeded, 0 bookmarks
- **User rating**: No ratings yet

## Pricing

from $0.30 / 1,000 accepted rows

This Actor is paid per event. You are not charged for the Apify platform usage, but only a fixed price for specific events.

Learn more: https://docs.apify.com/actors/running/actors-in-store.md#pay-per-event

## What's an Apify Actor?

An Actor is a serverless cloud program that runs on the Apify platform. It has two run modes.
In Batch mode, an Actor accepts a well-defined JSON input, performs an action which can take anything from a few seconds to a few hours,
and optionally produces a well-defined JSON output, datasets with results, or files in key-value store.
In Standby mode, an Actor provides a web server which can be used as a website, API, or an MCP server.

Apify vocabulary and the platform model are defined once, in the agent quickstart at https://apify.com/agents.md.

## How to integrate an Actor?

If asked about integration, you help developers integrate Actors into their projects.
You adapt to their stack and deliver integrations that are safe, well-documented, and production-ready.

Do not guess an integration path. Every one of them is in the agent quickstart at https://apify.com/agents.md: the Apify MCP server, Agent Skills with the Apify CLI, the JavaScript and Python clients, the REST API, and the account-free path for an agent with no human to sign in. It also carries the rule on stating cost before the first paid run.

For examples already wired to this Actor's own input schema, see the [API](#api) section below.

Each client library has reference documentation the quickstart does not restate: [JavaScript/TypeScript](https://docs.apify.com/api/client/js/docs.md) (`npm install apify-client`) and [Python](https://docs.apify.com/api/client/python/docs.md) (`pip install apify-client`).

# README

## 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](https://api.apify.com/v2/key-value-stores/AhmedMohamed~purestream-contract/records/using-from-an-agent.md).

### 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.

```json
{
    "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](https://api.apify.com/v2/key-value-stores/AhmedMohamed~purestream-contract/records/pipeline-document.md)
and the [JSON Schema](https://api.apify.com/v2/key-value-stores/AhmedMohamed~purestream-contract/records/pipeline-document.schema.json)
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.

# Actor input Schema

## `sourceDatasetId` (type: `string`):

The dataset to clean. Picking it here is what grants this run read access, so a dataset named only inside a config file cannot be read. Leave empty to run a tiny built-in sample (what Apify Daily uses). A config file still needs this field, matching the file's source.dataset_id. Read-only: the Actor never writes to your source.

## `job` (type: `string`):

One job for this run. Leave empty to pass the dataset through. A config file below ignores this.

## `additionalDatasetIds` (type: `array`):

Used when Job is Append datasets (every picked dataset) or Join on keys (exactly one), and by a config file's inputs. Picking them here is what grants this run read access — a dataset named only inside a config file cannot be read.

## `pipeline` (type: `object`):

A config file for the whole cleaning, pasted as JSON. When set, Job is ignored, but Source dataset is still required, and the file's source.dataset_id must name that same dataset. See the operation reference for the format, or paste the prompt from the README into an LLM and describe your data. Use this or Config file URL, not both.

## `pipelineUrl` (type: `string`):

An https:// URL of the same config file. Use this or Config file, not both.

## `jobColumns` (type: `array`):

Accepted from a saved task or the API. The form shows a Columns field inside each job instead, and that field is the one a new run reads.

## `dedupeColumns` (type: `array`):

The columns that decide whether two rows are copies.

## `dedupeKeep` (type: `string`):

Used when Job is Dedupe by fields. Which copy to keep.

## `keepDropColumns` (type: `array`):

The columns to keep, or to drop, as chosen below.

## `columnsMode` (type: `string`):

Used when Job is Keep or drop columns.

## `filterColumns` (type: `array`):

The one column to compare. This job takes exactly one name.

## `filterOp` (type: `string`):

Used when Job is Filter rows. A text column compares as text, letter by letter: "10" sorts before "9", and letters after digits.

## `filterValue` (type: `string`):

Used when Job is Filter rows, and required then: a blank value is refused before anything is read. A number column takes a plain number, with no separators. A true or false column takes true or false. A date column is not filtered from here.

## `summariseColumns` (type: `array`):

The columns that define a group. Leave empty for one row covering the whole dataset.

## `aggregate` (type: `string`):

Used when Job is Summarise.

## `aggregateColumn` (type: `string`):

Used when Job is Summarise and the calculation is not count of rows. The column the calculation reads.

## `aggregateInto` (type: `string`):

Used when Job is Summarise. Name of the column the calculation is written to.

## `joinColumns` (type: `array`):

The key names shared by both datasets.

## `joinHow` (type: `string`):

Used when Job is Join on keys.

## `joinCardinality` (type: `string`):

Used when Job is Join on keys. Required for that job. The run refuses a join that omits it. The run checks what you choose, except Both sides may repeat a key, which is taken as stated.

## `joinSuffix` (type: `string`):

Used when Job is Join on keys. When both datasets hold a column besides the keys, the second dataset's copy is renamed with this suffix: url beside url_right.

## `joinKeysAsText` (type: `boolean`):

Used when Job is Join on keys. On, both datasets' keys are compared as text, so a key held as a number in one (5) matches the same digits held as text in the other ("5"), and the key is written as text. Off, a key must hold the same type in both.

## `dryRun` (type: `boolean`):

Check without writing. It reports the cleaning that would run, the columns you would get, and every configuration error at once. It writes nothing, but it does read every input the run would read, not just Source dataset: Append datasets and Join on keys name more datasets under Additional datasets, and a config file names its own inputs. Those are read in full, the same way the real run does, so the cost grows with how many inputs the config file declares. It reports what the real run would be charged and what the size check would decide, and refuses on neither: the real run checks the charge against your spending limit before it reads anything, and the size once its input is downloaded.

## `previewRows` (type: `integer`):

Try the cleaning on a sample. It reads at most this many items, runs the real operations over them, writes nothing, and returns the first rows it would accept and the first it would reject. This is a sample, not a dry run. The columns of a sample can differ from the columns of the whole dataset, because a name or a value that only appears later cannot be seen here. Leave empty for a normal run.

## `quarantineDatasetName` (type: `string`):

The six jobs do not reject rows, so a job run does not create this dataset. A config file names its quarantine inside the file.

## `engine` (type: `string`):

How the run executes. Streaming keeps memory bounded and is the default. In-memory is faster but holds the whole dataset at once. Some operations cannot stream, and with a memory ceiling set those are refused by name before the run starts rather than left to run out of memory.

## `redactValues` (type: `boolean`):

On by default. A rejection names the field, never the value, so a run's logs never carry your data.

## `releaseLeafInputs` (type: `boolean`):

A diagnostic switch for memory measurement, not an ordinary run option — leave it on. On (default), a combined dataset's memory is freed as soon as the last operation that reads it finishes; off holds every input live for the whole run, which is only useful for comparing peak memory between the two.

## `refuseOversized` (type: `boolean`):

A diagnostic switch for memory measurement, not an ordinary run option — leave it on. On (default), a run too large for the container is refused: before any item is read when the datasets' stored sizes alone show it cannot fit, and otherwise once its input has been downloaded and measured, before it is loaded into memory. Off lets it run anyway, which is only useful for measuring where the limit really is.

## Actor input object example

```json
{
  "dedupeKeep": "first",
  "columnsMode": "keep",
  "filterOp": "eq",
  "aggregate": "count",
  "aggregateInto": "count",
  "joinHow": "left",
  "joinSuffix": "_right",
  "joinKeysAsText": false,
  "dryRun": false,
  "quarantineDatasetName": "purestream-quarantine",
  "engine": "streaming",
  "redactValues": true,
  "releaseLeafInputs": true,
  "refuseOversized": true
}
```

# Actor output Schema

## `cleaned` (type: `string`):

The accepted rows. Rejected rows are never here. A check that rejects a row writes it to the quarantine dataset named in OUTPUT as outputs.quarantined.dataset. That name is present when the run had a quarantine sink, and outputs.quarantined.rows is how many rows were rejected. The dataset is created only when a row was rejected. A quick job that removes rows does not store them; the report counts those as dropped. A summary keeps its result rows; the report's dropped count is how many source rows were combined.

## `report` (type: `string`):

OUTPUT: counted rows, whether they reconcile, the columns ingest had to widen, the predicted schema, and the document that ran.

# API

You can run this Actor programmatically using our API. Below are code examples in JavaScript, Python, and CLI, as well as the OpenAPI specification and MCP server setup.

## JavaScript example

```javascript
import { ApifyClient } from 'apify-client';

// Initialize the ApifyClient with your Apify API token
// Replace the '<YOUR_API_TOKEN>' with your token
const client = new ApifyClient({
    token: '<YOUR_API_TOKEN>',
});

// Prepare Actor input
const input = {
    "dedupeKeep": "first",
    "columnsMode": "keep",
    "filterOp": "eq",
    "aggregate": "count",
    "aggregateInto": "count",
    "joinHow": "left",
    "joinSuffix": "_right",
    "joinKeysAsText": false,
    "dryRun": false,
    "quarantineDatasetName": "purestream-quarantine",
    "engine": "streaming",
    "redactValues": true,
    "releaseLeafInputs": true,
    "refuseOversized": true
};

// Run the Actor and wait for it to finish
const run = await client.actor("ahmedmohamed/purestream").call(input);

// Fetch and print Actor results from the run's dataset (if any)
console.log('Results from dataset');
console.log(`💾 Check your data here: https://console.apify.com/storage/datasets/${run.defaultDatasetId}`);
const { items } = await client.dataset(run.defaultDatasetId).listItems();
items.forEach((item) => {
    console.dir(item);
});

// 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/js/docs

```

## Python example

```python
from apify_client import ApifyClient

# Initialize the ApifyClient with your Apify API token
# Replace '<YOUR_API_TOKEN>' with your token.
client = ApifyClient("<YOUR_API_TOKEN>")

# Prepare the Actor input
run_input = {
    "dedupeKeep": "first",
    "columnsMode": "keep",
    "filterOp": "eq",
    "aggregate": "count",
    "aggregateInto": "count",
    "joinHow": "left",
    "joinSuffix": "_right",
    "joinKeysAsText": False,
    "dryRun": False,
    "quarantineDatasetName": "purestream-quarantine",
    "engine": "streaming",
    "redactValues": True,
    "releaseLeafInputs": True,
    "refuseOversized": True,
}

# Run the Actor and wait for it to finish
run = client.actor("ahmedmohamed/purestream").call(run_input=run_input)

# Fetch and print Actor results from the run's dataset (if there are any)
print(f"💾 Check your data here: https://console.apify.com/storage/datasets/{run.default_dataset_id}")
for item in client.dataset(run.default_dataset_id).iterate_items():
    print(item)

# 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/python/docs/quick-start

```

## CLI example

```bash
echo '{
  "dedupeKeep": "first",
  "columnsMode": "keep",
  "filterOp": "eq",
  "aggregate": "count",
  "aggregateInto": "count",
  "joinHow": "left",
  "joinSuffix": "_right",
  "joinKeysAsText": false,
  "dryRun": false,
  "quarantineDatasetName": "purestream-quarantine",
  "engine": "streaming",
  "redactValues": true,
  "releaseLeafInputs": true,
  "refuseOversized": true
}' |
apify call ahmedmohamed/purestream --silent --output-dataset

```

## MCP server setup

```json
{
    "mcpServers": {
        "apify": {
            "type": "http",
            "url": "https://mcp.apify.com/?tools=fetch-actor-details,ahmedmohamed/purestream"
        }
    }
}
```

The hosted server signs you in with OAuth on first connect, so no API token belongs in this config. Clients without OAuth support can send an `Authorization: Bearer <APIFY_API_TOKEN>` header instead, using a token from API & Integrations in Apify Console (https://console.apify.com/settings/integrations).

## OpenAPI specification

Download the OpenAPI definition: https://api.apify.com/v2/actors/azkq3hs2pKiy6jBp1/builds/5vhy3M5ccSEXZIA5J/openapi.json