Scheduling & Automation

Incremental collection, generating cron/systemd entries, and webhook notifications

Incremental Collection

By default, anysite dataset collect collects all inputs every run. With --incremental, it tracks which input values have already been collected (in metadata.json alongside the dataset) and skips them on subsequent runs.

# First run: collects everything
anysite dataset collect dataset.yaml

# Subsequent runs: only collects new inputs
anysite dataset collect dataset.yaml --incremental

Per-Source Refresh Control

sources:
  - id: companies
    endpoint: /api/linkedin/search/companies
    params: { keywords: "AI startup", count: 100 }
    refresh: always    # Always re-collect (time-sensitive data)

  - id: employees
    endpoint: /api/linkedin/company/employees
    dependency: { from_source: companies, field: urn.value }
    input_key: companies
    refresh: auto      # Respect --incremental flag (default)

Value	Behavior
`auto`	Respects the `--incremental` flag (default)
`always`	Always re-collects, even with `--incremental`

Use refresh: always for search results, trending content, or any data that changes frequently. Use refresh: auto for static data like user profiles.

Reset Cursors

# Reset all sources — next run collects everything
anysite dataset reset-cursor dataset.yaml

# Reset one source only
anysite dataset reset-cursor dataset.yaml --source profiles

Generating Cron / systemd Entries

anysite dataset schedule does not run a scheduler itself — it prints a crontab line, or writes out systemd .service/.timer unit files, for you to install:

name: daily-monitoring
description: Daily competitor monitoring pipeline

sources:
  - id: competitor_posts
    endpoint: /api/linkedin/company/posts
    from_file: competitors.txt
    input_key: company
    parallel: 3

schedule:
  cron: "0 9 * * *"    # Every day at 9:00 AM

storage:
  format: parquet
  path: ./data/

# Print a crontab line (default when neither --crontab nor --systemd is given)
anysite dataset schedule dataset.yaml --incremental --load-db pg

Add the printed line yourself with crontab -e, or redirect it into a crontab file. For systemd:

anysite dataset schedule dataset.yaml --systemd --incremental --load-db pg

This writes <service-name>.service and <service-name>.timer unit content to stdout — save each to /etc/systemd/system/, then systemctl daemon-reload && systemctl enable --now <name>.timer yourself. Neither mode starts a background process or daemon on its own.

Common Cron Expressions

Expression	Schedule
`0 9 * * *`	Daily at 9:00 AM
`0 /6 * *`	Every 6 hours
`0 9 * * 1`	Every Monday at 9:00 AM
`0 9 1 * *`	First day of each month at 9:00 AM
`/30 * * *`	Every 30 minutes

Webhook Notifications

notifications:
  on_complete:
    - url: "https://hooks.slack.com/services/xxx/yyy/zzz"
  on_failure:
    - url: "https://alerts.example.com/anysite-failure"

Each hook accepts an optional headers map for custom HTTP headers. The POSTed JSON payload is:

on_complete: event, dataset, timestamp, record_count, source_count, duration
on_failure: event, dataset, timestamp, error, duration

It does not include a run ID — use anysite dataset history <name> to look up the run.

Run History and Logs

anysite dataset history my-dataset
anysite dataset history my-dataset --limit 5

Shows a list of runs (default 20) with status, timestamp, duration, record count, and source count.

anysite dataset logs my-dataset --run 42

Shows detailed logs for a specific run (defaults to the latest run if --run is omitted).

Auto-Load to Database

anysite dataset schedule dataset.yaml \
  --incremental \
  --load-db pg

--load-db <connection> bakes a --load-db flag into the generated crontab/systemd command, so each scheduled dataset collect run also loads results into that database connection — see Database Operations.

Complete Automated Pipeline Example

name: lead-monitoring
description: Automated lead discovery and enrichment

sources:
  - id: search
    endpoint: /api/linkedin/search/users
    params:
      keywords: "Head of Engineering"
      count: 100
    refresh: always

  - id: profiles
    endpoint: /api/linkedin/user
    dependency: { from_source: search, field: urn.value }
    input_key: user
    parallel: 5
    rate_limit: "10/s"
    on_error: skip
    refresh: auto

  - id: enriched
    type: llm
    dependency: { from_source: profiles, field: name }
    llm:
      - type: classify
        categories: "high_priority,medium,low"
        output_column: lead_score

storage:
  format: parquet
  path: ./data/

schedule:
  cron: "0 9 * * 1"    # Every Monday at 9 AM

notifications:
  on_complete:
    - url: "https://hooks.slack.com/services/xxx"
  on_failure:
    - url: "https://alerts.example.com/fail"

Next Steps

Database Connections

Set up SQLite, PostgreSQL, or ClickHouse connections for data storage

LLM Analysis

Add AI-powered enrichment to your pipelines