调度与自动化

增量采集、生成 cron/systemd 条目，以及 Webhook 通知

增量采集

默认情况下，anysite dataset collect 每次运行都会采集所有输入。加上 --incremental 后，它会跟踪 哪些输入值已经采集过（记录在数据集旁的 metadata.json 中），并在后续运行中跳过它们。

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

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

按数据源控制刷新

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)

值	行为
auto	遵循 --incremental 标志（默认）
always	即使使用了 --incremental 也总是重新采集

对搜索结果、热门内容或任何频繁变化的数据使用 refresh: always。对用户资料等静态数据使用 refresh: auto。

重置游标

# 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

生成 Cron / systemd 条目

anysite dataset schedule 本身不会运行调度器 — 它会打印一行 crontab 内容，或者写出 systemd .service/.timer 单元文件，由您自行安装：

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

请自行用 crontab -e 添加打印出的这一行，或将其重定向到 crontab 文件中。对于 systemd：

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

这会将 <service-name>.service 和 <service-name>.timer 单元内容写到标准输出 — 请分别保存到 /etc/systemd/system/，然后自行执行 systemctl daemon-reload && systemctl enable --now <name>.timer。这两种模式都不会自行启动后台进程或守护进程。

常见 Cron 表达式

表达式	调度周期
0 9 * * *	每天上午 9:00
0 */6 * * *	每 6 小时一次
0 9 * * 1	每周一上午 9:00
0 9 1 * *	每月第一天上午 9:00
*/30 * * * *	每 30 分钟一次

Webhook 通知

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

每个钩子都可以接受一个可选的 headers 映射，用于自定义 HTTP 头。POST 的 JSON 负载内容为：

• on_complete：event、dataset、timestamp、record_count、source_count、duration
• on_failure：event、dataset、timestamp、error、duration

其中不包含运行 ID — 请使用 anysite dataset history <name> 来查找该次运行。

运行历史与日志

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

显示运行列表（默认 20 条），包括状态、时间戳、耗时、记录数和数据源数量。

anysite dataset logs my-dataset --run 42

显示某次运行的详细日志（省略 --run 时默认使用最近一次运行）。

自动加载到数据库

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

--load-db <connection> 会把一个 --load-db 标志固化进生成的 crontab/systemd 命令中，因此每次调度运行的 dataset collect 也会将结果加载到该数据库连接 — 见 Database Operations。

完整自动化管道示例

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"

后续步骤

Database Connections

为数据存储设置 SQLite、PostgreSQL 或 ClickHouse 连接

LLM Analysis

为您的管道添加 AI 驱动的增强能力