企业官网建设流程全解析

• 人工智能
• 机器学习
• 后端
• 编译器

Couler 与 SQLFlow 的联合设计是 SQLFlow 演进到 Kubernetes 原生工作流模式的关键一环。本篇技术指南以 Couler and SQLFlow 设计文档 为主体，结合仓库中的 couler 代码生成包、实验性 codegen_couler 与 IR 定义 等源码实现，完整讲解 SQLFlow 如何将"多条 codegen 后端"收敛为单一 Couler 程序生成器、如何用couler.run_container描述训练步骤、如何通过sqlflow -parse与runtime.*.train管道衔接 IR 与机器学习任务，以及如何接入 Model Zoo 自定义模型镜像。读完本文，你将掌握 SQLFlow 工作流模式下从扩展 SQL 到 Argo YAML 的完整链路，以及每一条 Couler Step 背后的 IR 协议与容器约定。

设计背景：Couler 与 SQLFlow 的分工

在设计文档的构想中，两个编译器各司其职：

• Couler是一个把"用 Python 程序描述的工作流"翻译成 Argo YAML 的编译器，产出的 YAML 可以运行在 Argo——Kubernetes 原生的工作流执行引擎上；同时 Couler 也是一套框架，指导用户以 Python 函数的方式定义 step 和 workflow。
• SQLFlow是一个把 SQL 程序翻译成 Python 程序（即submitter）的编译器。迁移之前，SQLFlow 拥有多个被称为codegen的编译后端，例如codegen_xgboost.go会生成一个调用 ODPS 执行常规 SQL 查询、调用 XGBoost 完成模型训练与预测的 submitter；这些调用 ODPS 和 XGBoost 的 Python 代码存放在仓库的python/目录下（典型如 python/runtime/xgboost/train.py）。

迁移的本质，是把"SQLFlow 负责翻译 SQL → Python submitter"与"Couler 负责翻译 Python 工作流 → Argo YAML"这两条流水线首尾相接，让 SQLFlow 不再为每种执行后端各写一套 codegen，而是统一产出 Couler 程序。

迁移目标：三大组成部分

原设计文档将迁移划分为三个部分，它们共同构成"SQL → IR → Couler 程序 → Argo YAML"的新链路。

一、把 submitter 调用的 Python 源码转换为 Couler 定义

原先 submitter 中直接执行的 Python 逻辑，被逐一改写成 Couler 的步骤级 API。设计文档给出了如下映射示例：

• couler.{odps,mysql,hive}.query(sql)：在 ODPS/MySQL/Hive 上执行一条 SQL 程序/语句；
• couler.{odps,mysql,hive}.export(table, filename)：把 ODPS/MySQL/Hive 中的表导出为 RecordIO 文件；
• couler.{xgboost,tensorflow,elasticdl}.train(model_def, data)：训练 XGBoost/TensorFlow/ElasticDL 模型；
• couler.{xgboost,tensorflow,elasticdl}.predict(trained_model, data)：用训练好的 XGBoost/TensorFlow/ElasticDL 模型做预测。

也就是说，原先生成"直接调用 ODPS/XGBoost 的 Python 脚本"，迁移后生成"调用 Couler 步骤 API 的工作流描述"，数据库方言（odps/mysql/hive）与学习框架（xgboost/tensorflow/elasticdl）被抽象成 Couler 命名空间下的函数。

二、将高频复用工作流沉淀为 Couler 函数

为了不让每个 SQL 程序都重复编写步骤组合，设计文档提出把可复用的工作流封装成sqlflow.couler下的高阶函数：

• sqlflow.couler.query(db_info, sql)：内部调用couler.{odps,mysql,hive}.query(sql)；
• sqlflow.couler.{xgboost,tensorflow,elasticdl}.train(train_ir)：先调用sqlflow.couler.query取数、再调用sqlflow.couler.export导出数据，最后调用couler.{xgboost,tensorflow,elasticdl}.train(...)完成训练。

这样一条SELECT ... TO TRAIN语句在用户视角只是"一个训练函数调用"，而底层则被展开为"查询 → 导出 → 训练"多个工作流步骤。

三、用唯一 codegen_couler.go 取代多个 codegen

这是迁移的收敛点：不再维护多个 codegen 后端，而是只保留一个 codegen_couler.go。它把 SQL 程序的**中间表示（IR）**翻译成一个 Couler 程序，随后 SQLFlow 调用 Couler 编译器继续转换成 Argo YAML 并提交执行。

例如，codegen_couler.go会把一条SELECT ... TO TRAIN语句转换成对sqlflow.couler.{xgboost,tensorflow,elasticdl}.train(...)的调用。这里的 IR 衔接了 SQLFlow 与 Couler 两侧，其总体设计可参考 intermediate_representation.md，具体类型定义见 go/ir/ir.go。

Couler Step Function：一条训练语句如何变成一个工作流步骤

设计文档以一个 XGBoost 训练步骤为例，展示了 Couler 步骤函数的形态：

couler.run_container( cmd=''' echo "SELECT ... TO TRAIN xgboost.booster" | sqlflow -parse | python -m runtime.xgboost.train ''', env={"SQLFLOW_DATASOURCE": "mysql://user:pass@192.168.1.1:3306"}, # set session message as the env vars. image="sqlflow/sqlflow:step" )

这段代码中的三个要素各自对应 SQLFlow 中的一个组件：

• sqlflow -parse（SQLFlow 命令行工具）：把输入的扩展 SQL 编译成序列化到文件中的 IR；
• runtime.xgboost.train（SQLFlow submitter Python 模块）：读取 IR protobuf 文本文件并提交机器学习任务；
• sqlflow/sqlflow:step（Docker 镜像）：打包了 SQLFlowruntimePython 模块与 SQLFlow 命令行工具的运行环境。

关于镜像内容的实现，可参考 docker/step/Dockerfile：它把构建出的step可执行文件放入/usr/bin/，把解析器 jar 包放入/opt/sqlflow/parser/，并把python/runtime复制到/opt/sqlflow/python/runtime/，同时设置PYTHONPATH指向该目录，从而保证容器内既能运行sqlflow/step命令，也能import runtime.*。

命令管道解析

上述cmd是一段 shell 管道：echo输出扩展 SQL →sqlflow -parse解析出 IR →python -m runtime.xgboost.train消费 IR 并训练。设计文档强调，env中通过SQLFLOW_DATASOURCE这样的环境变量传递 Session 消息，例如mysql://user:pass@192.168.1.1:3306。

在仓库的实际实现中，这条"字符串管道"演变为更结构化的代码生成方式。查看 go/workflow/couler/template.go 中的 Couler 模板，可以看到每个 SQL 语句最终都会生成一个couler.run_container(...)调用，其命令为step -e "<SQL>"：

couler.run_container(command=["bash", "-c", step_command('''{{ $ss.OriginalSQL}}''', step_log_file)], image="{{ $ss.DockerImage}}", env=step_envs, secret=sqlflow_secret, resources=resources)

这里的step -e正是 go/cmd/step/step.go 中定义的命令行入口：step -e "select * from table1"会调用step.RunSQLProgramAndPrintResult执行 SQL 并打印结果。也就是说，实际落地时"sqlflow -parse | python -m runtime.xgboost.train"的管道被统一成了"step -e <original_sql>"，由step进程在容器内完成解析、IR 生成与运行时调用——这与设计文档描述的职责分工一致，只是把解析与执行整合进了step命令。

sqlflow -parse：SQLFlow 命令行工具与 IR 序列化

设计文档把sqlflow -parse定义为：接受一条扩展 SQL，输出 SQLFlow 的 IR，格式为protobuf 文本（protobuf text format）。IR 是 SQLFlow 与各类运行时之间的契约，设计文档给出的 protobuf 定义如下：

message FieldDesc { required string name = 1; required FieldType dtype = 2; optional string delimiter = 3; repeated int shap = 4; required bool is_sparse = 5; required map<string, string> vocabulary = 6; required int32 max_id = 7; } message NumericColumn { FieldDesc field_meta = 1; } message BucketColumn { ... } message FeatureColumn { oneof feature_column { NumericColumn nc = 1; BucketColumn bc = 2; CrossColumn cc = 3; ... } } message Session { string token = 1; string db_conn_str = 2; bool exit_on_submit = 3; string user_id = 4; // for loading CSV to hive string hive_location = 5; string hdfs_namenode_addr = 6; string hdfs_user = 7; string hdfs_pass = 8; } message IR { required string datasource = 1; required string select = 2; optional string validation_select = 3; required string estimator = 4; optional map<string, string> attributes = 5; optional map<string, FeatureColumn> features = 6; optional FeatureColumn label = 7; optional Session session = 8; }

各字段含义可以这样理解：

• FieldDesc描述一个字段的元信息：字段名name、数据类型dtype、稀疏特征的分隔符delimiter、张量形状shap、是否稀疏is_sparse、词表vocabulary与最大 IDmax_id；
• NumericColumn/BucketColumn/CrossColumn是特征列（FeatureColumn）的具体形态，通过oneof互斥；
• Session携带用户会话信息：token、数据库连接串、用户 ID，以及向 Hive 加载 CSV 所需的hive_location、hdfs_namenode_addr、hdfs_user、hdfs_pass；
• 顶层IR把数据源、训练 SELECT、验证 SELECT、estimator、WITH 属性、特征列与标签、Session 打包成一条完整语句的可执行描述。

与当前仓库 IR 实现的对应

设计文档注释指出"更多 IR 定义细节见ir.go"。在仓库中，go/ir/ir.go 用 Go 结构体实现了同一套模型：

• TrainStmt对应 IR 中训练语句的核心信息，其Select、ValidationSelect、Estimator、Attributes、Features、Label、Into、PreTrainedModel等字段与 protobuf 定义一一对应；ModelImage字段则对应TO TRAIN a_data_scientist/regressors:v0.2/MyDNNRegressor中的 Docker 镜像名；
• SQLFlowStmt接口由TrainStmt、PredictStmt、ExplainStmt、EvaluateStmt、NormalStmt、ShowTrainStmt、OptimizeStmt、RunStmt共同实现，IsExtended()区分普通 SQL 与扩展 SQL；
• 会话侧的Session消息定义在 go/proto/sqlflow.proto，除设计文档中的字段外还扩展了submitter、service_account、wf_namespace等字段，用于工作流模式下的提交者选择与 RBAC。

SQLFlow Submitter Python Module：IR 驱动的训练入口

设计文档明确了 submitter 模块的调用约定：runtime.{tensorflow,xgboost,elasticdl}.train接受一个 protobuf 文本格式的 SQLFlow IR，然后提交 TensorFlow、XGBoost 或 ElasticDL 训练任务，命令形如：

cat ir.proto_text | python -m runtime.xgboost.train

这个约定在仓库中有完整实现。python/runtime/xgboost/train.py 中的train函数接收datasource、select、model_params、train_params、feature_metas、feature_column_names、label_meta、validation_select等参数：先通过xgb_dataset从数据源构造 XGBoost 的DMatrix数据迭代器，再调用xgb.train迭代训练，最后在rank == 0的节点上收集元数据（collect_metadata）、保存模型与model_meta.json；针对 PAI 分布式场景还提供dist_train入口，配合PaiXGBoostTracker/PaiXGBoostWorker完成多机 rabit 初始化与参数服务器协调。

在实验性的 step 代码生成路径中，上述参数不再是"从 protobuf 文本反序列化"，而是由 Go 端直接填充进 Python 函数模板：参见 codegen_step_train.go 中GenerateTrain生成的step_entry_N()函数，它把Select、ValidationSelect、ModelParamsJSON、TrainParamsJSON、FeatureColumnCode、LabelColumnCode、Save（INTO 表名）、Load（预训练模型）等直接拼装进 Python 代码，再经 Couler 模板以couler.run_script提交执行。普通 SQL 语句则生成调用runtime.dbapi.connect的步骤，见 codegen_normal_stmt.go。

Couler Step Function 与 Model Zoo：自定义模型镜像的接入

Model Zoo 中的自定义模型会被打包成 Docker 镜像，用户在 SQL 中通过镜像名引用模型，例如：

SELECT ... TO TRAIN regressors:v0.2/MyDNNRegressor ...

此时 Couler 步骤函数变为：

couler.sqlflow.run(''' echo "SELECT ... TO TRAIN regressors:v0.2/MyDNNRegressor ..." | sqlflow -parse | python -m runtime.xgboost.train ''', env={"SQLFLOW_DATASOURCE": "mysql://user:pass@192.168.1.1:3306"}, # set session message as the env vars. image="regressors:v0.2")

与默认步骤的关键差异在于image 字段：训练该自定义模型时，工作流步骤不再使用sqlflow/sqlflow:step，而是直接使用模型仓库镜像regressors:v0.2。仓库实现中，codegen.go 的GenFiller对*ir.TrainStmt做了专门处理：当ModelImage非空时，将该语句步骤的DockerImage覆盖为i.ModelImage；预测、评估、解释步骤则通过findModelGenerationTrainStmt向前回溯同程序内产出该模型的TrainStmt，取其镜像，否则从 Model Zoo 或数据库中的模型元数据解析镜像名（见 codegen_step.go）。

自定义镜像约束与本地调试

设计文档特别强调两点约束与便利：

1. 自定义模型 Docker 镜像必须基于sqlflow/sqlflow:step构建，从而继承 SQLFlow 命令行工具与runtimePython 模块；
2. 自定义模型容器也可以在宿主机上直接启动，便于脱离 Kubernetes 调试 SQLFlow 流程：

> docker run --rm -it -v$PWD:/models regressors:v0.2/MyDNNRegressor bash > sqlflow -parse < a.sql > ir.json > python -m runtime.tensorflow.train < ir.json

即：进入容器后，先把 SQL 解析为 IR 文件（sqlflow -parse < a.sql > ir.json），再把 IR 交给runtime.tensorflow.train消费。这条调试路径与设计文档给出的生产管道保持同一套 IR 协议，只是拆成手动步骤执行。Model Zoo 的整体设计可参考 model_zoo.md。

仓库中的工程落点：couler 代码生成包

设计文档描述的"唯一 codegen_couler.go"在仓库中落在两个层次：go/codegen/experimental/与go/workflow/couler/。

GenFiller 与 Couler 模板

go/workflow/couler/codegen.go 是 Couler 程序生成的核心：

• GenFiller(programIR, session, useCoulerSubmitter)遍历 IR 语句列表，按语句类型生成sqlStatement：
  • 普通 SQL 与PredictStmt/ExplainStmt/EvaluateStmt使用默认镜像sqlflow/sqlflow（可用SQLFLOW_WORKFLOW_STEP_IMAGE覆盖）；
  • TrainStmt优先使用ModelImage；
  • 当SQLFLOW_submitter为katib时，TrainStmt会转成 Katib 超参搜索语句（见 katib.go）；
  • RunStmt（SELECT TO RUN）使用语句中指定的ImageName；
  • 无法识别的 IR 类型直接报错返回。
• escapeStepSQL对 SQL 文本中的\、"、反引号与$做转义，防止注入到 shell 命令中破坏步骤脚本；
• GenCode将Filler渲染进 template.go 中的 Couler 模板，产出import couler.argo as couler开头的 Python 工作流程序。

环境变量、Secret、资源与 TTL

模板渲染时还会注入一组工作流级配置，它们全部来自 sqlflowserver 进程的环境变量：

• 步骤环境变量：GetStepEnvs会把 sqlflowserver 上所有SQLFLOW_前缀（排除SQLFLOW_WORKFLOW_前缀）的环境变量连同 Session 中的SQLFLOW_USER_TOKEN、SQLFLOW_DATASOURCE、SQLFLOW_USER_ID、SQLFLOW_HIVE_LOCATION、SQLFLOW_HDFS_NAMENODE_ADDR、SQLFLOW_HADOOP_USER、SQLFLOW_HADOOP_PASS等一起传给每个步骤容器；
• Secret：SQLFLOW_WORKFLOW_SECRET以 JSON 形式（{name: {key: value, ...}}）声明工作流 Secret，模板中以couler.create_secret(..., dry_run=True)引用——设计上要求该 Secret 在启动 sqlflowserver 之前已创建，dry_run仅用于生成 YAML 时引用元数据；
• 资源配额：SQLFLOW_WORKFLOW_RESOURCES必须是合法 JSON（如{"memory": "32Mi", "cpu": "100m"}），由VerifyResources校验后作为resources传入couler.run_container；
• TTL 清理：workflowTTL默认24 * 3600秒，可通过SQLFLOW_WORKFLOW_TTL调整，最终通过couler.config_workflow(time_to_clean=workflow_ttl)生效；
• 日志收集：SQLFLOW_WORKFLOW_STEP_LOG_FILE指定容器内步骤日志文件路径，模板会拼接mkdir -p、step -e ... 2>&1 | tee <log>、sleep退出等待等命令序列，并依赖/opt/sqlflow/init_step_container.sh做容器初始化。

从 Couler 程序到 Argo YAML

GenYAML把生成的 Couler 程序写入临时文件后执行python <file>，利用 Couler 编译器将其翻译为 Argo YAML（实现见 codegen.go 的GenYAML）。若启用useCoulerSubmitter（即使用 Couler Submitter 模式），则要求配置SQLFLOW_COULER_SERVER_ADDR与SQLFLOW_COULER_CLUSTER，模板末尾会引入CoulerSubmitter并把工作流提交到指定集群。

在 codegen_couler.go 的GenerateCodeCouler中，整条链路被封装为三步：parseToIR(sqlProgram, session)生成各语句 IR →GenerateStepCodeAndImage为每条语句生成步骤 Python 代码与镜像 →CodeGenCouler汇总渲染 Couler 程序。

测试与验证

仓库为 Couler 代码生成提供了直接可验证的测试证据，见 go/workflow/couler/codegen_test.go：

• TestCoulerCodegen用MockSQLProgramIR（一条SELECT * FROM iris.train limit 10;普通语句加一条训练语句）验证：生成的 Couler 程序包含原始 SQL 与step_envs["SQLFLOW_OSS_AK"]等步骤环境变量、couler.create_secret(..., dry_run=True)、resources=json.loads(...)；再经GenYAML得到的 Argo YAML 中能匹配出step -e "SELECT * FROM iris.train limit 10;"；设置SQLFLOW_WORKFLOW_STEP_LOG_FILE后，YAML 中会出现mkdir -p与tee日志管道；
• TestCoulerCodegenSpecialChars验证包含反引号、$、双引号、反斜杠的 SQL 经转义后仍能正确生成 YAML；
• TestStringInStringSQL验证训练 SQL 内部嵌套字符串（如validation.select="..."）的多层转义；
• TestKatibCodegen验证SQLFLOW_submitter=katib时，TO TRAIN xgboost.gbtree WITH objective=..., max_depth=[2, 10]这类语句能走 Katib 超参搜索路径。

这些测试从侧面印证了设计文档的迁移结论：无论底层是 ODPS、MySQL、Hive 还是 PAI，SQLFlow 面向工作流的产出物已经统一为 Couler 程序 + Argo YAML，SQL 程序与机器学习执行之间只隔着一层可序列化的 IR。

• 人工智能
• 机器学习
• 后端
• 编译器