从零构建Dagster分区管道：时间+类别分区实战案例

分区是Dagster中的核心抽象概念，它允许我们管理大型数据集、处理增量更新并提高管道性能。本文将详细介绍如何创建和实现基于时间和类别的分区资产。

在这里插入图片描述

什么是分区？

分区是将数据集划分为更小、更易管理的部分的技术。在Dagster中，分区可以基于时间、类别或其他自定义逻辑创建，从而优化数据处理流程。

创建时间分区资产

基于时间的月度分区

首先，我们将创建一个按月份分区的资产，用于计算每个销售代表的月度绩效：

monthly_partition = dg.MonthlyPartitionsDefinition(start_date="2023-01-01")@dg.asset(partitions_def=monthly_partition,compute_kind="duckdb",group_name="analysis",deps=[joined_data]
)
def monthly_sales_performance(context: dg.AssetExecutionContext, duckdb: DuckDBResource):partition_date_str = context.partition_keymonth_to_fetch = partition_date_str[:-3]  # 格式化为YYYY-MMwith duckdb.get_connection() as conn:# 创建表（如果不存在）conn.execute("""CREATE TABLE IF NOT EXISTS monthly_sales_performance (partition_date varchar,rep_name varchar,product varchar,total_dollar_amount double);""")# 删除该月已有数据conn.execute(f"""DELETE FROM monthly_sales_performance WHERE partition_date = '{month_to_fetch}';""")# 插入新数据conn.execute(f"""INSERT INTO monthly_sales_performanceSELECT '{month_to_fetch}' AS partition_date, rep_name, product_name AS product, SUM(dollar_amount) AS total_dollar_amountFROM joined_dataWHERE strftime(date, '%Y-%m') = '{month_to_fetch}'GROUP BY '{month_to_fetch}', rep_name, product_name;""")# 预览数据preview_query = f"SELECT * FROM monthly_sales_performance WHERE partition_date = '{month_to_fetch}';"preview_df = conn.execute(preview_query).fetchdf()row_count = conn.execute(f"""SELECT COUNT(*) FROM monthly_sales_performance WHERE partition_date = '{month_to_fetch}';""").fetchone()[0] if conn.execute(f"""SELECT COUNT(*) FROM monthly_sales_performance WHERE partition_date = '{month_to_fetch}';""").fetchone() else 0return dg.MaterializeResult(metadata={"row_count": dg.MetadataValue.int(row_count),"preview": dg.MetadataValue.md(preview_df.to_markdown(index=False))})

创建类别分区资产

基于产品类别的分区

接下来，我们创建一个基于预定义产品类别的静态分区资产：

product_category_partition = dg.StaticPartitionsDefinition(["Electronics", "Books", "Home and Garden", "Clothing"
])@dg.asset(deps=[joined_data],partitions_def=product_category_partition,group_name="analysis",compute_kind="duckdb"
)
def product_performance(context: dg.AssetExecutionContext, duckdb: DuckDBResource):product_category_str = context.partition_keywith duckdb.get_connection() as conn:# 创建表（如果不存在）conn.execute("""CREATE TABLE IF NOT EXISTS product_performance (product_category varchar,product_name varchar,total_dollar_amount double,total_units_sold double);""")# 删除该类别已有数据conn.execute(f"""DELETE FROM product_performance WHERE product_category = '{product_category_str}';""")# 插入新数据conn.execute(f"""INSERT INTO product_performanceSELECT '{product_category_str}' AS product_category, product_name, SUM(dollar_amount) AS total_dollar_amount, SUM(quantity) AS total_units_soldFROM joined_dataWHERE category = '{product_category_str}'GROUP BY '{product_category_str}', product_name;""")# 预览数据preview_query = f"SELECT * FROM product_performance WHERE product_category = '{product_category_str}';"preview_df = conn.execute(preview_query).fetchdf()row_count = conn.execute(f"""SELECT COUNT(*) FROM product_performance WHERE product_category = '{product_category_str}';""").fetchone()[0] if conn.execute(f"""SELECT COUNT(*) FROM product_performance WHERE product_category = '{product_category_str}';""").fetchone() else 0return dg.MaterializeResult(metadata={"row_count": dg.MetadataValue.int(row_count),"preview": dg.MetadataValue.md(preview_df.to_markdown(index=False))})

将分区资产添加到Definitions

完成资产定义后，需要将它们添加到Dagster的Definitions对象中：

defs = dg.Definitions(assets=[products, sales_reps, sales_data, joined_data, monthly_sales_performance, product_performance,], asset_checks=[missing_dimension_check],resources={"duckdb": DuckDBResource(database="data/mydb.duckdb")}
)

物化分区资产

在Dagster UI中操作这些分区资产的步骤：

导航到"Assets"页面
点击"Reload definitions"重新加载定义
选择"monthly_sales_performance"资产，然后点击"Materialize selected"
- 确保选择所有分区
- 启动回填(backfill)作业
选择"product_performance"资产，然后点击"Materialize selected"
- 确保选择所有分区
- 启动回填作业