本系列基于 SQLMesh 官方文档https://sqlmesh.readthedocs.io/en/stable/concepts/models/python_models/整理共 3 篇面向初学者。本篇是完结篇覆盖工程化进阶能力并给出全系列的避坑总表。第一篇基础语法与核心概念第二篇取数与依赖管理、四种引擎的 DataFrame 实战1. 前后置语句pre/post-statements前置/后置语句让你在模型运行前后执行 SQL。典型用途修改会话设置、创建索引。⚠️ 并发提醒不要写会与其他并发模型冲突的语句例如创建物理表并发执行时行为不可预测。1.1 在装饰器里声明pre_statements/post_statements接收一个列表元素可以是 SQL 字符串、SQLGlot 表达式或宏调用model(db.test_model,kindfull,columns{id:int,name:text,},pre_statements[SET GLOBAL parameter value;,exp.Cache(thisexp.table_(x),expressionexp.select(1)),],post_statements[CREATE_INDEX(this_model, id)],)defexecute(context,start,end,execution_time,**kwargs)-pd.DataFrame:returnpd.DataFrame([{id:1,name:name}])其中CREATE_INDEX是自定义宏在项目的macros目录里这样定义——仅在creating建表阶段执行macro()defcreate_index(evaluator:MacroEvaluator,model_name:str,column:str,):ifevaluator.runtime_stagecreating:returnfCREATE INDEX idx ON{model_name}({column});returnNone项目级默认值也可以在配置的model_defaults里为整个项目定义 pre/post 语句所有模型自动继承并与模型级语句合并默认语句先执行。1.2 在函数体内声明规则很简单前置语句写在return/yield之前任意位置即可后置语句必须把return改成yield然后写在yield之后因为后置语句要在函数产出数据之后才执行。defexecute(context:ExecutionContext,start:datetime,end:datetime,execution_time:datetime,**kwargs:t.Any,)-pd.DataFrame:# pre-statementcontext.engine_adapter.execute(SET GLOBAL parameter value;)# post-statement 要求用 yield 而不是 returnyieldpd.DataFrame([{id:1,name:name}])# post-statementcontext.engine_adapter.execute(CREATE INDEX idx ON example.pre_post_statements (id);)2. on_virtual_update虚拟层更新后执行on_virtual_update在 Virtual Update 完成后执行 SQL典型用途是给虚拟层的视图授权model(db.test_model,kindfull,columns{id:int,name:text,},on_virtual_update[GRANT SELECT ON VIEW this_model TO ROLE dev_role],)defexecute(context,start,end,execution_time,**kwargs)-pd.DataFrame:returnpd.DataFrame([{id:1,name:name}])注意这些语句的表名解析发生在虚拟层。在名为dev的环境中跑 plan 时db.test_model和this_model都会解析成db__dev.test_model而不是物理表名。同样支持在model_defaults中配置项目级默认语句。3. 蓝图Blueprinting一个模板批量生成多个模型当多个模型逻辑相同、只是参数不同比如每个客户一张表不必复制粘贴 N 份代码——用blueprints属性传一个键值字典列表一个文件就能打印出多个模型。规则模型名必须用蓝图里的变量做参数化语法是{变量名}。importtypingastfromdatetimeimportdatetimeimportpandasaspdfromsqlmeshimportExecutionContext,modelmodel({customer}.some_table,# 用蓝图变量 customer 参数化模型名kindFULL,blueprints[{customer:customer1,field_a:x,field_b:y},{customer:customer2,field_a:z,field_b:w},],columns{field_a:text,field_b:text,customer:text,},)defentrypoint(context:ExecutionContext,start:datetime,end:datetime,execution_time:datetime,**kwargs:t.Any,)-pd.DataFrame:returnpd.DataFrame({field_a:[context.blueprint_var(field_a)],field_b:[context.blueprint_var(field_b)],customer:[context.blueprint_var(customer)],})上面的定义会产生两个独立模型customer1.some_table和customer2.some_table各自使用对应的参数映射变量通过context.blueprint_var读取。3.1 动态生成蓝图列表蓝图映射也可以由宏动态构造适合从 CSV 等外部数据源读取清单model({customer}.some_table,blueprintsgen_blueprints(),...)fromsqlmeshimportmacromacro()defgen_blueprints(evaluator):return(((customer : customer1, field_a : x, field_b : y), (customer : customer2, field_a : z, field_b : w)))还可以配合EACH宏和全局列表变量valuesmodel({customer}.some_table,blueprintsEACH(values, x - (customer : schema_x)),...)4. 模型属性里使用宏变量小心 cron 的坑Python 模型的属性支持宏变量但当宏变量出现在字符串内部时要特别小心。典型场景是把调度时间做成参数化 cron# 正确写法整个表达式用引号包住并加 前缀model(my_model,cron*/{mins} * * * *,# 注意 ... 语法...)# 配合蓝图变量同样适用model({customer}.scheduled_model,cron0 {hour} * * *,blueprints[{customer:customer_1,hour:2},# 凌晨 2 点跑{customer:customer_2,hour:8},# 早上 8 点跑],...)为什么要这么麻烦因为 cron 表达式本身常用表示别名daily、hourly会和 SQLMesh 的宏语法冲突...的写法能确保正确解析。5. 全系列避坑清单Best Practices#规则原因1columns声明必须与实际返回的 DataFrame 完全一致SQLMesh 先建表再跑代码schema 不符会引发意外行为2保持模型幂等同一区间重跑结果必须一致否则回刷数据时会产生脏数据3永远不要return空 DataFrame可能为空时改用条件yieldyield from ()4用 Spark/Snowflake/BigQuery 时优先返回对应原生 DataFramecontext.spark/snowpark/bigframe让计算分布式执行避免本地内存瓶颈5读上游模型必先resolve_table直接硬编码表名会在 dev/prod 环境切换时拿错数据6depends_on显式声明会覆盖函数体内的动态引用避免依赖图与预期不符7pre/post 语句避免创建物理表多模型并发执行时会产生冲突8后置语句必须配合yield不能return后置语句要在函数产出数据之后执行9变量走函数参数时必须带默认值且不藏在kwargs里变量缺失时保证模型可加载10输出太大就用生成器分批yield降低单批内存占用11含宏变量的 cron 字符串用...包裹避免符号解析冲突12Python 模型不能用VIEW/SEED/MANAGED/EMBEDDEDkind需要这些 kind 时改用 SQL 模型6. 汇总一个串联全系列知识点的完整示例把增量 kind、依赖解析、区间过滤、幂等产出写在一起作为出师检验importtypingastfromdatetimeimportdatetimeimportpandasaspdfromsqlmeshimportExecutionContext,modelfromsqlmesh.core.model.kindimportModelKindNamemodel(docs_example.final_demo,kinddict(nameModelKindName.INCREMENTAL_BY_TIME_RANGE,time_columnevent_date,),columns{id:int,name:text,event_date:date,},depends_on[docs_example.upstream_model],crondaily,)defexecute(context:ExecutionContext,start:datetime,end:datetime,execution_time:datetime,**kwargs:t.Any,)-pd.DataFrame:# 1. 解析上游表名自动登记依赖tablecontext.resolve_table(docs_example.upstream_model)# 2. 只取本次时间区间内的数据 —— 保证增量与幂等dfcontext.fetchdf(fSELECT id, name, event_date FROM{table}fWHERE event_date {start} AND event_date {end})# 3. 用 pandas 做业务逻辑df[name]df[name].str.strip().str.lower()# 4. 可能为空的结果用 yield 而不是 returnifdf.empty:yieldfrom()else:yielddf结语三篇文章读完后SQLMesh Python 模型的心法浓缩成三句话一个model装饰器 一个execute函数 一个模型元数据字段与 SQL 模型一一对应schema 先于代码——columns必填且必须与返回的 DataFrame 严格一致让数据待在引擎里——能返回 Spark/Snowpark/Bigframe DataFrame 就不要落到 Pandas输出太大就用生成器分批yield。你已经跨过了初学者到工程实践的门槛。下一步建议阅读官方文档的 model kinds 与 宏系统 章节把增量策略和参数化能力用得更深。参考资料SQLMesh 官方文档 — Python modelshttps://sqlmesh.readthedocs.io/en/stable/concepts/models/python_models/