Source Types

Six source types for building flexible data collection pipelines

Overview

Dataset pipelines support 6 source types, each designed for a different data collection pattern. Sources can be combined to build complex multi-step workflows. Independent, From File, and Dependent below are all the same underlying type: api source (the default when type is omitted) — they differ only in which fields are set.

Type	Purpose	Key Config
Independent	Single API call with static parameters	`endpoint`, `params`
From File	Batch calls iterating over a file	`from_file`, `input_key`
Dependent	Batch calls using values from a parent source	`dependency`, `input_key`
Union	Combine records from multiple sources	`type: union`, `sources`
LLM	Process data through an LLM model	`type: llm`, `llm`
SQL	Query Parquet views or a named database connection	`type: sql`, `query`

Independent Source

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

When to use: Starting point for pipelines, search queries, single profile lookups.

From File Source

sources:
  - id: company_profiles
    endpoint: /api/linkedin/company
    from_file: companies.txt
    input_key: company
    parallel: 3
    rate_limit: "10/s"
    on_error: skip

Supported file formats:

TXT — one value per line
CSV — uses the column matching input_key
JSON/JSONL — uses the field matching input_key

When to use: You have a pre-existing list of URLs, IDs, or search terms.

Dependent Source

sources:
  - id: companies
    endpoint: /api/linkedin/search/companies
    params:
      keywords: "AI startup"
      count: 100

  - 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"
    on_error: skip

Dependency Configuration

Field	Description
`from_source`	ID of the parent source
`field`	Field path to extract from parent results (dot-notation supported)
`dedupe`	Remove duplicate values before processing (default: `false`)

Shorthand: for a single dependent parameter, ${source.field} inline in params expands to the equivalent dependency/input_key pair automatically:

  - id: employees
    endpoint: /api/linkedin/company/employees
    params:
      companies: "${companies.urn.value}"

is equivalent to the explicit dependency: {from_source: companies, field: urn.value} / input_key: companies form above. The two forms cannot be mixed on the same source.

Multi-level chains are supported — a dependent source can itself be the parent of another dependent source:

sources:
  - id: companies
    endpoint: /api/linkedin/search/companies
    params: { keywords: "AI", count: 50 }

  - id: employees
    endpoint: /api/linkedin/company/employees
    dependency: { from_source: companies, field: urn.value }
    input_key: companies

  - id: profiles
    endpoint: /api/linkedin/user
    dependency: { from_source: employees, field: urn.value }
    input_key: user

  - id: posts
    endpoint: /api/linkedin/user/posts
    dependency: { from_source: profiles, field: urn.value }
    input_key: user

When to use: Multi-step data enrichment, going from search results to detailed profiles to activity data.

Union Source

sources:
  - id: search_cto
    endpoint: /api/linkedin/search/users
    params: { keywords: "CTO", count: 50 }

  - id: search_vp
    endpoint: /api/linkedin/search/users
    params: { keywords: "VP Engineering", count: 50 }

  - id: all_leaders
    type: union
    sources: [search_cto, search_vp]
    dedupe_by: urn.value

Union Configuration

Field	Description
`type`	Must be `union`
`sources`	List of source IDs to combine (must share the same endpoint)
`dedupe_by`	Field to deduplicate by (optional)

When to use: Merging results from multiple searches, combining data from different platforms.

LLM Source

sources:
  - id: profiles
    endpoint: /api/linkedin/user
    from_file: users.txt
    input_key: user

  - id: profiles_analyzed
    type: llm
    dependency:
      from_source: profiles
      field: name
    llm:
      - type: classify
        categories: "developer,recruiter,executive,other"
        output_column: role_type

      - type: enrich
        add:
          - "seniority:junior/mid/senior/lead"
          - "is_technical:boolean"

      - type: summarize
        max_length: 50
        output_column: bio_summary

LLM Operations

Operation	Description
`classify`	Categorize records into predefined categories
`enrich`	Extract new attributes (enums, strings, booleans, numbers)
`summarize`	Generate concise summaries
`generate`	Create text using templates with field placeholders

Pipeline LLM steps support these 4 operations only — match and deduplicate are available as standalone anysite llm match / anysite llm deduplicate commands, not as type: llm pipeline steps. Requires the llm extra: pip install "anysite-cli[llm]".

When to use: Adding AI-powered enrichment to your pipeline, categorizing or summarizing collected data.

SQL Source

Runs a SQL query and stores the result as its own source — either against the Parquet views of already-collected sources in the same pipeline, or against a named database connection:

sources:
  # Query dataset Parquet files via DuckDB (no external DB needed)
  - id: enriched_profiles
    type: sql
    query: |
      SELECT u.*, c.description as company_desc
      FROM profiles u
      LEFT JOIN companies c ON u._input_value = c._input_value

  # Query an external database via a named connection ('anysite db add')
  - id: billing_data
    type: sql
    connection: pg
    query: "SELECT name, email FROM subscriptions"

query_file (path to a .sql file) can be used instead of an inline query. Omit connection to query the pipeline's own collected sources.

Per-Source Transform & Export

sources:
  - id: companies
    endpoint: /api/linkedin/company
    from_file: companies.txt
    input_key: company
    transform:
      filter: '.employee_count > 10'
      fields: [name, url, employee_count]
      add_columns:
        batch: "q1-2026"
    export:
      - type: file
        path: ./output/companies-{{date}}.csv
        format: csv
    db_load:
      key: _input_value
      sync: full
      fields: [name, url, employee_count]

Transform Options

Field	Description
`filter`	jq-style filter expression to keep matching records
`fields`	List of fields to include in the output
`add_columns`	Static columns to add to every record

Export Options

Field	Description
`type`	Export type: `file` or `webhook`
`path`	Output file path (`file` type; supports `{{date}}` template)
`format`	Export format: `csv`, `json`, `jsonl` (default: `jsonl`)
`url`	Webhook URL (`webhook` type)
`headers`	Extra HTTP headers for the webhook

Database Load Options (per-source)

Field	Description
`key`	Unique key column for incremental sync
`sync`	Sync mode: `full` (default, includes DELETE) or `append` (no DELETE)
`fields`	Fields to load into the database

Input Templates

sources:
  - id: employees
    endpoint: /api/linkedin/company/employees
    dependency:
      from_source: companies
      field: urn.value
    input_key: companies
    input_template:
      companies:
        - type: company
          value: "{value}"
      count: 5

The {value} placeholder is replaced with each input value from the dependency.

Common Source Options

These options apply to type: api sources (independent, from_file, dependent):

Option	Description	Default
`parallel`	Number of concurrent workers	`1`
`rate_limit`	Maximum request rate (e.g., `"10/s"`)	No limit
`on_error`	Error handling: `stop` or `skip` (no `retry` at the source-config level)	`skip`
`refresh`	Incremental behavior: `auto`, `always`	`auto`

Next Steps

Scheduling

Set up incremental collection, cron/systemd entries, and webhooks

Database Loading

Load pipeline results into SQLite, PostgreSQL, or ClickHouse