SQLFlow 基于 Couler 的工作流代码生成设计：从多 Codegen Submitter 到统一的 Argo 工作流-北京尧图网络

人工智能机器学习后端编译器【免费下载链接】sqlflowBrings SQL and AI together.项目地址https://gitcode.com/gh_mirrors/sq/sqlflow点击查看免费下载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:pass192.168.1.1:3306}, # set session message as the env vars. imagesqlflow/sqlflow:step )这段代码中的三个要素各自对应 SQLFlow 中的一个组件sqlflow -parseSQLFlow 命令行工具把输入的扩展 SQL 编译成序列化到文件中的 IRruntime.xgboost.trainSQLFlow submitter Python 模块读取 IR protobuf 文本文件并提交机器学习任务sqlflow/sqlflow:stepDocker 镜像打包了 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:pass192.168.1.1:3306。在仓库的实际实现中这条字符串管道演变为更结构化的代码生成方式。查看 go/workflow/couler/template.go 中的 Couler 模板可以看到每个 SQL 语句最终都会生成一个couler.run_container(...)调用其命令为step -e SQLcouler.run_container(command[bash, -c, step_command({{ $ss.OriginalSQL}}, step_log_file)], image{{ $ss.DockerImage}}, envstep_envs, secretsqlflow_secret, resourcesresources)这里的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 -parseSQLFlow 命令行工具与 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 mapstring, 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 mapstring, string attributes 5; optional mapstring, FeatureColumn features 6; optional FeatureColumn label 7; optional Session session 8; }各字段含义可以这样理解FieldDesc描述一个字段的元信息字段名name、数据类型dtype、稀疏特征的分隔符delimiter、张量形状shap、是否稀疏is_sparse、词表vocabulary与最大 IDmax_idNumericColumn/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 ModuleIR 驱动的训练入口设计文档明确了 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、SaveINTO 表名、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:pass192.168.1.1:3306}, # set session message as the env vars. imageregressors: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。自定义镜像约束与本地调试设计文档特别强调两点约束与便利自定义模型 Docker 镜像必须基于sqlflow/sqlflow:step构建从而继承 SQLFlow 命令行工具与runtimePython 模块自定义模型容器也可以在宿主机上直接启动便于脱离 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.goRunStmtSELECT 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等一起传给每个步骤容器SecretSQLFLOW_WORKFLOW_SECRET以 JSON 形式{name: {key: value, ...}}声明工作流 Secret模板中以couler.create_secret(..., dry_runTrue)引用——设计上要求该 Secret 在启动 sqlflowserver 之前已创建dry_run仅用于生成 YAML 时引用元数据资源配额SQLFLOW_WORKFLOW_RESOURCES必须是合法 JSON如{memory: 32Mi, cpu: 100m}由VerifyResources校验后作为resources传入couler.run_containerTTL 清理workflowTTL默认24 * 3600秒可通过SQLFLOW_WORKFLOW_TTL调整最终通过couler.config_workflow(time_to_cleanworkflow_ttl)生效日志收集SQLFLOW_WORKFLOW_STEP_LOG_FILE指定容器内步骤日志文件路径模板会拼接mkdir -p、step -e ... 21 | tee log、sleep退出等待等命令序列并依赖/opt/sqlflow/init_step_container.sh做容器初始化。从 Couler 程序到 Argo YAMLGenYAML把生成的 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.goTestCoulerCodegen用MockSQLProgramIR一条SELECT * FROM iris.train limit 10;普通语句加一条训练语句验证生成的 Couler 程序包含原始 SQL 与step_envs[SQLFLOW_OSS_AK]等步骤环境变量、couler.create_secret(..., dry_runTrue)、resourcesjson.loads(...)再经GenYAML得到的 Argo YAML 中能匹配出step -e SELECT * FROM iris.train limit 10;设置SQLFLOW_WORKFLOW_STEP_LOG_FILE后YAML 中会出现mkdir -p与tee日志管道TestCoulerCodegenSpecialChars验证包含反引号、$、双引号、反斜杠的 SQL 经转义后仍能正确生成 YAMLTestStringInStringSQL验证训练 SQL 内部嵌套字符串如validation.select...的多层转义TestKatibCodegen验证SQLFLOW_submitterkatib时TO TRAIN xgboost.gbtree WITH objective..., max_depth[2, 10]这类语句能走 Katib 超参搜索路径。这些测试从侧面印证了设计文档的迁移结论无论底层是 ODPS、MySQL、Hive 还是 PAISQLFlow 面向工作流的产出物已经统一为 Couler 程序 Argo YAMLSQL 程序与机器学习执行之间只隔着一层可序列化的 IR。赞分享人工智能机器学习后端编译器【免费下载链接】sqlflowBrings SQL and AI together.项目地址https://gitcode.com/gh_mirrors/sq/sqlflow点击查看免费下载相关推荐SQLFlow 工作流设计把 SQL 程序编译为 Argo 工作流SQLFlow 工作流设计把 SQL 程序编译为 Argo 工作流本文围绕 SQLFlow 的工作流Workflow设计文档展开为什么 SQLFlow人工智能机器学习后端编译器SQLFlow workflow 包设计以统一接口对接 Argo/Tekton 工作流引擎的 Codegen 与 Submit/Fetch 机制SQLFlow workflow 包设计以统一接口对接 Argo/Tekton 工作流引擎的 Codegen 与 Submit/Fetch 机制本文基于 S人工智能机器学习后端编译器VS Code Copilot Chat 的 Claude Code 集成架构、消息流与扩展点全解析VS Code Copilot Chat 的 Claude Code 集成架构、消息流与扩展点全解析本指南以开源仓库 vscode copilot chat人工智能机器学习后端编译器上一篇fabio gRPC 透明代理GRPC Proxy配置与源码级原理详解下一篇WaylandCraft如何扫描你的桌面应用XDG .desktop条目解析与应用启动器实现详解创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

SQLFlow 基于 Couler 的工作流代码生成设计：从多 Codegen Submitter 到统一的 Argo 工作流

相关推荐

React异步状态管理与电商分类导航实战

中药材原料选购全攻略：道地性、性价比与供应商选择

游戏脚本开发的法律风险与合规实践指南

更多资讯

数据库课程设计实战：从建库到事务闭环的完整指南

ai-agent-book：从入门到 Agent 工程实践（一）

【PMSM滤波代码2】PMSM未知负载转矩观测EKF。MATLAB例程（负载增广EKF与实际仿真结果）。订阅专栏后可查看完整代码

MySQL 8.4课程设计实战：从教科书模型到生产级数据库

解Bug之路-隔离故障点后业务不恢复?-热点Key处理与亚稳态故障

PostGIS 3.5.0 + PostgreSQL 17 Windows二进制捆绑包详解

今日更新

声子晶体仿真：三角晶格与六角晶格的能带计算实战

度假村安保定岗定编：岗位梳理、编制测算与排班设计

COMSOL等离子体模块实战：氩气双层DBD仿真建模全解析

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？