> ## Agent Instructions
>
> Base URL: https://api.anysite.io
> Authentication: send the `access-token` header. Do NOT use `Authorization: Bearer`.
> Full endpoint catalog: https://app.anysite.io/docs
# 数据集管道

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

## 概述

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

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

## 创建管道

```bash
anysite dataset init my-dataset
```

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

```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/
```

## 运行采集

```bash
# 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
```

## 多数据源管道示例

```yaml
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](/docs/cli/datasets/source-types?lang=zh)） |
| `storage.format` | 会被接受但对原始采集步骤不生效 — 此 CLI 版本中，采集到的数据始终以 Parquet 格式写入 |
| `storage.path` | 输出目录（默认：`./data/`） |
| `schedule.cron` | 供 `dataset schedule` 使用的 cron 表达式（见 [Scheduling](/docs/cli/datasets/scheduling?lang=zh)） |
| `notifications` | 用于成功/失败事件的 Webhook URL |

单个数据源的 `export.format`（`csv`、`json` 或 `jsonl`）*确实*已经实现 — 见
[Per-Source Transform & Export](/docs/cli/datasets/source-types?lang=zh#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 通知