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 のエントリーポイントを使用します。サポート対象外のプラットフォームでは、インストールに成功したことになりません。
• シェルの PATH — 新しいシェル／ツール呼び出しごとに export PATH="$HOME/.local/bin:$HOME/.goldsky/bin:$PATH" を設定し直してください。export は呼び出し間で保持されません。
• ログイン済みであること — goldsky project list が未認証の場合は、/auth-setup に従って goldsky login を自分で実行してください
• PostgreSQL、ClickHouse、Kafka などを使用する場合は、シンク用のシークレットを作成してください（/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
ファンアウト	1 つのソース → 複数のシンク（異なるビュー／サブセット）	templates/fan-out-pipeline.yaml
ファンイン	複数のイベントタイプ → UNION ALL → 1 つのテーブル	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	最大スループット。最大 6.3M 行／分

まずは小さく始めて、必要に応じてスケールアップしてください。余裕を持ったサイズ設定により、リソースの浪費を避けられます。

resource_size が設定するのはパイプラインのサイズであり、シンクのサイズではありません。この2つは独立しています。 サイズ不足の宛先に l／xl のパイプラインを向けると、「デプロイ済みなのに何も書き込まれない」状態になります。検証とデプロイが完了して Running と報告された後、すべての書き込みが失敗します。Neon の無料枠（512 MB）では、高ボリュームのチェーンの数か月分のバックフィルを保持できません。could not extend file because project size limit (512 MB) has been exceeded というエラーが発生し、チェックポイントがタイムアウトします。デプロイ前に、バックフィルの範囲（開始位置と終了境界がある場合はその両方）に対して、宛先のストレージ上限が十分か確認してください。デプロイ後ではなく、デプロイ前に確認することが重要です。

シンクの選択

宛先	シンクタイプ	最適な用途
アプリケーション DB	postgres	行単位の検索、結合、アプリケーションへのデータ提供
リアルタイム集計	postgres_aggregate	トリガーを介した残高、カウンター、累計値
MySQL 互換 DB	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	データを書き込まずにパイプラインを検証

シンクのフィールド仕様の詳細については、ドキュメントのシンク（シンクタイプごとに1ページ）をご覧ください。

ストリーミングモードとジョブモード

シナリオ	モード	理由
リアルタイムダッシュボード	ストリーミング	継続的な更新が必要
6か月分の履歴をバックフィル	ジョブ	1回限りの実行で、完了すると停止
リアルタイム処理とデプロイ時のキャッチアップ	ストリーミング	start_at: earliest はバックフィル後にストリーミングを開始
データを S3 に1回エクスポート	ジョブ	継続的な処理は不要
イベント発生時の Webhook 通知	ストリーミング	イベントの発生に応じた処理が必要
履歴データを使ったロードテスト	ジョブ	処理して確認した後、破棄

ジョブモードのルール： 完了するまで実行され、終了から約1時間後に自動削除されます。再デプロイする前に削除する必要があります。一時停止、再開、再起動はできません。

パイプラインの分割

ユーザーが同じソースデータを複数の宛先に送信したい場合は、複数のシンクを持つ1つのパイプラインをデフォルトとしてください。1つの Turbo パイプラインは複数のシンクをネイティブにサポートしています。各シンクには、名前で指定したソースまたは変換を指す from: フィールドがあり、シンクはそれぞれ独立して実行されます（1つが失敗してもほかのシンクはブロックされず、それぞれに独自の batch_size / batch_flush_interval を設定できます）。

宛先が複数あるという理由だけで、パイプラインを複数に分割しないでください。 シンクごとにパイプラインを生成すると、ソースの取り込みが重複してリソースを浪費し、本来は一体としてデプロイすべきものが切り離されます。

パイプラインを複数に分割するのは、次の場合に限ってください。 ソースが根本的に異なる（ライフサイクルが独立した別々のチェーンである）、宛先ごとに異なるリソースサイズが必要である、またはユーザーがパイプラインの分割を明示的に求めている場合です。

ファンアウト（1つのソース → 複数のシンク）とファンイン（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 のみ：開始スロット。省略すると最新スロットになり、バックフィルは行われません
end_block	いいえ	Solana のみ：このスロットで停止します。EVM では無視されます
filter	いいえ	SQL の WHERE 句。取り込み時に事前フィルタリングします（効率的）

コントラクトアドレスやブロック範囲の指定（大まかな事前フィルタリング）には filter を使用してください。細かなフィルタリングには、変換の WHERE を使用してください。

すべてのデータセットソースで開始位置を明示的に設定し、どの値を選んだかを記載してください。start_at を使用するチェーン（EVM、NEAR、Bitcoin、Stellar）では、これを省略しても「現在から開始する」という意味にはなりません。バックエンドは利用可能な最も古いデータから開始するため、パイプラインはチェーンの全履歴をバックフィルします。ライブデータに到達するまでに数日分の再生と数百万行の処理が発生し、シンクはそのすべてを保持する必要があります。ブロック番号は有効な start_at 値ではありません。EVM の範囲を指定するには、ソースで事前適用される filter に block_number の条件を含め、1 回限りのバックフィルには job: true を指定してください。EVM では end_block は無視されます。一方、Solana のデフォルトは逆です。数値の start_block を使用し、省略すると最新スロットから開始します。そのため、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 Chaining

from を使用して変換をチェーンします。

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、dynamic table の変換については、/turbo-transforms を参照してください。

Throttle Transform

レコードをバッチにバッファリングし、一定の最小間隔で各バッチを出力することで、ストリームのスループットに上限を設けます。Throttle はデータを変更しません。すべての入力レコードは変更されずに通過します。下流のシンクや外部 API のレート制限を超えないようにする場合、バーストの多いソースを平滑化する場合、または HTTP ハンドラーにレコードを一定のペースで送る場合に使用します。

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

フィールド	必須	説明
type	はい	throttle
from	はい	読み取り元のソースまたは変換
max_batch_size	いいえ	バッチあたりの最大レコード数
min_batch_interval	いいえ	バッチ間の最小時間（例：10s、500ms、1m）

実効最大スループット ≈ max_batch_size / min_batch_interval（レコード／秒）です。Throttle は最大レートを制限するもので、最小レートを保証するものではありません。上流の処理が遅い場合、バッチは小さくなり、到着頻度も低くなります。

**ボトルネックの近く（レート制限のあるシンクまたはハンドラーの直前）に throttle を配置してください。**これにより、上流の変換は引き続き最大速度で処理できます。

シンクの設定

一般的なシンクの簡単な例です。すべてのシンクタイプのフィールド仕様については、ドキュメントの Sinks（シンクタイプごとに 1 ページ）を参照してください。

primary_key の配置場所はシンクタイプによって異なり、変換に限定されません。 SQL 変換のキーはその出力を示し、シンクのキーは出力先での動作を制御します。

Turbo シンク	シンクレベルの primary_key
PostgreSQL	任意。指定すると upsert が有効になります。通常の insert の場合は省略します。
MySQL	任意。指定すると upsert が有効になります。通常の insert の場合は省略します。
ClickHouse	必須。テーブルの順序付けと重複排除に使用するカラムを設定します。
Kafka	任意。メッセージキーのカラムを選択します。省略した場合、利用可能であれば上流のキーが使用されます。

その他のシンクタイプについては、primary_key を追加または削除する前に、それぞれのフィールドリファレンスを確認してください。Mirror は異なるシンクスキーマを使用します。Turbo のシンクフィールドを 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

---

チェックポイントの動作

• パイプラインの更新時、デフォルトで保持されます
• ソース名に紐付けられます。ソース名を変更すると、そのチェックポイントはリセットされます
• パイプライン名に紐付けられます。パイプライン名を変更すると、すべてのチェックポイントがリセットされます

チェックポイントをリセットするには、ソース名またはパイプライン名を変更します。警告：これにより、すべての過去データが再処理されます。

---

スターターテンプレート

テンプレート	説明	ユースケース
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	1 つのソース → 複数のシンク	複数の出力先
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 — シンクの認証情報管理