构建 Pipeline

Pipeline 将源表的区块数据交给 Function 计算，再把结果写入目标表。同一份配置可以执行历史回填，也可以持续处理新增区块。开始前，先按追踪流动性池储备变化创建状态表 demo.pool_reserves.eth 和 Function demo.read_pool_reserves。

配置触发源和计算函数

在 demo 中创建 Notebook，输入下面的脚本。源表中的 Sync 事件触发储备读取；ROW 把当前事件传给已保存的 Function。

from blockdb import Column
from blockdb.operator import filter
from blockx import Pipeline, ROW

SPACE = 'demo'

pipeline = Pipeline(
    sources=[{
        'table_id': 'chain.event.eth',
        'func_id': f'{SPACE}.read_pool_reserves',
        'args_tmpl': [ROW],
        'operator': filter(
            (Column('contract_id') == '0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc')
            & (Column('name') == 'Sync')
        ),
    }],
    target=f'{SPACE}.pool_reserves.eth',
)

pipeline.backfill(start=25939925, end=25939934)

配置	作用
`sources`	声明触发计算的源表，可以有多个触发源
`table_id`	触发源的完整表 ID
`func_id`	已保存的 Function ID
`args_tmpl`	位置参数模板；`ROW` 所在位置会被替换为源记录
`operator`	对源记录预先筛选；这里限定池地址和事件名
`target`	接收函数返回记录的目标表

示例 Function 使用事件的 block_id 查询合约，保证储备值属于该事件对应的区块状态。

运行历史回填

保存脚本，选择计算规格并点击 Run。backfill() 的起止高度都包含在处理范围内；本例处理十个区块。完成后查看 Notebook 输出：本例处理 10 个区块、写入 2 条记录。再打开目标表核对记录和已处理区间。

声明依赖表

若计算函数还会读取另一个区块状态表，把该表加入 requires。例如，读取一个自建区块状态表时：

pipeline = Pipeline(
    sources=sources,
    target='your_space.output.eth',
    requires=['your_space.reference_state.eth'],
)

将 sources 替换为已配置的触发源列表，两个表 ID 替换为实际表。依赖表不会自动成为函数参数，也不会单独触发计算；它要求对应高度的数据先准备好。函数仍需显式读取该表。

持续更新

在小范围回填验证通过后，将脚本最后一行改为：

pipeline.update()

update() 根据目标表的起始高度和已处理区间补齐缺口，然后持续处理新事件。

持续运行的配置见创建持续运行调度。停止维护时禁用调度，再检查其运行实例是否已经结束。

概览

快速开始

使用指南

参考

配置触发源和计算函数

运行历史回填

声明依赖表

持续更新