turbo-pipelines

Turbo 管道配置与架构

Turbo 管道的 YAML 配置参考与架构指南。交互式构建管道，请使用 /turbo-builder。排查问题，请使用 /turbo-doctor。实现转换，请使用 /turbo-transforms。

解码后的合约事件是转换的输出，不是可供使用的 <chain>.decoded_logs 数据集。 从 <chain>.raw_logs 获取 EVM 合约日志，然后在转换中解码（参见 /turbo-transforms）。预解码的代币数据集，例如 <chain>.erc20_transfers、<chain>.erc721_transfers 和 <chain>.erc1155_transfers，是独立且可用的数据集；请使用 /datasets 验证链的覆盖情况。

重要： 在向用户展示完整的管道 YAML 或部署之前，务必使用 goldsky turbo validate <file.yaml> 验证 YAML。

---

快速开始

name: my-first-pipeline
resource_size: s
sources:
  transfers:
    type: dataset
    dataset_name: base.erc20_transfers
    version: 1.2.0
    start_at: latest
transforms: {}
sinks:
  output:
    type: blackhole
    from: transfers

goldsky turbo validate pipeline.yaml   # Validate first
goldsky turbo apply pipeline.yaml -i   # Deploy + inspect

前置条件

• Goldsky CLI — 如果 goldsky 不在 PATH 中，请按照 /auth-setup 操作
• Turbo 扩展 — 按照 /auth-setup 操作，并使用 turbo 运行随附的安装程序。不要触发交互式自动安装程序。验证 goldsky turbo --version 能否成功运行；已发布的 Linux 二进制文件要求 x64 和 glibc 2.39+，Mac 二进制文件要求 Apple Silicon。Windows 使用 WSL 入口。平台不受支持不代表安装成功。
• Shell PATH — 在每次新的 Shell/工具调用中恢复 export PATH="$HOME/.local/bin:$HOME/.goldsky/bin:$PATH"；导出设置不会在调用之间保留。
• 已登录 — 如果 goldsky project list 未通过身份验证，请按照 /auth-setup 操作并自行运行 goldsky login
• 如果使用 PostgreSQL、ClickHouse、Kafka 等作为 sink，请为其创建密钥（参见 /secrets）

---

架构决策

源类型选择

场景	源类型	原因
从日志解码合约事件	dataset	需要 raw_logs + _gs_log_decode()
跟踪代币转账	dataset	erc20_transfers 包含结构化数据
历史回填 + 实时数据	dataset	start_at: earliest 会处理历史数据
实时代币余额	kafka	latest_balances_v2 是流式主题
实时状态快照	kafka	Kafka 持续提供最新状态
只需要今后新增的数据	二者皆可	使用 start_at: latest 的数据集或 Kafka

注意： Kafka 源在生产环境中会使用，但未记录在 Goldsky 官方文档中。请联系 Goldsky 支持团队获取主题名称。

数据流模式

模式	适用场景	模板
线性	单个源、单个目标，处理简单	templates/linear-pipeline.yaml
扇出	一个源 → 多个 sink（不同视图/子集）	templates/fan-out-pipeline.yaml
汇入	多种事件类型 → UNION ALL → 一张表	templates/fan-in-pipeline.yaml
多链	在多条链上使用相同逻辑（分别创建管道）	templates/multi-chain-templated.yaml

如需了解详细的模式图、YAML 示例和多链部署指南，请阅读 references/architecture-patterns.md。

资源规格

规格	工作进程	CPU	内存	适用场景
xs	—	0.4	0.5 Gi	小型数据集、轻量测试
s	1	0.8	1.0 Gi	简单筛选、单个数据源／接收端、低吞吐量（默认）
m	4	1.6	2.0 Gi	多个接收端、Kafka 流处理、中等复杂度的转换
l	10	3.2	4.0 Gi	多事件解码加 UNION ALL、高吞吐量回填
xl	20	6.4	8.0 Gi	大型链回填、复杂 JOIN
xxl	40	12.8	16.0 Gi	最高吞吐量；最高可达 630 万行／分钟

从小规格开始，再逐步扩容——采用保守的规格设置可避免浪费资源。

resource_size 设置的是管道规格，而不是接收端规格——两者相互独立。 将 l／xl 规格的管道指向规格不足的目标端，会出现“已部署但没有写入”的情况：验证通过、部署成功、状态显示为 Running，之后每次写入都会失败。Neon 免费套餐（512 MB）无法容纳高吞吐量链数月的回填数据——它会报错 could not extend file because project size limit (512 MB) has been exceeded，检查点也会超时。部署前就应根据回填范围（起始位置加上任何结束边界）检查目标端的存储上限，而不是等部署后再检查。

接收端选择

目标端	接收端类型	最适用场景
应用数据库	postgres	行级查询、联接、为应用提供服务
实时聚合	postgres_aggregate	通过触发器计算余额、计数器、累计总数
兼容 MySQL 的数据库	mysql	现有 MySQL 技术栈、基于主键的更新插入
分析查询	clickhouse	大规模聚合、时间序列数据
事件处理	kafka	下游消费者、事件驱动系统
AWS 消息传递	sqs_sink	解耦的 AWS 消费者（仅标准队列）
GCP 消息传递	pubsub	Google Cloud Pub/Sub 主题（仅限 Turbo）
无服务器流处理	s2_sink	S2.dev 流，可作为 Kafka 的替代方案
通知	webhook	Lambda 函数、API 回调、告警
数据湖	s3_sink	长期归档、批处理
测试	blackhole	验证管道，但不写入数据

如需查看接收端字段的完整说明，请参阅文档中的 接收端（每种接收端类型各有一个页面）。

流处理模式与作业模式

场景	模式	原因
实时仪表板	流处理	需要持续更新
回填 6 个月的历史数据	作业	一次性运行，完成后停止
实时处理并在部署时追赶进度	流处理	start_at: earliest 会先回填，再开始流处理
一次性将数据导出到 S3	作业	无需持续处理
事件发生时发送 webhook 通知	流处理	需要在事件发生时作出响应
使用历史数据进行负载测试	作业	处理并检查数据，然后丢弃

作业模式规则： 运行至完成，完成约 1 小时后自动删除。重新部署前必须先删除。无法暂停、恢复或重启。

流水线拆分

默认使用一个流水线和多个接收端，适用于用户希望将相同的源数据发送到多个目标的情况。单个 Turbo 流水线原生支持多个接收端——每个接收端都有一个 from: 字段，指向按名称指定的源或转换步骤，并且接收端独立运行（一个接收端失败不会阻塞其他接收端，且每个接收端都可以有自己的 batch_size / batch_flush_interval）。

不要仅仅因为有多个目标就拆分成多个流水线。 为每个接收端生成一个流水线会重复摄取源数据、浪费资源，并使本应原子化的部署彼此脱节。

仅在以下情况下拆分为多个流水线： 源本质上不同（生命周期独立的不同链）、目标需要不同的资源规格，或用户明确要求使用单独的流水线。

有关扇出（一个源 → 多个接收端）和扇入（UNION ALL）的 YAML 示例，请参阅 references/architecture-patterns.md；可直接改编的配置请参阅 templates/multi-sink-pipeline.yaml / templates/fan-out-pipeline.yaml / templates/fan-in-pipeline.yaml。

---

配置参考

流水线结构

name: my-pipeline          # Required: unique identifier (lowercase, hyphens)
resource_size: s            # Required: xs/s/m/l/xl/xxl
job: true                   # Optional: one-time batch (default: false = streaming)

sources:
  source_name:
    type: dataset           # or: kafka
    # ... source config

transforms:                 # Optional
  transform_name:
    type: sql               # or: script, handler, dynamic_table
    # ... transform config

sinks:
  sink_name:
    type: postgres           # or: clickhouse, kafka, webhook, s3_sink, etc.
    # ... sink config

源配置

数据集源

sources:
  my_source:
    type: dataset
    dataset_name: <chain>.<dataset_type>
    version: <version>
    start_at: latest | earliest    # EVM, NEAR, Bitcoin, Stellar
    # start_block: <slot_number>   # Solana only — omit to start at the latest slot
    # end_block: <block_number>    # Solana only: bounded backfill (ignored on EVM)
    # filter: >-                   # Optional: SQL WHERE for source-level pre-filtering
    #   address = '0x...' AND block_number >= 10000000

字段	必填	说明
type	是	区块链数据使用 dataset
dataset_name	是	格式：<chain>.<dataset_type>
version	是	数据集版本（例如 1.2.0）
start_at	EVM 系列	latest 或 earliest；在 Stellar 上也可以是账本序列号。适用于 EVM、NEAR、Bitcoin、Stellar。省略 = earliest = 完整链历史
start_block	否	仅适用于 Solana：起始 slot。省略 = 最新 slot，即不回填
end_block	否	仅适用于 Solana：在此 slot 停止。在 EVM 上会被静默忽略
filter	否	SQL WHERE 子句——在摄取时预先过滤（高效）

使用 filter 设置合约地址和区块范围（粗略预过滤）。使用转换步骤中的 WHERE 进行精细过滤。

为每个数据集源明确设置起始位置，并说明选择了哪个位置。对于使用 start_at 的链（EVM、NEAR、Bitcoin、Stellar），省略它并不意味着“现在开始”：后端会从最早的可用数据开始，因此流水线会回填整条链的历史数据——在到达实时数据前，可能需要数天重放并处理数百万行，而接收端必须容纳所有数据。区块号不是有效的 start_at 值；要限定 EVM 范围，请在 filter 中使用 block_number 谓词（在源端预先应用），并设置 job: true 以进行一次性回填——end_block 在 EVM 上会被静默忽略。Solana 的默认行为正好相反：它使用数字形式的 start_block，省略时会从最新 slot 开始，因此 Solana 回填始终是用户主动要求的——使用 end_block 或 block_ranges 限定范围。

有关链前缀和数据集类型，请参阅 /datasets。

Kafka 源

sources:
  my_source:
    type: kafka
    topic: base.raw.latest_balances_v2

没有 start_at 或 version 字段。可选项：filter、include_metadata、starting_offsets。

转换配置

类型	用途
sql	过滤、投影、SQL 函数
script	自定义 TypeScript/WASM 逻辑
handler	调用外部 HTTP API 进行数据丰富
dynamic_table	由数据库支持的查找表
throttle	将吞吐量限制为固定的每秒记录数

SQL Transform

transforms:
  filtered:
    type: sql
    primary_key: id
    sql: |
      SELECT id, sender, recipient, amount
      FROM source_name
      WHERE amount > 1000

字段	必填	描述
type	是	sql
primary_key	是	用于唯一性判定和排序的列
sql	是	SQL 查询（按名称引用数据源）
from	否	覆盖默认数据源（用于链式处理）

Transform 链式处理

使用 from 链接 transforms：

transforms:
  step1:
    type: sql
    primary_key: id
    sql: SELECT * FROM source WHERE amount > 100
  step2:
    type: sql
    primary_key: id
    from: step1
    sql: SELECT *, 'processed' as status FROM step1

有关 TypeScript、handler 和动态表 transforms，请参阅 /turbo-transforms。

Throttle Transform

通过将记录缓冲为批次，并按固定的最小时间间隔发出每个批次，限制流的吞吐量。Throttle 不会修改数据——每条输入记录都会原样通过。可使用它避免超过下游 sink 或外部 API 的速率限制、平滑突发数据源，或控制记录进入 HTTP handlers 的速率。

transforms:
  throttled:
    type: throttle
    from: my_source
    max_batch_size: 100
    min_batch_interval: 10s

字段	必填	描述
type	是	throttle
from	是	要读取的数据源或 transform
max_batch_size	否	每个批次的最大记录数
min_batch_interval	否	批次之间的最小时间间隔（例如 10s、500ms、1m）

有效的最大吞吐量 ≈ max_batch_size / min_batch_interval 条记录/秒。Throttle 限制的是最大速率，而不是最小速率——如果上游较慢，批次会更小，发送频率也会更低。

将 throttle 放在靠近瓶颈的位置（紧靠受速率限制的 sink 或 handler 之前），这样上游 transforms 仍可全速处理。

Sink 配置

常见 sink 的快速示例。有关所有 sink 类型的完整字段规范，请参阅文档中的 Sinks（每种 sink 类型单独占一页）。

primary_key 的位置取决于 sink 类型，并非仅用于 transforms。 SQL transform 的键描述其输出，而 sink 的键控制目标端行为：

Turbo sink	Sink 级别的 primary_key
PostgreSQL	可选；启用 upsert。不填写则执行普通插入。
MySQL	可选；启用 upsert。不填写则执行普通插入。
ClickHouse	必填；设置表排序和去重列。
Kafka	可选；选择消息键列。如果未填写，则在上游键可用时使用该键。

对于其他 sink 类型，添加或移除 primary_key 前请先查看对应的字段参考。Mirror 使用不同的 sink schema；不要将 Turbo sink 字段复制到 Mirror 配置中。

PostgreSQL

sinks:
  output:
    type: postgres
    from: my_transform
    schema: public
    table: my_table
    secret_name: MY_POSTGRES_SECRET
    primary_key: id

ClickHouse

sinks:
  output:
    type: clickhouse
    from: my_transform
    table: my_table
    secret_name: MY_CLICKHOUSE_SECRET
    primary_key: id

Pub/Sub（仅 Turbo）

sinks:
  output:
    type: pubsub
    from: my_transform
    topic: my-topic
    secret_name: MY_PUBSUB_SECRET

密钥包含 GCP 项目 ID 和服务账号 JSON。主题必须已存在于 GCP 中。

Blackhole（测试）

sinks:
  output:
    type: blackhole
    from: my_transform

---

检查点行为

• 更新 pipeline 时，默认会保留检查点
• 检查点与数据源名称绑定——重命名数据源会重置其检查点
• 检查点与 pipeline 名称绑定——重命名 pipeline 会重置所有检查点

要重置检查点：重命名数据源或 pipeline。警告：这会重新处理所有历史数据。

---

入门模板

模板	描述	使用场景
minimal-erc20-blackhole.yaml	最简单的管道，无需凭据	快速测试
filtered-transfers-sql.yaml	按合约地址筛选	USDC、特定代币
postgres-output.yaml	写入 PostgreSQL	生产环境数据存储
pubsub-output.yaml	写入 Google Cloud Pub/Sub	基于 GCP 的事件消费者
multi-chain-pipeline.yaml	组合多个链	跨链分析
solana-transfers.yaml	Solana SPL 代币	非 EVM 链
multi-sink-pipeline.yaml	多个输出	归档 + 告警 + 流式处理
linear-pipeline.yaml	简单的解码 → 筛选 → 接收器	基本线性流程
fan-out-pipeline.yaml	一个源 → 多个接收器	多目标
fan-in-pipeline.yaml	多个事件 → UNION ALL	活动流
multi-chain-templated.yaml	按链配置管道的模式	独立部署各链

模板位置： templates/（相对于此技能目录）

---

CLI 快速参考

操作	命令
安装 Goldsky CLI	按照 /auth-setup 操作并使用其捆绑的安装程序
安装 Turbo 扩展	按照 /auth-setup 操作，并使用其捆绑的安装程序运行 turbo；要求版本检查成功
验证（必需）	goldsky turbo validate pipeline.yaml
部署/更新	goldsky turbo apply pipeline.yaml
部署并检查	goldsky turbo apply pipeline.yaml -i
列出管道	goldsky turbo list
列出数据集	goldsky dataset list --output json
列出密钥	goldsky secret list

有关生命周期命令（暂停/恢复/重启/删除）和监控（检查/日志），请参阅 /turbo-operations。

---

故障排除

CLI 挂起、验证错误和运行时错误，请参阅 references/troubleshooting.md。

---

相关内容

• /turbo-builder — 交互式向导，逐步构建管道
• /turbo-doctor — 诊断并修复管道问题
• /turbo-operations — 生命周期命令和监控参考
• /turbo-transforms — SQL、TypeScript 和动态表转换参考
• /datasets — 数据集名称和链前缀
• /secrets — 接收器凭据管理