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

news2025/5/14 8:16:32

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

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

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

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

在这里插入图片描述

实现步骤详解

1. 定义事件驱动型资产

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

from dagster import asset, MaterializeResult, Config
import duckdb

class AdhocRequestConfig(Config):
    """请求参数配置"""
    department: str
    product: str
    start_date: str
    end_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_sales
        FROM joined_data
        WHERE 
            date >= '{config.start_date}' AND 
            date < '{config.end_date}' AND
            department = '{config.department}' AND
            product_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, AssetGroup

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