当前位置：首页 > news >正文

告别定时任务！用Dagster监听器实现秒级数据响应自动化

news 来源：原创 2025/5/20 22:25:06

在数据管道开发中，我们经常面临需要根据外部事件触发计算任务的场景。传统基于时间的调度方式存在资源浪费和时效性不足的问题。本文将通过Dagster的**传感器（Sensor）**功能，演示如何构建事件驱动的数据处理流程。

场景模拟：动态销售报表生成系统

假设业务部门需要实时获取特定产品在指定时间段的销售分析报表。传统方案需要人工手动触发任务，而我们希望通过以下方式实现自动化：

当新的销售请求文件到达时自动触发计算
根据请求参数动态生成报表
仅在检测到有效请求时运行作业

在这里插入图片描述

实现步骤详解

1. 定义事件驱动型资产

首先创建一个接收动态参数的资产，该资产将根据请求参数查询数据仓库生成报表：

from dagster import asset, MaterializeResult, Config
import duckdbclass AdhocRequestConfig(Config):"""请求参数配置"""department: strproduct: strstart_date: strend_date: str@asset(deps=["joined_data"], compute_kind="Python")
def adhoc_request(config: AdhocRequestConfig,duckdb: duckdb.DuckDBResource
) -> MaterializeResult:"""动态销售报表生成"""query = f"""SELECT department, rep_name, product_name, SUM(dollar_amount) AS total_salesFROM joined_dataWHERE date >= '{config.start_date}' AND date < '{config.end_date}' ANDdepartment = '{config.department}' ANDproduct_name = '{config.product}'GROUP BY department, rep_name, product_name"""with duckdb.get_connection() as conn:preview_df = conn.execute(query).fetchdf()return MaterializeResult(metadata={"preview": MaterializeResult.MetadataValue.md(preview_df.to_markdown(index=False))})

2. 构建事件监听传感器

使用@sensor装饰器创建传感器，持续监控指定目录下的请求文件：

import os
import json
from dagster import sensor, SensorEvaluationContext, RunRequest@sensor(job=adhoc_request_job)
def adhoc_request_sensor(context: SensorEvaluationContext):"""请求文件监听传感器"""requests_dir = os.path.join(os.path.dirname(__file__), "../data/requests")current_state = {}for filename in os.listdir(requests_dir):if filename.endswith(".json"):file_path = os.path.join(requests_dir, filename)file_mtime = os.path.getmtime(file_path)# 检测新文件或修改过的文件if filename not in current_state or current_state[filename] != file_mtime:with open(file_path) as f:request_config = json.load(f)# 生成唯一运行标识run_key = f"adhoc_request_{filename}_{file_mtime}"yield RunRequest(run_key=run_key,run_config={"ops": {"adhoc_request": {"config": request_config}}})current_state[filename] = file_mtime

3. 部署与测试

更新Dagster定义文件并启动服务：

from dagster import Definitions, AssetGroupdefs = Definitions(assets=[adhoc_request],sensors=[adhoc_request_sensor],resources={"duckdb": duckdb.DuckDBResource(database="data/mydb.duckdb")}
)