Dataset Pipelines

Build declarative YAML workflows with chained data sources and automated collection

Overview

Dataset pipelines define multi-source data collection workflows in YAML. Sources can depend on each other, forming chains like: search → company profiles → employees → posts. The CLI handles execution order, parallelism, error handling, and Parquet storage automatically.

Dataset pipelines require the data extra: pip install "anysite-cli[data]"

Create a Pipeline

anysite dataset init my-dataset

This creates a dataset.yaml file with a starter configuration:

name: my-dataset
description: My data collection pipeline

sources:
  - id: search_results
    endpoint: /api/linkedin/search/users
    params:
      keywords: "software engineer"
      count: 50
    parallel: 1
    rate_limit: "10/s"
    on_error: stop

storage:
  format: parquet
  path: ./data/

Run Collection

# Full collection
anysite dataset collect dataset.yaml

# Preview what will be collected (no API calls)
anysite dataset collect dataset.yaml --dry-run

# Collect a specific source only
anysite dataset collect dataset.yaml --source search_results

# Skip LLM processing steps
anysite dataset collect dataset.yaml --no-llm

# Cap how many input values a dependent/from_file source processes (pilot runs)
anysite dataset collect dataset.yaml --limit 20

Multi-Source Pipeline Example

name: competitor-research
description: Collect and analyze competitor company data

sources:
  # Step 1: Search for companies
  - id: companies
    endpoint: /api/linkedin/search/companies
    params:
      keywords: "AI startup"
      count: 100
    parallel: 1

  # Step 2: Get employee list for each company (depends on Step 1)
  - id: employees
    endpoint: /api/linkedin/company/employees
    dependency:
      from_source: companies
      field: urn.value
      dedupe: true
    input_key: companies
    parallel: 3
    rate_limit: "10/s"

  # Step 3: Get full profiles for each employee (depends on Step 2)
  - id: profiles
    endpoint: /api/linkedin/user
    dependency:
      from_source: employees
      field: urn.value
    input_key: user
    parallel: 5
    rate_limit: "10/s"
    on_error: skip

storage:
  format: parquet
  path: ./data/

The CLI resolves the dependency graph and executes sources in the correct order.

Pipeline Configuration Reference

Field	Description
`name`	Pipeline name (used for history and logging)
`description`	Optional description
`sources`	List of data sources (see Source Types)
`storage.format`	Accepted but not used for the raw collection step — collected data is always written as Parquet in this CLI version
`storage.path`	Output directory (default: `./data/`)
`schedule.cron`	Cron expression consumed by `dataset schedule` (see Scheduling)
`notifications`	Webhook URLs for success/failure events

Per-source export.format (csv, json, or jsonl) is implemented — see Per-Source Transform & Export — it writes an additional exported file alongside the Parquet snapshot, it doesn't change how the snapshot itself is stored.

Dataset Commands

Command	Description
`anysite dataset init <name>`	Create a new dataset with starter YAML
`anysite dataset collect <yaml>`	Run the collection pipeline
`anysite dataset collect <yaml> --dry-run`	Preview execution plan without API calls
`anysite dataset collect <yaml> --incremental`	Skip previously collected inputs
`anysite dataset collect <yaml> --source <id>`	Collect a single source
`anysite dataset collect <yaml> --limit <n>`	Cap input values per dependent/from_file source
`anysite dataset status <yaml>`	Check collection status
`anysite dataset collect <yaml> --load-db <conn>`	Auto-load results into a database
`anysite dataset query <yaml> --sql "..."`	Query collected data with SQL
`anysite dataset stats <yaml> --source <id>`	Per-column statistics for one source
`anysite dataset profile <yaml>`	Per-source quality summary
`anysite dataset validate <yaml>`	Validate config without collecting (no API calls)
`anysite dataset history <name>`	View run history
`anysite dataset logs <name> --run`	View logs for a specific run
`anysite dataset reset-cursor <yaml>`	Reset incremental collection cursors

Next Steps

Source Types

Learn about the 5 source types: independent, from_file, dependent, union, and LLM

Scheduling

Set up incremental collection, cron/systemd entries, and webhook notifications