如何为Dagster中的软件定义资产添加物化运行时追踪？是否有内置功能？

不用手动写计时样板代码，Dagster提供了更规范的实现方式，以下两种方案可以满足需求：

1. 利用AssetExecutionContext内置时间戳 #

每个资产执行时的上下文（AssetExecutionContext）已经自动记录了执行开始和结束时间，直接通过上下文计算耗时即可：

from dagster import asset, Output, AssetExecutionContext

@asset
def my_asset(context: AssetExecutionContext):
    # 你的业务逻辑
    x = 1 + 1
    # 计算并添加耗时元数据
    duration = context.execution_time - context.start_time
    return Output(x, metadata={"duration": duration.total_seconds()})

2. 自定义通用装饰器批量处理 #

如果需要给多个资产统一添加耗时追踪，可封装一个装饰器，避免重复代码：

from dagster import asset, Output, AssetExecutionContext
import functools
import time

def track_duration(func):
    @functools.wraps(func)
    def wrapper(context: AssetExecutionContext, *args, **kwargs):
        start_time = time.time()
        result = func(context, *args, **kwargs)
        duration = time.time() - start_time
        
        # 兼容原函数返回Output或普通值的情况
        if isinstance(result, Output):
            updated_metadata = {**(result.metadata or {}), "duration": duration}
            return Output(result.value, metadata=updated_metadata)
        else:
            return Output(result, metadata={"duration": duration})
    return wrapper

# 给资产添加追踪
@track_duration
@asset
def my_asset(context: AssetExecutionContext):
    return 1 + 1

@track_duration
@asset
def another_asset(context: AssetExecutionContext):
    return [i for i in range(1000)]

补充：Dagster UI默认会展示每个资产的执行时长，若需将时长作为元数据持久化（用于后续分析、规则校验等），上述方案均可实现。

内容的提问来源于stack exchange，提问作者MYK