Apache Texera 1.2.0-incubating 算子工作流进阶实战详解

目标：在已编译启动的 Texera 1.2.0-incubating 服务上，实测算子元数据接口的全部 163 个算子，讲清 operatorType 注册与反序列化机制、算子分组全景、工作流 JSON 装配结构与 Python UDF 生命周期，并给出可直接复用的元数据查询方法。已在 Ubuntu 24.04（x86_64）+ OpenJDK 17.0.20.1 + sbt 1.12.9 + PostgreSQL 16 环境实测通过。

一、环境信息

• 服务版本：Texera v1.2.0-incubating（commit c3aa6e4ba36bdd0cdc9f76bd7896c024dda392b3）
• 运行形态：sbt "WorkflowExecutionService/stage" 产物 texera-web-application，参数 -J-Xmx512m，监听 8080（API）/8081（管理）
• 数据库：PostgreSQL 16.15（texera_db 已初始化 28 张表）
• 元数据接口：GET http://localhost:8080/api/resources/operator-metadata
• 算子源码位置：common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/
• 类型注册源码：同目录 LogicalOp.scala

二、算子元数据全景（163 个，实测）

服务启动后，算子元数据由后端扫描全部算子描述符类动态生成。实测返回的 JSON 顶层结构是 {"operators":[...]}，每个元素恰好 4 个字段：

sample keys of one op: ['operatorType', 'jsonSchema', 'additionalMetadata', 'operatorVersion']

• operatorType：算子类型唯一名，即工作流 JSON 中引用的标识；
• jsonSchema：该算子参数表的 JSON Schema（含 properties、required、options、definitions）；
• additionalMetadata：名称、描述、分组、端口等展示元数据；
• operatorVersion：版本号，实测多数为 N/A。

算子总数与首个算子实测：

curl -s http://localhost:8080/api/resources/operator-metadata \
  | python3 -c "import sys,json; d=json.load(sys.stdin); print(len(d['operators']), d['operators'][0]['operatorType'])"

163 TwitterFullArchiveSearch

按 additionalMetadata.operatorGroupName 对 163 个算子做全量统计（实测）：

Sklearn 28 / Sklearn Training 26 / Basic 16 / Scientific 14 / Data Input 8
Statistical 8 / Data Cleaning 5 / Financial 5 / Utilities 5 / Python 5
External API 4 / Media 4 / Hugging Face 4 / Advanced Sklearn 4 / Set 4
Search 4 / Sort 3 / Database Connector 3 / Join 3 / Control Block 2
Advanced 2 / R 2 / Java 1 / Aggregate 1 / Machine Learning General 1 / Visualization 1

机器学习相关分组（Sklearn 系 + Hugging Face + Machine Learning General）合计 63 个，占了近四成，是 Texera 算子库的重心之一。

💡 技术拓展：把「算子即服务」做成元数据驱动是 Texera 与手工编码数据管道的关键差异——前端算子面板、参数表单、校验规则全部由这一份元数据生成，新增一个算子只需要提供描述符类与注解，不触碰前端代码。本篇后续章节的分组面板、参数表单均以该接口为唯一事实来源。

三、operatorType 注册与反序列化机制

3.1 LogicalOp 的 JsonTypeInfo 注解

工作流 JSON 反序列化的入口是 common/workflow-operator/.../operator/LogicalOp.scala，实测第 159 行起的注解：

@JsonTypeInfo(
  use = JsonTypeInfo.Id.NAME,
  include = JsonTypeInfo.As.PROPERTY,
  property = "operatorType"
)
@JsonSubTypes(
  Array(
    new Type(value = classOf[IfOpDesc], name = "If"),
    new Type(value = classOf[SankeyDiagramOpDesc], name = "SankeyDiagram"),
    new Type(value = classOf[IcicleChartOpDesc], name = "IcicleChart"),
    new Type(value = classOf[FileListerSourceOpDesc], name = "FileLister"),
    new Type(value = classOf[CSVScanSourceOpDesc], name = "CSVFileScan"),
    new Type(value = classOf[JSONLScanSourceOpDesc], name = "JSONLFileScan"),
    ...

property = "operatorType" 说明：提交到后端的工作流 JSON 中，每个算子对象的 operatorType 字符串就是在这里登记的名字，Jackson 据此实例化对应的 OpDesc 子类。实测该文件内活跃的 new Type(value = ...) 注册共 131 条（另有 1 条被注释禁用的 ParallelCSVFileScan，源码注释说明是为避免用户混淆，实验场景可手工恢复）。

3.2 元数据接口的 163 从哪里来

131 与 163 的差异是正常的：@JsonSubTypes 面向「反序列化」，而元数据接口的 163 个算子来自 OperatorMetadataGenerator 对全部描述符类的扫描（含未列入 @JsonSubTypes 的历史与实验性算子）。因此判断一个 operatorType 能否被引擎接受，应以元数据接口的实际返回为准——本篇所有 operatorType 均已逐一在该接口验证存在。

3.3 描述符类的包分布

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/ 下的包结构与源文件数（实测 find 统计，.scala 与 .java 合计）：

visualization: 69    sklearn: 56    source: 52    machineLearning: 12
udf: 9               huggingFace: 4

其余子包包括 aggregate、cartesianProduct、dictionary、difference、distinct、filter、flatmap、hashJoin、ifStatement、intersect、intervalJoin、keywordSearch、limit、map、projection、randomksampling、regex、reservoirsampling、sink、sleep、sort、sortPartitions、split、substringSearch、symmetricDifference、typecasting、union 等，与第四章的分组清单一一对应。

四、算子族分组实战清单

以下 30 个算子的三项元数据（友好名 / 分组 / 输入输出端口数）全部取自元数据接口实测回显，可直接作为搭建工作流的选型表。

4.1 数据读取（Data Input / Database Connector / External API）

Data Input 分组共 8 个算子，CSV/JSONL 文件扫描与内联文本输入覆盖了绝大多数本地数据接入场景；关系库接入走 MySQL/PostgreSQL Source；网页抓取走 URL Fetcher。

4.2 行变换与集合运算（Search / Data Cleaning / Set / Join / Sort / Aggregate）

端口数是接线时的硬约束：HashJoin、IntervalJoin、Difference、Intersect 均为双输入，Union 单输入多上游汇聚；聚合、排序、去重都属于阻塞型算子，需上游全部数据到齐后才输出。

4.3 可视化（Basic / Media）

可视化算子通常放在工作流末端，结果经 Iceberg 结果表回传前端渲染。

4.4 自定义计算（Python / R / Java）

Python 分组共 5 个算子；R 分组 2 个、Java 分组 1 个。单输入场景用 PythonUDFV2，需要合并两路输入（如模型端口与数据端口）时用 DualInputPortsPythonUDFV2，生命周期规范见第六章。

五、工作流 JSON 装配结构

工作流在数据库与前端的持久化形态是一份 JSON。用仓库自带示例 bin/single-node/examples/workflows/[Example] Data Exploration on Movies Dataset.json 实测解析：

top keys: ['operators', 'operatorPositions', 'links', 'commentBoxes', 'settings']

5.1 算子对象字段

operator keys: ['operatorID', 'operatorType', 'operatorVersion', 'operatorProperties',
                'inputPorts', 'outputPorts', 'showAdvanced', 'isDisabled',
                'customDisplayName', 'dynamicInputPorts', 'dynamicOutputPorts', 'viewResult']

实测样例：

{"operatorID": "CSVFileScan-operator-ddb8baf2-03e1-4b7a-9b88-719db6a21199",
 "operatorType": "CSVFileScan",
 "customDisplayName": "Read Movie Dataset"}

字段装配要点：

• operatorID：全局唯一，惯例格式 <operatorType>-operator-<uuid>；
• operatorType：必须与元数据接口中的类型名逐字符一致（第三节的注册表）；
• operatorProperties：键值对，键名与取值约束来自该算子的 jsonSchema.properties，必填项来自 jsonSchema.required；
• inputPorts/outputPorts：显式声明端口实例；dynamicInputPorts/dynamicOutputPorts 为 true 的算子允许用户增删端口；
• viewResult：为 true 的算子结果会进入前端结果面板。

5.2 参数 Schema 与 operatorProperties 的对应

以 KeywordSearch 为例，实测其 JSON Schema：

KeywordSearch properties: ['dummyPropertyList', 'attribute', 'keyword']
required: ['attribute', 'keyword']

即工作流 JSON 中该算子的 operatorProperties 至少要提供 attribute 与 keyword 两个键，前端表单也会按同一 Schema 渲染出这两个必填输入框。任何新增算子只要描述符类注解写清 Schema，参数校验就自动生效。

5.3 连线对象字段

link keys: ['linkID', 'source', 'target']
link sample: {"linkID": "3fba9036-949b-41e6-b1ac-809074e95b3d",
              "source": {"operatorID": "Aggregate-operator-41c3fbf6-...", "portID": "output-0"},
              "target": {"operatorID": "PieChart-operator-...", "portID": "input-0"}}

连线以 source.portID → target.portID 表达数据流向，portID 命名惯例是 output-N/input-N。operatorPositions 与 commentBoxes 仅描述画布布局，settings 保存工作流级设置，三者均不参与执行编译。

六、Python UDF 生命周期与编写规范

官方教程 docs/tutorials/guide-to-use-python-udf.md 给出了 UDF 算子的完整生命周期 API，原文要点实测摘录如下：

1. `open() -> None` Open a context of the operator. ... It will be invoked once per operator.
2. `process(data, port: int) -> Iterator[Optional[data]]` Process an input data from the given port, ... invoked once for every unit of data.
3. `on_finish(port: int) -> Iterator[Optional[data]]` Callback when one input port is exhausted, ... invoked once per port.
4. `close() -> None` Close the context of the operator. ... invoked once per operator.

对应到代码骨架：

class MyOp(TexeraOperator):
    def open(self):
        # 每算子一次：加载模型、文件或 API 客户端
        ...

    def process(self, tuple_, port: int):
        # 每条数据一次：yield 0 到多条输出
        yield tuple_

    def on_finish(self, port: int):
        # 每输入端口耗尽一次：适合收尾聚合输出
        ...

    def close(self):
        # 每算子一次：释放资源
        ...

按处理粒度，教程还提供三个批量变体（实测摘录自同文档）：

def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]
def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]
def process_batch(self, batch: Batch, port: int) -> Iterator[Optional[BatchLike]]

选择建议：逐行变换用 process_tuple；整表算法（如一次读全量做归一化）用 process_table；模型推理类批量场景用 process_batch。双输入 UDF（DualInputPortsPythonUDFV2）中，port 参数区分来源端口，典型用法是 0 号端口传模型、1 号端口传数据（教程示例原注释 if port == 0: # models port）。

七、注意事项与常见问题

7.1 operatorType 大小写敏感且必须取自元数据接口

工作流 JSON 的 operatorType 与 @JsonSubTypes 登记名严格一致（如 CSVFileScan 而非 CsvFileScan）。不可凭记忆拼写，应从 GET /api/resources/operator-metadata 返回值复制。

7.2 部分算子在注册表中处于禁用态

源码中 ParallelCSVScanSourceOpDesc（ParallelCSVFileScan）被注释禁用，注释说明是为避免用户混淆，需要时可在 LogicalOp.scala 恢复并重新编译。同理，元数据接口可见但长期未列入 @JsonSubTypes 的算子，用于工作流反序列化前需先确认类型注册覆盖。

7.3 operatorVersion 多为 N/A

实测多数算子的 operatorVersion 为 N/A：该值来自 git 仓库定位（启动日志中 jgit 无法定位仓库时会退化），不影响功能，但意味着不能依赖它做算子级版本比对。

7.4 元数据接口需要服务就绪后才可调用

后端在首次响应元数据请求时才完成全部 163 个描述符的实例化与 Schema 生成（启动日志可见 OperatorMetadataGenerator 相关堆栈先于 Jetty 绑定完成）。若服务刚启动立即请求偶发失败，等待 Started @xxxxms 日志出现后再试即可。

八、参考链接

• Apache Texera 官网：https://texera.apache.org/
• 源码仓库：https://github.com/apache/texera（tag v1.2.0-incubating）
• 算子类型注册源码：仓库内 common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala
• Python UDF 官方教程：仓库内 docs/tutorials/guide-to-use-python-udf.md
• 示例工作流：仓库内 bin/single-node/examples/workflows/
• 算子元数据接口：GET /api/resources/operator-metadata（本篇全部统计来源）
• Apache 软件基金会孵化器声明：https://incubator.apache.org/