Pineapple：用数据冒险模型自动构建 DAG 的高性能流水线引擎

搜索、推荐、广告——这类在线系统的后端往往需要一条多步骤数据处理流水线：召回候选集、特征计算、过滤、排序、截断……步骤之间存在复杂的数据依赖，手动编排执行顺序既繁琐又容易出错。Pineapple 的回答是：让算子只声明「我读什么、写什么」，引擎借鉴 CPU 流水线的数据冒险分析，自动推导依赖、构建 DAG、并行调度。Python 声明，Go 执行，JSON 解耦——三条线各司其职，业务迭代不需要重编译 Go 代码，Go 服务自动热加载配置变更。

本文将从架构设计、DAG 构建算法、并行调度机制、Lua 嵌入层、控制流编译等维度，完整拆解 Pineapple 的核心原理，并附上真实的 benchmark 数据。

架构：Python 声明，Go 执行，JSON 解耦

Pineapple 由两个独立组件构成：

• Apple（Python）——声明式 DSL，面向业务团队。运行后产出 JSON 配置文件，不参与运行时计算。
• Pine（Go）——执行引擎，面向工程团队。解析 JSON 配置，构建 DAG，并行调度算子。

二者之间唯一的持久边界是 JSON 配置。这意味着业务团队修改 Python 流水线后，只需重新生成 JSON，Go 服务自动热加载，无需重编译、无需重启。

Python DSL  ──(compile)──>  JSON 配置
                                │
                                v
                 Go 引擎解析 JSON，推导算子依赖
                                │
                                v
                  构建 DAG，拓扑排序，并行执行

Pine 被定位为纯计算库——整个引擎只暴露两个 API：

engine, err := pine.NewEngine(jsonConfig)  // 编译：JSON → 不可变执行计划
result, err := engine.Execute(ctx, req)    // 执行：请求进，结果出

NewEngine 之后 Engine 不可变，可被任意数量的 goroutine 并发调用 Execute。配置重载？创建新 Engine，通过 atomic.Pointer 原子替换，旧实例由 GC 回收。没有 Reload 方法，没有读写锁，读路径零锁竞争。

这种极简 API 设计意味着 Pine 可以被嵌入到任何部署形态——HTTP 服务、gRPC 服务、离线批处理 Runner——而不需要任何修改。

六种算子类型

Pineapple 将所有算子严格分为六种类型，每种类型决定了允许的输出方法和 DAG 语义：

类型	预期角色	允许的输出方法	DAG 语义
Recall	产生新行	AddItem	追加型写者，多 Recall 可并行
Transform	变异字段值	SetCommon、SetItem	字段级 RAW/WAW/WAR 追踪
Filter	移除行	RemoveItem	屏障
Merge	合并/去重	SetItem、RemoveItem	屏障
Reorder	改变顺序	SetItemOrder	屏障
Observe	只读副作用	无	只读，不阻塞下游

类型约束在运行时强制执行——算子返回后，调度器检查 OperatorOutput 中实际调用了哪些方法，不符合类型约束直接返回 error。这不是约定，是硬校验。

数据冒险驱动的 DAG 构建

这是 Pineapple 最核心也最独特的设计。

传统的 DAG 引擎需要用户手动声明算子之间的依赖关系，或者按照拓扑层级手动编排。Pineapple 的做法不同：算子只需声明自己读写哪些字段，引擎自动推导依赖。这个推导算法直接借鉴了计算机体系结构中 CPU 流水线的**数据冒险（data hazard）**模型。

三种数据冒险

// internal/dag/dag.go

// Phase 2: Apply data hazards for common and item fields separately
addEdges(g, sequence, operators, true)  // common fields
addEdges(g, sequence, operators, false) // item fields

引擎对 common 字段和 item 字段分别执行一遍数据冒险分析，追踪三种冒险类型：

冒险类型	含义	建边规则
RAW（Read-After-Write）	读者必须等待写者完成	reader → 依赖 writer
WAW（Write-After-Write）	后一个写者必须等待前一个写者	writer₂ → 依赖 writer₁
WAR（Write-After-Read）	写者必须等待所有先序读者完成	writer → 依赖 readers

对于学过计算机体系结构的读者，这三种冒险应该很熟悉——它们正是经典五级流水线需要处理的数据相关性。Pineapple 将这套模型从指令级搬到了算子级。

逐字段追踪器

每个字段独立维护一个追踪器：

type fieldTracker struct {
    lastMutWriter   int   // 最近一个覆写型写者
    additiveWriters []int // 追加型写者（recall）
    activeReaders   []int // 活跃读者
}

关键的精妙之处在于区分了两种写入语义：

• 覆写型写者（mutating writer）：Transform 的 SetItem，修改已有行，产生完整的 WAW 和 WAR 冲突。
• 追加型写者（additive writer）：Recall 的 AddItem，追加新行，不影响已有行。

追加型写者之间不存在 WAW/WAR 冲突——这正是多路召回可以完全并行的理论基础。引擎不需要用户声明「这几个 Recall 可以并行」，数据冒险分析自动得出这个结论。

isAdditiveWrite := !isCommon && opType == types.OpTypeRecall

if isAdditiveWrite {
    // 追加型：只记录为 additive writer，不建 WAW/WAR 边
    ft.additiveWriters = append(ft.additiveWriters, i)
} else {
    // 覆写型：完整的 WAW + WAR 处理
    if ft.lastMutWriter >= 0 {
        addEdge(g, ft.lastMutWriter, i) // WAW
    }
    for _, reader := range ft.activeReaders {
        addEdge(g, reader, i) // WAR
    }
    ft.lastMutWriter = i
    ft.additiveWriters = nil
    ft.activeReaders = nil
}

屏障语义

Filter、Merge、Reorder 这三种算子会改变 item 集合的「结构」（组成或顺序），因此被标记为屏障（barrier）。屏障算子的语义很简单：所有前序算子必须执行完毕，屏障执行完毕后后续算子才能开始。

func addBarrierEdges(g *Graph, sequence []string, operators map[string]config.OperatorConfig) {
    for i, name := range sequence {
        if !opType.IsBarrier() {
            continue
        }
        for j := 0; j < i; j++ {
            addEdge(g, j, i) // 所有前序 → 屏障
        }
        for j := i + 1; j < n; j++ {
            addEdge(g, i, j) // 屏障 → 所有后序
        }
    }
}

这等价于经典并行计算中的 barrier synchronization 原语。

行集合级别的依赖

有些算子需要「所有 item 都就绪」才能执行——比如统计 item 总数的 transform_size。Pineapple 通过一个虚拟的哨兵字段 _row_set_ 来建模这种行集合级因果关系：

const rowSetSentinel = "_row_set_"

if isAdditiveWrite {
    writeFields = append(writeFields, rowSetSentinel) // Recall 是 _row_set_ 的追加型写者
}
if opCfg.RowDependency {
    readFields = append(readFields, rowSetSentinel)   // 声明了行依赖的算子是 _row_set_ 的读者
}

这个设计的巧妙之处在于：行集合依赖完全复用了已有的 RAW/WAW/WAR 机制，没有引入任何新的推导逻辑。Recall 对 _row_set_ 做追加型写入，需要完整行集合的算子声明读取 _row_set_，标准的 RAW 依赖自然保证了执行顺序。

Channel-per-Node 并行调度

DAG 构建完成后，调度器的实现出人意料地简洁。核心思想只有一句话：为每个算子节点创建一个 channel，节点完成时 close 该 channel，后继节点通过 select 等待所有前驱 channel。

func Run(ctx context.Context, plan *Plan, frame *dataframe.Frame, stats *Stats) ([]Warning, []types.OpTrace, error) {
    n := len(plan.Graph.Nodes)
    done := make([]chan struct{}, n)
    for i := 0; i < n; i++ {
        done[i] = make(chan struct{})
    }

    for i := 0; i < n; i++ {
        go func(idx int) {
            defer close(done[idx])

            // 等待所有前驱完成
            for _, pred := range node.Preds {
                select {
                case <-done[pred]:
                case <-ctx.Done():
                    return
                }
            }

            // 构建输入快照 → 执行算子 → 应用输出
            // ...
        }(i)
    }
}

这是一种隐式拓扑排序调度——不需要维护就绪队列、不需要中央调度器轮询，DAG 的拓扑约束完全通过 channel 的阻塞-唤醒语义自然实现。无依赖的算子在 DAG 构建完成的瞬间就开始并行执行，调度延迟趋近于零。

close(channel) 而非 send 是关键选择：一次 close 可以唤醒任意数量的等待者，天然支持一对多依赖。而 ctx.Done() 分支确保错误发生时所有等待中的 goroutine 快速退出。

引擎托管读写

算子并不直接读写 DataFrame。调度器在执行前后分别持锁构建输入快照和应用输出：

// 构建只读输入快照
mu.Lock()
input := dataframe.BuildInput(frame, commonInput, cop.Config.Meta.ItemInput, ...)
mu.Unlock()

// 算子执行——不持锁，完全并行
output := types.NewOperatorOutput()
execErr = cop.Instance.Execute(ctx, input, output)

// 应用输出
mu.Lock()
applyErr := dataframe.ApplyOutput(frame, output, cop.Name, cop.Config.Recall)
mu.Unlock()

锁只保护 BuildInput（从 map 中按字段名取值）和 ApplyOutput（写回 map）这两个极快的操作。算子执行本身——也就是耗时的大头——完全并行，不持锁。在 DAG 依赖已经保证顺序正确的前提下，这把 mutex 只是为了防止 Go 的 map 并发读写 panic，而非用于同步业务逻辑。

Trace 预分配与零 GC 竞争

traces = make([]types.OpTrace, n) // 按节点数预分配
// 每个 goroutine 直接写自己的槽位
traces[idx] = types.OpTrace{Name: cop.Name, Duration: duration, ...}

trace 数组按节点索引预分配，每个 goroutine 写自己的槽位，无需加锁。DAG 中止时未执行的算子不会出现在最终 trace 中——trace 的长度本身就是故障定位的线索。

Panic 恢复

每个算子的执行都包裹在 defer recover() 中。panicking 的算子被转化为 PanicError（带完整堆栈），终止当前 DAG，但进程继续服务其他请求。这是一条铁律：绝不因为单个请求的算子 bug 而 crash 进程。

Lua 嵌入：零 CGo，池化 VM

Pineapple 内置了 transform_by_lua 算子，允许用户用 Lua 脚本定义特征计算和条件逻辑，无需新增 Go 代码即可快速迭代。

底层使用 gopher-lua——一个纯 Go 实现的 Lua 5.1 VM，零 CGo 依赖，交叉编译和部署与纯 Go 项目无异。

两种执行模式

• Item 模式（function_for_item）：Go 对每个 item 调用一次 Lua 函数。common 字段作为标量全局变量设置一次，item 字段在每次迭代前更新为当前行的值。
• Common 模式（function_for_common）：Go 调用一次 Lua 函数。item 字段被转化为 Lua table（1-indexed 数组），包含所有行的值，支持跨 item 聚合。

State Pool：池化 + 全局变量清理

Lua VM 的状态管理是这个模块最精巧的部分。每个 Lua 算子实例持有一个基于 sync.Pool 的状态池：

1. 初始化：创建第一个 LState，加载并编译脚本，调用 snapshotGlobals 捕获 _G 全局变量的基线集合。
2. 借出：从 pool 获取一个已编译脚本的 LState，高并发时自动创建新实例。
3. 归还：调用 clearNonBaseline 遍历 _G，将所有非基线全局变量设为 nil，然后放回 pool。

这保证了每次借出的 LState 都是干净的——上一次执行设置的 item 字段全局变量不会泄漏到下一次执行。脚本只编译一次，sync.Pool 在高峰期自动扩张，空闲时由 GC 回收，完美适应并发度波动。

性能实测

我们设计了五个复杂度档次的 benchmark，对比 Lua 与等价 Go 原生算子的性能差距：

档次	计算逻辑	端到端 Lua/Go（1000 items）	隔离 Lua/Go（1000 items）
L1 identity	纯字段透传	1.2x	1.9x
L2 arithmetic	单次乘加	1.3x	1.7x
L3 branching	4 级条件分支	1.3x	1.8x
L4 multi-field	多字段加权求和	1.5x	2.7x
L5 iterative	5 次多项式（Horner 法）	2.1x	3.5x

端到端测试中，引擎框架（DataFrame 构建、DAG 调度）占总耗时的 50-70%，显著稀释了 Lua VM 的纯计算差距。这意味着在真实系统中，简单逻辑下 Lua 仅比 Go 慢约 1.3x——对于推荐系统场景，这个开销远小于 Redis 查询或特征服务调用的 I/O 延迟，可以忽略不计。

只有在密集数值计算（L5 级别）且 item 数量较大时，Go 原生算子才有显著优势。这给了业务团队一个清晰的决策框架：快速迭代用 Lua，热路径上的密集计算用 Go。

控制流编译：if/else 降级为数据依赖

Pineapple 的 Python DSL 支持条件分支：

flow.if_("user_age < 18") \
    .transform_dispatch(common_field="default_score", item_field="score") \
.else_() \
    .transform_by_lua(lua_script="...", function_for_item="score") \
.end_if_()

但 Go 引擎没有任何 if/else 原语。控制流在编译阶段就被完全消解了。

降级策略

编译器将每个分支转化为一个 transform_by_lua 控制算子，该算子计算一个布尔值写入编译器生成的隐藏 common 字段（如 _if_1、_else_2）：

-- if 分支的控制算子
function evaluate()
  if (user_age < 18) then return false else return true end
end

语义反转是关键：true 表示「跳过」，false 表示「执行」。分支内的业务算子通过 skip 字段引用对应的控制属性——当控制属性为 true 时，算子被跳过。

elseif/else 的控制算子会额外读取前面所有分支的控制属性，链式依赖保证了分支互斥。嵌套 if 天然支持，内层控制算子吸收外层条件。

对引擎零侵入

这个设计的优雅之处在于：控制流完全降级为数据依赖。控制属性作为普通 common 字段存在于 DataFrame 中，参与标准的 DAG 数据冒险分析，调度器通过已有的 skip 评估逻辑处理跳过。Go 引擎不需要任何分支处理代码，控制流的正确性完全由 DAG 依赖保证。

而且，控制属性是可观测的——开启 debug 模式就能看到每个分支的计算结果，排查分支逻辑错误变得非常直观。

单一事实来源：从 Go Schema 到 Python DSL

Pineapple 的代码生成流水线确保了 Go 算子注册信息是唯一的事实来源。

// 算子注册：Go 侧定义 Schema
func init() {
    pine.Register(pine.OperatorSchema{
        Name:        "transform_normalize",
        Type:        pine.OpTypeTransform,
        Description: "Normalizes a numeric field to [0, 1] range.",
        Params: map[string]pine.ParamSpec{
            "field": {Type: "string", Required: true, Description: "Target field."},
            "min":   {Type: "float64", Required: true, Description: "Range minimum."},
            "max":   {Type: "float64", Required: true, Description: "Range maximum."},
        },
    }, func() pine.Operator { return &NormalizeOp{} })
}

一次 go run ./cmd/pineapple-codegen 从注册表中读取所有 Schema，同时生成：

• Python 类型化 Helper（apple_generated/operators.py）——带完整类型注解和 IDE 补全
• Markdown 算子文档（doc/operators/）——参数表格、元数据契约、用法示例
• Python 资源类（apple_generated/resources.py）——资源声明的类型化绑定

Go Schema (注册表)
    │  codegen
    ├─────────> Python helpers (apple_generated/)
    ├─────────> Markdown docs (doc/operators/)
    │
    │  DSL 声明
    v
Python Flow → compile → JSON → Go 引擎加载

Register() 中强制校验 Description 和所有参数的 Description 非空——缺少就 panic，代码根本无法注册成功，更不可能上线。这比常见的「lint 检查文档注释」方案更为激进：编译时强制，而非运行时建议。

Python DSL 还有一个巧妙的设计：Flow.__getattr__ 拦截所有未知方法调用，将其记录为 OpCall。这意味着即使没有跑 codegen，flow.some_future_op(...) 也能工作——DSL 天然前向兼容任何未来算子。codegen 生成的类型化 Helper 是锦上添花（IDE 补全和类型检查），而非必需品。

编译期四重校验

Apple 编译器在生成 JSON 之前执行四重静态校验：

1. 字段覆盖——每个算子声明读取的字段必须有上游产出（「定义前使用」检查）
2. 死代码检测——产出字段未被下游消费的算子被标记为无效
3. 写后覆写检测——同一字段被多次写入时发出警告
4. 控制流完整性——if_ 必须有对应的 end_if_，空分支报错

这些校验全部是声明式的——只需要算子序列和 Flow 契约，不需要运行时信息。将错误拦截在开发阶段，而非部署后。

动态资源管理

算子在执行时经常需要读取定时刷新的外部数据——特征索引表、AB 实验配置、轻量模型参数。Pineapple 的 pkg/resource 包提供了一个与 Engine 生命周期完全独立的资源管理器。

rm := resource.NewManager()
rm.Register("user_features", fetchFeatureTable, 5*time.Minute)
rm.Start(ctx)  // 首次同步拉取，后续异步定时刷新
defer rm.Stop()

设计要点：

• 无锁读：atomic.Value 实现读写分离，读路径零锁竞争。刷新 goroutine 构建完整新版本后原子替换。
• 首次同步，后续异步：Start() 首次同步拉取（失败则 fail fast），后续刷新不阻塞请求。
• 刷新失败保留旧版本：不清空旧数据，保证降级可用。
• 编译期依赖校验：DSL 编译器校验所有 resource_name 引用是否有对应声明；引擎启动时 ValidateResourceDeps 交叉检查 pipeline 引用与 ResourceManager 中已注册的资源名。

算子侧只依赖一个极简的只读接口：

rp := resource.FromContext(ctx)
val, ok := rp.Get("user_features")

测试时注入 mock 同样简单：resource.NewStatic(map[string]any{"user_features": testData})。

第三方扩展：零侵入的插件化

Pineapple 的扩展机制完全基于 Go 原生的 init() + blank import 模式，没有引入任何插件框架。第三方项目在不修改 Pineapple 源码的前提下添加自定义算子和资源：

// my-project/operators/my_scorer/scorer.go
func init() {
    pine.Register(schema, func() pine.Operator { return &MyScorer{} })
}

// my-project/cmd/my-server/main.go
import (
    _ "github.com/Liam0205/pineapple/operators" // 内置算子
    _ "my-project/operators/my_scorer"            // 自定义算子
    "github.com/Liam0205/pineapple/pkg/server"
)

func main() {
    server.Run(server.Config{ConfigPath: *configPath, Addr: *addr})
}

自定义算子与内置算子享有完全相同的能力——DAG 调度、trace、hot reload、codegen 生成 Python 绑定和文档。整个扩展体验的核心是：写算子代码和一个几行的 main.go wrapper，框架帮你搞定一切。

小结

回顾 Pineapple 的设计，几条贯穿性的理念值得提炼：

编译时强制优于运行时检查。从注册时强制填写描述（缺失直接 panic），到 DSL 编译器的四重静态校验，到启动阶段的资源依赖交叉检查——每一层都在把错误拦截在尽可能早的阶段。这不是防御性编程，而是将「正确性」从程序员的记忆力转移到系统的保证上。

借鉴成熟的抽象模型。数据冒险模型来自 CPU 流水线设计，barrier 语义来自并行计算，channel-per-node 调度利用了 Go 的并发原语。Pineapple 没有发明新的依赖推导理论，而是将经过数十年验证的模型搬到了新的应用场景。这种「站在巨人肩上」的策略，让系统设计者可以直接引用学术文献来论证正确性，而不是依赖测试覆盖率和直觉。

对称性降低认知负担。算子注册和资源注册完全对称（Schema + Factory）；MetadataHolder 和 DebugHolder 完全对称；控制流降级复用了 Lua 算子和 skip 机制。学会一种模式，就能理解整个系统中类似的构造。

极简 API 背后是深思熟虑的约束。Engine 不可变，没有 Reload；算子类型决定允许的输出方法，没有例外；追加型写者和覆写型写者的区分是硬编码的，不是配置项。这些「做不到的事」正是系统可以自动做出正确决策的前提——正因为 Recall 只能 AddItem，引擎才能安全地让多个 Recall 并行执行。

Pineapple 仍在 pre-1.0 阶段积极迭代。如果你正在构建一个需要多步骤数据处理流水线的系统，不妨尝试一下这种「声明即并行」的开发方式。

项目地址：github.com/Liam0205/pineapple