源类型

用于构建灵活数据采集管道的六种数据源类型

概述

数据集管道支持 6 种数据源类型，每种都对应不同的数据采集模式。多种数据源可以组合起来构建复杂的多步骤工作流。下面的独立型（Independent）、文件型（From File）和依赖型（Dependent）本质上都是同一种底层 type: api 数据源（省略 type 时的默认值）— 它们的区别仅在于设置了哪些字段。

类型	用途	关键配置
独立型（Independent）	使用静态参数的单次 API 调用	`endpoint`、`params`
文件型（From File）	遍历文件进行批量调用	`from_file`、`input_key`
依赖型（Dependent）	使用父数据源的值进行批量调用	`dependency`、`input_key`
合并型（Union）	合并多个数据源的记录	`type: union`、`sources`
LLM	通过 LLM 模型处理数据	`type: llm`、`llm`
SQL	查询 Parquet 视图或指定的数据库连接	`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

适用场景： 管道的起点、搜索查询、单个资料查询。

文件型数据源（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

支持的文件格式：

TXT — 每行一个值
CSV — 使用与 input_key 匹配的列
JSON/JSONL — 使用与 input_key 匹配的字段

适用场景： 您已经有一份 URL、ID 或搜索词的列表。

依赖型数据源（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

依赖配置

字段	描述
`from_source`	父数据源的 ID
`field`	从父数据源结果中提取的字段路径（支持点号表示法）
`dedupe`	处理前去除重复值（默认：`false`）

简写形式： 对于单个依赖参数，在 params 中内联使用 ${source.field} 会自动展开为等价的 dependency/input_key 组合：

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

等价于上面显式的 dependency: {from_source: companies, field: urn.value} / input_key: companies 形式。同一个数据源上不能混用这两种写法。

支持多级链式依赖 — 一个依赖型数据源本身也可以作为另一个依赖型数据源的父数据源：

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

适用场景： 多步骤数据增强，从搜索结果到详细资料再到活动数据。

合并型数据源（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

合并配置

字段	描述
`type`	必须为 `union`
`sources`	要合并的数据源 ID 列表（必须共用同一个端点）
`dedupe_by`	用于去重的字段（可选）

适用场景： 合并多次搜索的结果，整合来自不同平台的数据。

LLM 数据源（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 操作

操作	描述
`classify`	将记录分类到预定义类别
`enrich`	提取新属性（枚举、字符串、布尔值、数字）
`summarize`	生成简洁摘要
`generate`	使用带字段占位符的模板生成文本

管道中的 LLM 步骤仅支持这 4 种操作 — match 和 deduplicate 只能作为独立的 anysite llm match / anysite llm deduplicate 命令使用，不能作为 type: llm 管道步骤使用。需要 llm 可选依赖包：pip install "anysite-cli[llm]"。

适用场景： 为管道添加 AI 驱动的增强能力，对采集到的数据进行分类或摘要。

SQL 数据源（SQL Source）

运行一次 SQL 查询，并将结果存储为它自己的数据源 — 既可以针对同一管道中已采集数据源的 Parquet 视图，也可以针对一个指定的数据库连接：

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（指向 .sql 文件的路径）代替内联的 query。省略 connection 表示查询该管道自身已采集的数据源。

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]

转换选项

字段	描述
`filter`	jq 风格的过滤表达式，保留匹配的记录
`fields`	要包含在输出中的字段列表
`add_columns`	为每条记录添加的静态列

导出选项

字段	描述
`type`	导出类型：`file` 或 `webhook`
`path`	输出文件路径（`file` 类型；支持 `{{date}}` 模板）
`format`	导出格式：`csv`、`json`、`jsonl`（默认：`jsonl`）
`url`	Webhook URL（`webhook` 类型）
`headers`	Webhook 的额外 HTTP 头

数据库加载选项（按数据源）

字段	描述
`key`	用于增量同步的唯一键列
`sync`	同步模式：`full`（默认，包含 DELETE）或 `append`（不执行 DELETE）
`fields`	要加载到数据库中的字段

输入模板

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

{value} 占位符会被依赖关系中的每个输入值替换。

通用数据源选项

以下选项适用于 type: api 数据源（独立型、文件型、依赖型）：

选项	描述	默认值
`parallel`	并发工作进程数量	`1`
`rate_limit`	最大请求速率（例如 `"10/s"`）	无限制
`on_error`	错误处理方式：`stop` 或 `skip`（数据源配置层级没有 `retry`）	`skip`
`refresh`	增量行为：`auto`、`always`	`auto`

后续步骤

Scheduling

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

Database Loading

将管道结果加载到 SQLite、PostgreSQL 或 ClickHouse