建你的第一条索引管道

这一页做一件具体的事：从公共表 chain.trace.eth（内部调用记录）里解析出 ETH 转账，写进你自己的一张新表，然后让它跟着新块持续更新。

全程都在浏览器里操作。Chaintable 的 Python SDK 只在平台的 Notebook 运行时里可用，不需要在本地安装任何东西。

建一张目标表

进入你的 Space，在左侧 Tables 列表里点 New Table，弹窗里填：

Table Type 选 Block Event（这次要写的是”发生过的事情”——一笔转账一旦发生就不会变，符合 Event 表的语义）
Chain 选 eth
Table name 填 transfers

点 Save。表名会自动变成 <你的space>.transfers.eth，并直接跳进这张新表的详情页——现在是空的，后面几步会把数据填进来。

New Table 弹窗：Table Type 选 Block Event，Chain 选 eth

表类型怎么选、四种类型的区别，见选哪种表类型。

定义列

新表现在是空的，详情页会提示”当前没有内容，请先 Edit Columns”——这一步是必须的，不是可选项：目标表没有初始化过列的话，下面组装 Pipeline 那一步会直接报错。点 Edit Columns。Block Event 类型的表自带几个系统列（block_height、block_id、block_timestamp，还有一个默认隐藏的 _updated_at），保留它们不用管；再点 Add Column，把转换函数 to_transfer() 会返回的几个字段加上：

列名	类型	说明
`id`	String	必须叫这个名字——平台按列名认主键，类型要选 String / Address / Hash 这类”字符串家族”
`from_addr`	Address
`to_addr`	Address
`value`	Float
`tx_id`	TxId

保存。存完这一步表才算初始化完成，才能开始写数据。

Edit Columns 面板 — Edit Columns：加上 id / from_addr / to_addr / value / tx_id

列名要跟转换函数 return 的 dict key 完全对上。block_height / block_id / block_timestamp 这几个系统列由平台按当前处理的区块自动填，函数不用返回它们。完整的类型清单和加改列的细节见列与数据类型。

打开一个 Notebook

回到 Space，在左侧 Notebooks 列表里点 New Notebook，打开一个空白编辑器。

Chaintable 的 Notebook 不是 Jupyter 那种一个一个 cell 的模型——一个 Notebook 就是一份完整的 Python 脚本。编辑器上方有个 Run 按钮，点一下就把当前整份代码作为一次任务提交去跑，输出实时显示在下面的 Output 面板里（旁边还有 Resources 看资源用量、BlockX 看提交的计算任务）。改完代码想再跑一次，还是点 Run，会重新整份执行一遍，不是接着上次的状态继续。

Notebook 是驱动，负责组装任务并提交给平台；实际的转换逻辑是一个单独的 Python 函数。

新建的空白 Notebook 编辑器：上方 Run 按钮，下方 Output 面板

写转换逻辑，跑一次小范围回填

把下面这段代码整个粘贴进编辑器。Pipeline 是官方推荐的入口——只需要描述”源表 → 函数 → 目标表”，调度、并发、重试都由它处理；最后一行先用 backfill 对一小段历史区块跑一次，确认转换逻辑是对的。这里特意选了正好 1000 块（一个 bundle 的大小），实际大规模回填走的也是这条路径，跟这次验证是同一套逻辑：

from blockx import Pipeline, Trigger, SOURCE_ROW

def to_transfer(record, chain_id):
    # record 是源表这一行的数据（dict）；返回 dict 会被写入目标表，返回 None 会被丢弃
    value = float(record.get("value") or 0)
    if value <= 0 or not record.get("tx_id"):
        return None
    return {
        "id": record["id"],
        "from_addr": record["from_addr"],
        "to_addr": record["to_addr"],
        "value": value,
        "tx_id": record["tx_id"],
    }

pipe = Pipeline(
    triggers=[Trigger(table="chain.trace.eth", func=to_transfer, params=[SOURCE_ROW, "eth"])],
    target_table="<你的space>.transfers.eth",
)

result = pipe.backfill(block_start=25_000_001, block_end=25_001_000)

点 Run。跑完 Output 面板里会打印一份汇总：提交了多少块、失败了多少、写了多少行。

Output 面板：backfill 跑完打印的汇总

转换函数跑在隔离沙箱里：不能引用函数体外的全局变量，函数里用到的模块要在函数体内重新 import。完整规则见沙箱限制。

“已提交”不等于”已落库”。 汇总里的成功数只代表任务被 worker 受理，真正写库是后台异步完成的，去下一步的表详情页确认。

回到表详情页，确认数据真的写进去了

切回 transfers 表的详情页（或者刷新），打开数据预览，应该能看到刚才回填的数据——这 1000 个区块里但凡有 ETH 转账，都会有对应的行。

transfers 表详情页：数据预览里能看到刚回填的数据

挂上去，让它一直跑

验证没问题后，回 Notebook 把最后一行换成：

pipe.update()  # 长驻不退出

再点一次 Run——这次会重新整份执行：重新定义 pipe，然后跑 update()。启动时会自动把目标表补到当前高度，之后转入订阅、跟着新块持续处理，这次运行会一直占着不返回。只要这次运行还在你的会话里，关掉浏览器标签页它就会停。想让它甩开浏览器持续跑，可以点 Switch to Background（一次性操作，点了就不能切回来）；要用于生产、需要崩溃自动重启这种更强的保障，就给这个 Notebook 挂一个常驻调度，操作方式见调度：常驻 / 定时 / 手动。

刚才发生了什么

Pipeline 内部替你做了这些事：历史区间按 bundle（1000 块一组）批量回填，末尾不满 1000 块的零头才会退化成逐块回填——刚才那 1000 块正好是一个完整 bundle，所以这次跑的就是真实大规模回填会走的那条路径，不是简化版。转入实时后，通过订阅跟着新块跑，不会重复处理已经覆盖过的高度。这些都不需要你操心，bundle 回填还有一些需要注意的地方（比如它不会触发下游订阅事件），见用 Pipeline 跑起来。

接下来

给这张表补更早的历史

教程 2：把 backfill 的范围扩大到全部历史高度

转换逻辑需要查另一张表

教程 3：把 raw amount 换算成人类可读金额

Function 的完整规则

参数、沙箱限制、调试方式

Pipeline 的底层原理

如果你需要比 Pipeline 更精细的控制

​刚才发生了什么

​接下来

给这张表补更早的历史

转换逻辑需要查另一张表

Function 的完整规则

Pipeline 的底层原理

刚才发生了什么

接下来