SQLMesh增量模型实战指南：时间范围分区

引言

在数据工程领域，处理大规模数据集和高频率数据更新是一项挑战。SQLMesh作为一款强大的数据编排工具，提供了增量模型功能，帮助数据工程师高效地管理和更新数据。本文将详细介绍如何使用SQLMesh创建和管理基于时间范围的增量模型，涵盖从开发到生产的完整工作流程。
在这里插入图片描述

需求背景

假设你是一名数据工程师，负责处理一家直接面向客户销售软件的公司的数据。你每天需要处理数百万笔销售交易数据，并且需要将这些数据与产品使用数据进行关联，以更好地理解销售趋势和产品使用情况。

你面临以下挑战：

如何处理延迟到达的数据？
如何处理UTC和PST时间戳的转换？
应该在什么时间运行这些任务？
如何测试这些数据？
如何高效地运行增量更新？
如何处理边缘情况下的历史数据错误？
如何编写单元测试？
如何确保生产环境的数据完整性？

本文将通过一个完整的示例，展示如何使用SQLMesh解决这些问题。

开发工作流程

在SQLMesh中，典型的开发工作流程如下：

sqlmesh plan dev: 创建一个新的开发环境
sqlmesh fetchdf: 在开发环境中预览数据
sqlmesh create_external_models: 自动生成原始源表的列级血缘文档
sqlmesh plan: 将模型从开发环境推广到生产环境
sqlmesh plan dev --forward-only: 在开发环境中进行代码更改，并仅处理新数据
sqlmesh fetchdf: 在开发环境中预览更改后的数据
sqlmesh create_test: 自动生成单元测试
sqlmesh test: 运行单元测试
sqlmesh plan: 将更改推广到生产环境

环境设置

我们将从一个现有的SQLMesh项目开始，该项目已经包含一些生产模型。假设我们已经有以下原始数据表：

原始产品使用数据

product_id	customer_id	last_usage_date	usage_count	feature_utilization_score	user_segment
PROD-101	CUST-001	2024-10-25 23:45:00+00	120	0.85	enterprise
PROD-103	CUST-001	2024-10-27 12:30:00+00	95	0.75	enterprise
…	…	…	…	…	…

原始销售数据

transaction_id	product_id	customer_id	transaction_amount	transaction_timestamp	payment_method	currency
TX-001	PROD-101	CUST-001	99.99	2024-10-25 08:30:00+00	credit_card	USD
TX-002	PROD-102	CUST-002	149.99	2024-10-25 09:45:00+00	paypal	USD
…	…	…	…	…	…	…

模型配置

我们将创建一个增量模型demo.incrementals_demo，该模型按天分区，并处理销售数据和产品使用数据的关联。

MODEL(name="demo.incrementals_demo",kind=INCREMENTAL_BY_TIME_RANGE(time_column="transaction_date",lookback=2,  # 处理过去2天的延迟数据),start="2024-10-25",  # 不回填此日期之前的数据cron="@daily",  # 每天午夜UTC运行grain="transaction_id",  # 主键audits=[UNIQUE_VALUES(columns=("transaction_id",)),NOT_NULL(columns=("transaction_id",)),]
)WITH sales_data AS (SELECTtransaction_id,product_id,customer_id,transaction_amount,transaction_timestamp,payment_method,currencyFROM sqlmesh-public-demo.tcloud_raw_data.salesWHERE transaction_timestamp BETWEEN @start_dt AND @end_dt
),product_usage AS (SELECTproduct_id,customer_id,last_usage_date,usage_count,feature_utilization_score,user_segmentFROM sqlmesh-public-demo.tcloud_raw_data.product_usageWHERE last_usage_date BETWEEN DATE_SUB(@start_dt, INTERVAL 30 DAY) AND @end_dt
)SELECTs.transaction_id,s.product_id,s.customer_id,s.transaction_amount,DATE(s.transaction_timestamp) as transaction_date,DATETIME(s.transaction_timestamp, 'America/Los_Angeles') as transaction_timestamp_pst,s.payment_method,s.currency,p.last_usage_date,p.usage_count,p.feature_utilization_score,p.user_segment,CASEWHEN p.usage_count > 100 AND p.feature_utilization_score > 0.8 THEN 'Power User'WHEN p.usage_count > 50 THEN 'Regular User'WHEN p.usage_count IS NULL THEN 'New User'ELSE 'Light User'END as user_type,DATE_DIFF(s.transaction_timestamp, p.last_usage_date, DAY) as days_since_last_usage
FROM sales_data s
LEFT JOIN product_usage pON s.product_id = p.product_idAND s.customer_id = p.customer_id

创建模型

首次创建模型时，我们需要将其添加到开发环境中：

sqlmesh plan dev

按照提示输入回填的起始和结束日期，SQLMesh将自动创建物理表并执行初始数据加载。

跟踪列级血缘

SQLMesh可以自动生成外部模型文档，记录原始表的列信息和数据类型：

sqlmesh create_external_models

通过SQLMesh UI，可以直观地查看列级血缘关系。

进行更改

假设我们需要调整“Power User”的定义，将阈值从100次使用调整为50次使用。我们可以使用--forward-only标志，仅对新数据应用更改：

sqlmesh plan dev --forward-only

SQLMesh会生成一个预览表，允许我们在开发环境中测试更改，而不会影响历史数据。

添加单元测试

使用sqlmesh create_test命令可以自动生成单元测试配置文件：

sqlmesh create_test demo.incrementals_demo \--query sqlmesh-public-demo.tcloud_raw_data.product_usage "select * from sqlmesh-public-demo.tcloud_raw_data.product_usage where customer_id='CUST-001'" \--query sqlmesh-public-demo.tcloud_raw_data.sales "select * from sqlmesh-public-demo.tcloud_raw_data.sales where customer_id='CUST-001'" \--var start_dt '2024-10-25' \--var end_dt '2024-10-27'

运行单元测试：