# Silverset | SQL, REST & Sheets to Analysis-Ready Data (`ahmedmohamed/silverset`) Actor Read SQL, REST APIs, Google Sheets, CSV/JSON files, a dataset, or pasted records. Raw rows land in Bronze, composable stages clean them, and analysis-ready Silver rows go to a Dataset or key-value store. Bad rows are quarantined, never dropped. Start with the Analysis-ready preset. - **URL**: https://apify.com/ahmedmohamed/silverset.md - **Developed by:** [Ahmed Alzoghaly](https://apify.com/ahmedmohamed) (community) - **Categories:** Automation, Developer tools, Agents - **Stats:** 2 total users, 1 monthly users, 0.0% runs succeeded, 0 bookmarks - **User rating**: No ratings yet ## Pricing from $0.40 / 1,000 silver rows This Actor is paid per event and usage. You are charged both the fixed price for specific events and for Apify platform usage. Learn more: https://docs.apify.com/actors/running/actors-in-store.md#pay-per-event ## What's an Apify Actor? Actors are web data automations that power AI and operations. They run on the Apify platform to scrape websites, process data, connect APIs, and automate workflows. 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. Actors are written with capital "A". ## 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. The best way to integrate Actors is as follows. - **AI agents and MCP clients** — the [Apify MCP server](https://docs.apify.com/integrations/mcp.md) at `https://mcp.apify.com` (remote, streamable HTTP, OAuth on first use). - **Agentic workflows and local Actor development** — [Agent Skills](https://apify.com/.well-known/agent-skills/index.json) with the [Apify CLI](https://docs.apify.com/cli/docs.md): `npm install -g apify-cli`, then `apify login`. - **JavaScript/TypeScript projects** — the official [JS/TS client](https://docs.apify.com/api/client/js/docs.md): `npm install apify-client`. - **Python projects** — the official [Python client](https://docs.apify.com/api/client/python/docs.md): `pip install apify-client`. - **Any other language** — the [REST API](https://docs.apify.com/api/v2.md). For usage examples, see the [API](#api) section below. For more details, see Apify documentation as [Markdown index](https://docs.apify.com/llms.txt) and [Markdown full-text](https://docs.apify.com/llms-full.txt). # README ### What does Silverset do? Silverset reads your tables, APIs, sheets, files, or pasted JSON, keeps every raw row in **Bronze**, runs named stages, and writes **Silver** rows that passed. A stage **reject** goes to **quarantine**. Missing-drop and dedup **remove** rows without a quarantine item (counted as `dropped` on MANIFEST). Fail fast writes every row the failing batch rejected to quarantine, then aborts (rows written in earlier batches stay written; any trailing flush is batch-local). Bronze is always raw, including when pii redacts Silver and quarantine. It is not a scraper and not a warehouse. You point it at data you already have. Delivery is **at-least-once with idempotency**, not exactly-once: a retried run can append a second copy of the same row. Silver lands on an Apify Dataset (or key-value store). ### Why use Silverset? - **Nine sources, one pipeline.** SQL, REST, Google Sheets, CSV/JSON files, another Actor's dataset, a key-value record, or paste. - **Raw is never thrown away.** Bronze is the replay source. Silver is what you analyse. Quarantine is rejects; missing-drop and dedup are counted as `dropped` on MANIFEST. - **Start without writing a pipeline.** The Console default is **Analysis-ready**: profile the data, change no values, do not coerce CSV strings to numbers. - **Schedule it, call it from n8n or the API, export CSV/Excel** — same Actor, same stores. See the **Input** tab for every field. This page is the walkthrough. ### How to run Silverset 1. Open the Actor and leave **Pipeline preset** on **Analysis-ready** (under Advanced). 2. Pick a **Recipe**. 3. Fill **Where** — the field's help text says what to type for that recipe (for PostgreSQL/MySQL that is `user@host/database`, not a driver URL). 4. If the recipe needs a password or token, put it in **Secret**. Leave Secret empty for files, datasets, paste, and SQLite. 5. Click **Start**. Wait until the run succeeds. 6. Open the **default dataset** — those are your Silver rows. The run report is key-value store `SUMMARY`, not a dataset trailer. 7. Open the default **key-value store** → `PROFILE` for the schema/profile reports, `MANIFEST` for counts (including `dropped`) and charges, `SUMMARY` for the run report. A rejected input (exit 2) is not billed. Tick **Dry run** (Advanced) to check config and list tables with no writes and no charges. The Console form is recipe + Where + Secret. n8n and the API keep sending `source_type` + `source_config` as before — that path ignores Recipe. Everything below is a complete API/n8n payload. #### Notes before your first run - **Console: Recipe selects the connector; Where tells it where to look.** You do not paste JSON for a first run. - **API / n8n: only `source_config` changes between sources.** `source_type` selects the connector; the object tells that one connector where to look. Keys from another source are rejected, not ignored — `file_url` will not accept `tables`. - **Credentials never go in JSON.** Secret (Console) and the hidden credential fields (API) are marked secret, so Apify encrypts them. The same value typed into `source_config` is stored in plain text in the run's INPUT record. - **Destination defaults to this run's Dataset.** Open Advanced to write Silver to a key-value store instead. - **Leave the rest of Advanced alone** until a run tells you otherwise. Its defaults are the documented behaviour. ### Sources **On the Console**, pick a Recipe and fill Where: | Recipe | Where | Secret | | -------------- | ----------------------------------------------------- | ----------------------- | | PostgreSQL | `user@host/database` (optional `:port`, default 5432) | database password | | MySQL | `user@host/database` (optional `:port`, default 3306) | database password | | SQLite file | `data.db` or `:memory:` | — | | File over HTTP | `https://example.com/data.csv` | — | | REST API | `https://api.example.com/v1/items` or `URL \| path` | Bearer token (optional) | | Google Sheets | `spreadsheetId` or `spreadsheetId/Tab` | OAuth access token | | Dataset | dataset id | — | | Key-value | record `key`, or `storeId/key` | — | | JSON / JSONL | paste the text | — | | Inline records | `[{"id": 1}]` | — | **On n8n and the API**, send `source_type` + `source_config` as below. `destination_type` still works; if you omit it the Actor writes to this run's Dataset. | Source | `source_type` | `source_config` | Credential | | ----------------------- | --------------- | --------------------------------------------------------------------- | --------------------------- | | SQL database | `sql` | `{}`, or `{"tables": ["customers"]}` to limit it | `sql_connection_string` | | File over HTTP | `file_url` | `{"url": "https://example.com/data.csv"}` | — | | REST API | `rest_api` | `{"url": "https://api.example.com/v1/items", "records_path": "data"}` | `rest_api_auth_token` | | Google Sheets | `google_sheets` | `{"spreadsheet_id": "1AbC...", "sheets": ["Sheet1"]}` | access token **or** API key | | Another Actor's dataset | `dataset` | `{"datasetId": "aBcD1234"}` | — | | A key-value record | `kvs` | `{"key": "OUTPUT"}` | — | | Pasted records | `inline` | `{"records": [{"id": 1}]}` | — | | Pasted JSON | `json` | `{"text": "[{\"id\": 1}]"}` | — | | Pasted JSONL | `jsonl` | `{"text": "{\"id\": 1}\n"}` | — | #### A complete run, for each source Every block below is a whole INPUT for the API or n8n. On the Console, pick the matching Recipe and fill Where instead. **SQL — one table, incremental.** Put the URL in `sql_connection_string` (encrypted), not in `source_config`. ```json { "source_type": "sql", "source_config": { "tables": ["customers"], "cursor_fields": { "customers": "updated_at" } }, "destination_type": "apify_dataset", "streams": [ { "name": "customers", "sync_mode": "incremental", "cursor_field": "updated_at" } ] } ``` **A CSV or JSON file over HTTP.** ```json { "source_type": "file_url", "source_config": { "url": "https://example.com/data.csv" }, "destination_type": "apify_dataset" } ``` **A REST endpoint with paging.** `records_path` is where the array sits in the response body; `next_url_path` is where the next-page link sits in that same body. `cursor_field` is a separate thing — incremental sync between runs, not paging within one. ```json { "source_type": "rest_api", "source_config": { "url": "https://api.example.com/v1/items", "records_path": "data", "pagination": "next_url", "next_url_path": "next", "cursor_field": "updated_at" }, "destination_type": "apify_dataset" } ``` **Google Sheets.** `sheets` is required; send the OAuth token as `google_sheets_access_token` (or an API key as `google_sheets_api_key`), never both. ```json { "source_type": "google_sheets", "source_config": { "spreadsheet_id": "1AbC...", "sheets": ["Sheet1"] }, "destination_type": "apify_dataset" } ``` **Paste records straight in** — the fastest way to see what the Actor does to your data. ```json { "source_type": "inline", "source_config": { "records": [{ "id": 1, "email": "A@Example.COM " }] }, "destination_type": "apify_dataset" } ``` **Clean another Actor's output**, then write it to a named dataset of your own. ```json { "source_type": "dataset", "source_config": { "datasetId": "aBcD1234" }, "destination_type": "apify_dataset", "destination_config": { "dataset_name": "my-silver" } } ``` **SQL** — use a **read-only** database user. Paste a SQLAlchemy async URL into the connection-string box: `postgresql+asyncpg://USER:PASSWORD@HOST:5432/DB`, `mysql+aiomysql://USER:PASSWORD@HOST:3306/DB`, or `sqlite+aiosqlite:///data.db` for a file. Optional `source_config` keys: `schema_name`, `tables`, `cursor_fields`, `query`. Leave **Streams** empty to sync every discovered table in `full_refresh`. To limit tables, set Streams to `[{"name": "customers"}]`. An empty list means zero tables, not all of them. **REST** — pages are read in order. The cursor is committed only at the end of the stream, so an interrupted run re-reads rather than skipping. **Paging.** `pagination` defaults to `none` — one request, one page. `page` fits APIs that take a page-number parameter (`?page=1`, `?page=2`, ...); `next_url` fits APIs that return a link to the next page in the response body. Offset-style APIs (`$offset`, `skip`, `offset`) aren't supported yet — `page` steps a raw parameter by 1, not by page size, so a second page overlaps and duplicates most of the first instead of advancing to new rows. That makes an offset-style API safe to read for exactly one page, not a slow-but-correct multi-page one — paging past the first page returns duplicates, not new data. To page, pass `source_config` alongside the recipe; your keys win over the recipe's: `{"recipe": "rest_api", "locator": "https://api.example.com/items", "source_config": {"pagination": "page", "page_param": "page", "max_pages": 50}}`. `source_config` is hidden in the Console, so this route is API / n8n only today. **Google Sheets** — `sheets` is required. Credential is **exactly one** of `access_token` or `api_key`, with read scope on that sheet only. **File / JSON / JSONL / KVS** — format is detected from the extension, then Content-Type; set `"format"` when neither is reliable. Add `"repair": true` to let the Actor fix malformed JSON (trailing commas, single quotes) instead of failing. Repair is **off** by default: it guesses, and a quiet wrong guess is worse than a stopped run. **Incremental vs full refresh.** Only `sql` and `rest_api` can resume. The other seven re-read everything every run — **Reset cursor** does nothing for them. On an incremental stream, a second run that reads 0 rows usually means the cursor is current; tick **Reset cursor** to re-read from scratch. ### Destinations `destination_type` is `apify_dataset` (default) or `apify_kv`. Rows cost the same either way. **Apify Dataset** is what the Console previews and what Export downloads (JSON, CSV, Excel, HTML). Leave **Destination configuration** empty unless you want a named Silver dataset — `{"dataset_name": "my-silver"}` — or a different stream-name field, `{"stream_field": "_source"}` (default `_stream`). **Key-value store** — set `destination_type` to `apify_kv`, optionally with `{"key_prefix": "run1-"}`. It writes one JSON key per batch: `{key_prefix}{stream}-{batch:06d}`. A re-run **overwrites those keys**; it does not delete leftover higher keys. If the first run wrote 20 rows (several keys) and the next wrote 4, the store still holds the old higher keys. Use a fresh `key_prefix` when the row count can shrink, or stay on Dataset. ### Output — what lands where You can download the Silver dataset as JSON, CSV, Excel, or HTML from the run's dataset tab. | Where to look | What is in it | | --------------------------------- | ------------------------------------------------------------------------------------------- | | Default dataset | Silver records (no SyncReport trailer — the report is KVS `SUMMARY`) | | Dataset `datapipeline-bronze` | Every row as read, nested under `data`, with `record_id`, `run_id`, `stream`, `ingest_time` | | Dataset `datapipeline-quarantine` | Rejected rows with `error_code`, `stage`, `run_id`, `record_id`, `config_hash` | | Default key-value store | `MANIFEST` and `SUMMARY` on every non-dry run; `PROFILE` when a reporting stage ran | Bronze and quarantine are **named** datasets: they persist and **accumulate across runs**. Filter by `run_id` to see one run. KV records (`MANIFEST`, `SUMMARY`, `PROFILE`) are overwritten each run. A **named** Silver dataset (`destination_config.dataset_name`) is the same kind of shared bucket: it **accumulates across runs**. Full refresh does **not** empty it. Silver rows have **no** `run_id`. Leave destination unnamed to get this run's fresh default dataset. Filter Bronze and quarantine by `run_id` to see one run. Bronze is **not** redacted — it is the replay source, including when pii redacts Silver and quarantine. Silver carries your fields plus `_stream`. The engine `record_id` used for replay sits on Bronze and quarantine items, not on Silver. After a fan-out (`normalize` with `record_path`), each child also carries `_record_id` (`{parent}#{index}`) so `dedup` can keep siblings distinct; payload `id` is shared, so key `dedup` on `_record_id` for one Silver row per child. A second fan-out extends rather than restarts the id (`{parent}#{index}#{index}`), so leaves of a chained fan-out stay distinct. Non-fan-out Silver omits `_record_id`. `PROFILE` is one record, nested by stream then report. A run with `schema`, `profile`, and `dedup` stores `PROFILE[""]["SCHEMA_REPORT"]`, `["PROFILE_REPORT"]`, and `["DEDUP_REPORT"]`. #### Example: before and after Input row, as read: ```json { "id": "1", "e_mail": "ADA@Example.com ", "signup": "2026-01-02", "notes": null } ``` Bronze stores that row **under `data`**. Query Bronze for `data.email`, not `email`. Custom pipeline (only when preset is **Custom**): ```json [ { "stage": "map", "config": { "field_map": [["id"], ["email", "e_mail"], ["signup"], ["notes"]] } }, { "stage": "types", "config": { "types": { "id": "int" } } }, { "stage": "pii", "config": { "fields": ["email"], "strategy": "hash", "salt": "example-salt" } } ] ``` Silver row: ```json { "id": 1, "email": "501821b0a30aeede2d2b66d8c2949c137a023cd1c70e7bab8a6b656aa416d760", "signup": "2026-01-02", "notes": null, "_stream": "people" } ``` `map` renamed `e_mail` → `email`, `types` turned `"1"` into `1`, `pii` hashed the address, the destination added `_stream`. `pii` hashes the value **as it arrived** — capitals and the trailing space are in the digest. Normalise before you redact if you will join on the hash. ### How much does it cost? Pay per event. You are **not** billed for Bronze rows. | What | Event | Price | | -------------------- | ---------------- | ----------------------------- | | Each run | `actor-start` | **$0.30** | | Each Silver row | `silver-row` | **$0.0004** ($0.40 per 1,000) | | Each quarantined row | `quarantine-row` | **$0.0002** ($0.20 per 1,000) | | Each stream profiled | `profile-report` | **$0.08** | Examples: 100 Analysis-ready rows (start + Silver + one PROFILE) → **$0.42**. 10,000 Silver rows, no PROFILE → **$4.30**. **What is free.** A run rejected for bad input is not charged. **Dry run** is free of every charge. **What is billed twice.** Rows are billed as they are written. If a batch is retried after a network failure, those rows are billed again — they appear again in the dataset (at-least-once). `MANIFEST.charges` records what this run billed and anything it failed to bill. Set these same event names and prices in **Apify Console → Monetization**. Do not also enable `apify-actor-start` or `apify-default-dataset-item`, or buyers are double-charged. ### Pipeline presets **Analysis-ready** (`pre_analysis`) is the Console default. It runs `pii → schema → validate → profile`. It profiles; it does not coerce CSV strings to numbers. It changes no values. It drops no records, with one exception: `profile` cannot key a value nested past roughly 3,000 levels, and that record goes to quarantine (`STAGE_UNCAUGHT_ERROR`) — quarantined, never silently lost; the rest of the run continues. Under this preset, `pii` and `validate` get empty config, so nothing is redacted and nothing is rejected. You get a `SCHEMA_REPORT` over up to `sample_size` records (1000 by default) and a `PROFILE_REPORT` over every record, with the redaction and validation slots already in a legal order. To map columns, convert types, dedup, or redact: set the preset to **Custom** and fill **Transform stages**. Setting a preset *and* a pipeline is rejected (`PIPELINE_PRESET_CONFLICT`). Neither `schema` nor `profile` ever modifies a record. Profiling is evidence, not permission to silently coerce types. ### Custom pipeline Only when **Pipeline preset** is **Custom**. Ordered list of `{"stage", "config"}`. Empty list is a passthrough. | Stage | What it does | | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `normalize` | Flattens nested objects to dotted keys; `record_path` fans a nested list into sibling rows | | `rename` | Renames keys, each keeping its original position | | `map` | Harmonises aliases (`e_mail`, `Email`) onto one canonical key; unmapped keys are dropped unless `keep_unmapped` is set | | `drop` | Removes keys | | `missing` | Fills, flags, or drops records on null — only `null` counts as missing, never `""`, `0`, or `false` | | `types` | Converts to int/float/str/bool/datetime/date — **rejects** values it cannot convert rather than nulling them (`on_error` can select `flag` or `null`) | | `derive` | Computes fields from a sandboxed expression over other fields | | `validate` | JSON Schema per record; failures go to quarantine | | `dedup` | Removes duplicates within the run, under a hard `max_keys` budget. Prefer `keep: first` on large streams. After `normalize.record_path`, payload `id` is shared — key on `_record_id` for one Silver row per child | | `pii` | Masks, hashes, or removes sensitive fields — top-level keys only. Put `pii` **before** `schema`/`profile` | | `schema` | Reports each field's JSON types and null ratios over a sample of at most `sampleSize` records | | `profile` | Reports completeness and distinct-value counts over **every** record | | `passthrough` | Emits each record unchanged | `types` targeting `date`/`datetime` becomes a Python date/datetime; the Dataset stores the string form. A `date` is `"2026-01-02"`; a `datetime` is `"2026-01-02 09:00:00+00:00"` (a space, not `T`). Convert to `str` if you need strict ISO-8601. `format:` is enforced (`email`, `date`, `date-time`, `uri`, `uuid`, `ipv4`, `ipv6`, and the other checkers this build ships). A schema naming a `format` this build cannot check is **rejected at startup** rather than silently ignored — before this change every `format:` was annotation-only, so a schema that looked like it validated emails did not. ### Input See the **Input** tab for the full form. The fields that surprise people: - **Recipe / Where / Secret** — the Store form. Recipe names the connector; Where is the locator; Secret is the password or token when that recipe needs one. - **`source_type` / `source_config`** — what n8n and the API send. When `source_type` is set, Recipe and Where are ignored. `source_config` ships no prefill: a prefilled example would be right for one source and rejected by the other eight. - **Secret vs JSON** — a password typed into `source_config` is stored in the run's INPUT record in the clear. Secret is encrypted. - **Streams** — omit to sync every table. `[]` means zero tables. - **Sync key** — optional cursor namespace. Set it to keep incremental cursors stable across config edits. - **Quality policy** — Fail fast writes every row the failing batch rejected to quarantine, then aborts (earlier batches stay written; trailing flush is batch-local). Quarantine continues and marks MANIFEST partial. Warn quarantines and counts those rows on `MANIFEST.counts.quarantined`, but MANIFEST status stays succeeded. Missing-drop and dedup are `dropped`, not quarantine. - **Fail on partial** — Exit non-zero when quarantined records made the run partial. Default off: Apify can show SUCCEEDED while MANIFEST.status is partial. Turn on if a scheduler should fail on rejects. Fail fast already aborts. Warn never counts as partial. - **On operational failure** — legacy; governs destination/stream failures, not bad rows. Quality policy is the control for bad records. A local `INPUT.json` is **not** the Console. The Console defaults `pipeline_preset` to `pre_analysis`. A hand-written file that omits the field falls back to `custom` with an empty pipeline (passthrough, no reports). Set it explicitly when you run locally. ### FAQ #### Why are my rows in quarantine, not Silver? A stage rejected them. Open `datapipeline-quarantine` and read `error_code` and `stage`. #### Why did the run exit 2 immediately? Input was rejected before any read. The status message names the field. Common cases: `PIPELINE_PRESET_CONFLICT` (preset and custom pipeline both set — set preset to `custom` or clear the pipeline); `PII_STAGE_ORDER_INVALID` (`pii` after `schema`/`profile` — move `pii` first); unknown stage name (the log lists registered stages). #### Why `PII_DEDUP_ORDER_INVALID`? A `pii` with `strategy: mask` or `remove` runs before a `dedup` that keys on a field it redacts. Move `pii` after `dedup`, switch to `strategy: hash`, or key `dedup` on a field `pii` does not touch. The check follows the value through `rename` / `map` / `derive` / `missing` (flag mode), and out of scope through `drop` or a `map` that projects it away. A `normalize` with `record_path` between the two cannot be traced, so that order is rejected. #### Why `SYNC_ABORTED`? A stage aborted the stream. Fail fast writes every row that batch rejected to quarantine first; rows from earlier batches stay written, and any trailing flush is batch-local. The status only says `SYNC_ABORTED`; the stage's code is in `streams[].error_codes` in `MANIFEST` and in the log. For `DEDUP_KEY_LIMIT_EXCEEDED`: raise `max_keys`, narrow `keys`, or set `on_limit: quarantine`. #### Why `SOURCE_CONFIG_INVALID`? The connection string could not be parsed. This is found at connect time, not at input validation. #### Will a retried run duplicate Silver rows? Yes, on Dataset: it is append-only. Deduplicate downstream on your own key, or join Silver back to Bronze on `record_id`. Silver itself has no engine `record_id` (fan-out children carry `_record_id` so `dedup` can key on it). #### Does this Actor scrape websites? No. It reads sources you configure (databases, APIs, sheets, files, Apify storage, pasted JSON). You are responsible for having a lawful basis to process that data. Bronze stores unredacted raw rows; use `pii` on a **Custom** pipeline if Silver must not contain personal data. Personal data is protected by the GDPR and similar laws — do not process it unless you have a legitimate reason. #### Where do I report a bug? Use this Actor's **Issues** tab. For programmatic runs, see the **API** tab. ### Automate it - **n8n** — two importable workflows, core nodes only: [`docs/n8n/`](docs/n8n/README.md) - **Agents** — operation patterns over the Actor and storage APIs: [`docs/agents/ops-patterns.md`](docs/agents/ops-patterns.md) # Actor input Schema ## `recipe` (type: `string`): What you are reading. Fill Where next. Put the password or token in Secret only if this recipe needs one (PostgreSQL, MySQL, REST, Google Sheets). ## `locator` (type: `string`): What you type depends on Recipe. • PostgreSQL / MySQL: user@host/database — optional :port (defaults 5432 / 3306). Password goes in Secret, not here: Where is NOT encrypted, so a password pasted inside a URL is stored in the run's INPUT record in the clear. A postgresql:// or mysql:// URL is also accepted (percent-encode a username containing @, e.g. user%40server); its own password wins over Secret if you include one. • SQLite: a file path, or :memory:. • File over HTTP: the file URL. • REST API: the endpoint URL, or URL | records\_path when the array is nested. • Google Sheets: spreadsheetId, or spreadsheetId/Tab (default tab is Sheet1). • Dataset: the dataset id. • Key-value store: the record key, or storeId/key to read another store. • JSON / JSONL: paste the text. • Inline records: a JSON array of objects, e.g. \[{"id": 1}]. ## `secret` (type: `string`): Password for PostgreSQL/MySQL, Bearer token for REST, or OAuth access token for Google Sheets. Leave empty for files, datasets, paste, and SQLite. Stored encrypted. ## `pii_salt` (type: `string`): Hash pepper for the pii stage's hash strategy. When set, it overrides any salt in pipeline JSON for every pii step. Leave empty to use the stage-config salt — a salt typed into pipeline JSON is stored on the run INPUT in plaintext. ## `destination_type` (type: `string`): Where cleaned Silver rows go. Apify Dataset is the usual choice — it is what the Console previews and what Export downloads. Key-value store writes one JSON key per batch instead, which is useful when a later step fetches whole batches by key. ## `destination_config` (type: `object`): Usually leave empty. For apify\_dataset: {"dataset\_name": "my-silver"} for a named dataset — note that a named dataset accumulates across runs and full refresh does NOT empty it, so re-running appends a second copy; leave this empty to get this run's own fresh dataset. "stream\_field" renames the \_stream column. For apify\_kv: {"key\_prefix": "run1-"} — one key per batch, named {prefix}{stream}-{batch:06d}, so a re-run overwrites the same keys instead of duplicating rows. ## `pipeline_preset` (type: `string`): Start here. Analysis-ready (pre\_analysis runs pii -> schema -> validate -> profile, which does not change values) profiles your data — it does not coerce CSV strings to numbers. Open Silver and PROFILE after the run. It drops no records either, with one exception: profile cannot compute a distinct-value key for a value nested past roughly 3000 levels, and that record is quarantined (STAGE\_UNCAUGHT\_ERROR) rather than reaching Silver — quarantined, never silently lost, and the run continues. Choose Custom only to map, convert types, dedup, or redact. Do not set a custom pipeline while this is still Analysis-ready. ## `pipeline` (type: `array`): Only when Pipeline preset is Custom. Ordered stages as \[{"stage", "config"}]. Available: normalize, rename, map, drop, missing, types, derive, validate, dedup, pii, schema, profile, passthrough. Empty list is a passthrough. Put pii before schema/profile. Prefer keep=first on large dedup runs. After normalize.record\_path, payload id is shared; key dedup on \_record\_id for one Silver row per child. See the README for a copy-paste example. ## `streams` (type: `array`): Leave unset to sync every discovered table in full\_refresh. To limit tables, list \[{"name": "customers"}]. Empty list means zero tables, not all of them. ## `batch_size` (type: `integer`): Records per batch moved through the pipeline. ## `sync_key` (type: `string`): Optional explicit cursor namespace. Set it to keep incremental cursors stable across config edits. ## `reset_cursor` (type: `boolean`): Tick this to re-read an incremental table from scratch. Full-refresh tables (the default) already re-read every run. ## `dry_run` (type: `boolean`): Check config and list tables without writing Bronze, Silver, or a manifest. No row charges. ## `quality_policy` (type: `string`): What to do with a row a stage rejects. Fail fast writes every row the failing batch rejected to quarantine (DLQ), then aborts. Quarantine (recommended) writes it to datapipeline-quarantine and continues; MANIFEST.status is partial. Warn writes it to quarantine (DLQ) and counts it on MANIFEST.counts.quarantined; MANIFEST.status stays succeeded. Rejected rows never reach Silver. Missing-drop and dedup remove rows without a quarantine item. ## `fail_on_partial` (type: `boolean`): Exit non-zero when quarantined records made the run partial. Default off: Apify can show SUCCEEDED while MANIFEST.status is partial. Turn on if a scheduler should fail on rejects. Fail fast already aborts. Warn never counts as partial. ## `partial_failure` (type: `string`): Legacy field, governs operational errors (a destination write or a whole stream failing) rather than data errors. fail\_fast stops the run; continue dead-letters the batch and keeps going. Leave unset unless you need it — Quality policy above is the control for bad records. ## `source_type` (type: `string`): API and n8n send this instead of Recipe. When it is set, Recipe and Where are ignored. Where to read from. Raw rows land in Bronze before any cursor moves. One of: dataset, file\_url, google\_sheets, inline, json, jsonl, kvs, rest\_api, sql. ## `source_config` (type: `object`): One JSON object describing the source you picked. Copy the line for your source and edit it — the README has the same examples with notes. • sql: {"tables": \["customers"]} — put the URL in Database connection string below, or set "connection\_string" here. • file\_url: {"url": "https://example.com/data.csv"}. • rest\_api: {"url": "https://api.example.com/v1/items", "records\_path": "data"} — "pagination" defaults to "none" (one request, one page); add "pagination": "page" with "page\_param"/"max\_pages" to read more (offset-style APIs like $offset/skip aren't supported yet), since this box is the only way to reach paging today. • google\_sheets: {"spreadsheet\_id": "1AbC...", "sheets": \["Sheet1"]}. • dataset: {"datasetId": "aBcD1234"}. • kvs: {"key": "OUTPUT"} — add "storeId" to read another run's store. • inline: {"records": \[{"id": 1}]}. • json / jsonl: {"text": "\[{"id": 1}]"}. Credentials go in the encrypted boxes below, not in here. Optional on file\_url, json, jsonl and kvs: "format" when the extension and Content-Type are both unreliable, and "repair": true to fix malformed JSON (trailing commas, single quotes) instead of failing — off by default, because a quiet wrong guess is worse than a stopped run. ## `sql_connection_string` (type: `string`): Used only when Source is SQL database. A SQLAlchemy async URL for a read-only user — postgresql+asyncpg://USER:PASSWORD@HOST:5432/DB, mysql+aiomysql://USER:PASSWORD@HOST:3306/DB, or sqlite+aiosqlite:///data.db for a file. Stored encrypted, which a URL pasted into Source configuration is not. Hidden on the Store form; Recipe + Where builds this for Console runs. ## `rest_api_auth_token` (type: `string`): Used only when Source is REST API, and optional. Sent as Authorization: Bearer … . Stored encrypted. Hidden on the Store form; Secret fills this for REST Console runs. ## `google_sheets_access_token` (type: `string`): Used only when Source is Google Sheets. An OAuth access token with read scope on that sheet. Fill this or API key, never both. Stored encrypted. Hidden on the Store form; Secret fills this for Sheets Console runs. ## `google_sheets_api_key` (type: `string`): Used only when Source is Google Sheets. Fill this or Access token, never both. Stored encrypted. Hidden on the Store form; send this field from the API when you authenticate with a key instead of Secret. ## `profile_policy` (type: `object`): enabled=false skips the profile stage. sample\_size is how many rows schema inspects (default 1000). Profiling never rewrites values. ## `bronze` (type: `object`): dataset\_name selects the dataset raw records land in before the cursor advances. Bronze is kept separate from Silver on purpose: it is the replay source, so every item carries record\_id and run\_id. ## Actor input object example ```json { "destination_type": "apify_dataset", "pipeline_preset": "pre_analysis", "batch_size": 500, "reset_cursor": false, "dry_run": false, "quality_policy": "quarantine", "fail_on_partial": false, "profile_policy": { "enabled": true, "sample_size": 1000 }, "bronze": { "dataset_name": "datapipeline-bronze" } } ``` # Actor output Schema ## `dataset` (type: `string`): Dataset containing the cleaned Silver records, one item per record. The run's SyncReport is in the key-value store under SUMMARY, not in this dataset. ## `reports` (type: `string`): Key-value store holding MANIFEST (versions, counts, checkpoint, output pointers, billing ledger), SUMMARY (the SyncReport), and PROFILE (per-stream schema/profile evidence). # 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 '' with your token const client = new ApifyClient({ token: '', }); // Prepare Actor input const input = { "bronze": { "dataset_name": "datapipeline-bronze" } }; // Run the Actor and wait for it to finish const run = await client.actor("ahmedmohamed/silverset").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 '' with your token. client = ApifyClient("") # Prepare the Actor input run_input = { "bronze": { "dataset_name": "datapipeline-bronze" } } # Run the Actor and wait for it to finish run = client.actor("ahmedmohamed/silverset").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 '{ "bronze": { "dataset_name": "datapipeline-bronze" } }' | apify call ahmedmohamed/silverset --silent --output-dataset ``` ## MCP server setup ```json { "mcpServers": { "apify": { "type": "http", "url": "https://mcp.apify.com/?tools=fetch-actor-details,ahmedmohamed/silverset" } } } ``` 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 ` 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/2oAec8HKKvyM2vEgd/builds/jcrWRaDOLKXLa2b0h/openapi.json