数据集管道

构建具有链式数据源和自动化采集功能的声明式 YAML 工作流

概述

数据集管道以 YAML 定义多数据源的采集工作流。数据源之间可以相互依赖，形成如下链条： 搜索 → 公司资料 → 员工 → 帖子。CLI 会自动处理执行顺序、并行度、错误处理和 Parquet 存储。

数据集管道需要 data 可选依赖包：pip install "anysite-cli[data]"

创建管道

anysite dataset init my-dataset

这会创建一个带有初始配置的 dataset.yaml 文件：

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/

运行采集

# 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

多数据源管道示例

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/

CLI 会解析依赖关系图，并按正确的顺序执行各个数据源。

管道配置参考

字段	描述
name	管道名称（用于历史记录和日志）
description	可选的描述
sources	数据源列表（见 Source Types）
storage.format	会被接受但对原始采集步骤不生效 — 此 CLI 版本中，采集到的数据始终以 Parquet 格式写入
storage.path	输出目录（默认：./data/）
schedule.cron	供 dataset schedule 使用的 cron 表达式（见 Scheduling）
notifications	用于成功/失败事件的 Webhook URL

单个数据源的 export.format（csv、json 或 jsonl）确实已经实现 — 见 Per-Source Transform & Export — 它会在 Parquet 快照之外额外写出一份导出文件，不会改变快照本身的存储方式。

数据集命令

命令	描述
anysite dataset init <name>	创建一个带有初始 YAML 的新数据集
anysite dataset collect <yaml>	运行采集管道
anysite dataset collect <yaml> --dry-run	预览执行计划，不发起 API 调用
anysite dataset collect <yaml> --incremental	跳过此前已采集的输入
anysite dataset collect <yaml> --source <id>	只采集单个数据源
anysite dataset collect <yaml> --limit <n>	限制依赖型/文件型数据源处理的输入值数量
anysite dataset status <yaml>	检查采集状态
anysite dataset collect <yaml> --load-db <conn>	自动将结果加载到数据库
anysite dataset query <yaml> --sql "..."	用 SQL 查询已采集的数据
anysite dataset stats <yaml> --source <id>	单个数据源的按列统计信息
anysite dataset profile <yaml>	按数据源的质量摘要
anysite dataset validate <yaml>	验证配置但不采集（不发起 API 调用）
anysite dataset history <name>	查看运行历史
anysite dataset logs <name> --run	查看某次运行的日志
anysite dataset reset-cursor <yaml>	重置增量采集游标

后续步骤

Source Types

了解 5 种数据源类型：独立型（Independent）、文件型（From File）、依赖型（Dependent）、合并型（Union）和 LLM

Scheduling

设置增量采集、cron/systemd 条目和 Webhook 通知