DAGSTER的学习#
Dagster 是一个以数据资产（asset）为中心的编排器：声明的不是"任务"，而是"数据以及生产它的函数"，DAG 由资产间的依赖自动推导。本文的例子取自我的数据平台 MVP（下称 edp），它的业务链路是：上传入湖 → 自动预标 → 人工标注 → 自动质检 → 冻结数据集 → 导出 → 训练。所有代码片段都做了简化，不依赖原项目上下文。
1. 关键概念：asset，带生产函数的数据#
一个 asset 对应一份持久化数据（一张表、一个文件、一个模型产物），以及生产这份数据的函数。函数名默认就是资产名，依赖声明决定 DAG 的边：
from dagster import asset
@asset
def episode():
"""入湖后的采集片段表。"""
...
@asset
def sample(episode): # 参数名 = 上游资产名，自动连边并注入返回值
"""从 episode 切出的训练样本。"""
...
Output 可以附带 metadata，物化后直接显示在 UI 的资产页上，适合放行数、耗时、产物路径这类观测信息：
from dagster import Output
@asset
def sample(episode) -> Output[list[str]]:
ids = do_slice(episode)
return Output(value=ids, metadata={"num_samples": len(ids)})
1.1 assert关系的两种写法：ins 与 deps#
ins=/ 函数参数：需要上游的返回值，值由 IOManager 存取（见第 5 节）。deps=：只建立血缘边，不传任何数据。
乍一看deps似乎没啥用，但是有两类场景必须用它。第一类是互斥分支的下游。edp 的质检资产同时依赖"自动标注"和"人工标注"两个上游，而这两者是二选一的分支，任何一次 run 里必有一个没物化——如果用 ins 接参数，会直接报"找不到上游输出"：
from dagster import asset, AssetKey
@asset(deps=[AssetKey("annotation_auto"), AssetKey("annotation_collect")])
def qc_result(context):
# 不从上游接参数，自己去存储里查增量：哪些标注还没有质检结果
target_ids = query_iceberg("annotation 中 review_status=passed 且无 qc_result 的行")
return run_qc(target_ids)
第二类原因和平台原则有关：“数据存不存在"应该问权威存储（这里是 Iceberg），而不是问编排器的物化记录。deps 保住图上的血缘，数据由函数自己查，两个诉求都满足。
1.2 multi_asset：产出多个资产#
当一次计算原子地产出多份数据时（edp 入湖：一个 Spark 作业同时写 raw_file/episode/sample 三张表），用 @multi_asset，图上仍是三个独立可查血缘的节点：
from dagster import multi_asset, AssetOut, Output
@multi_asset(outs={
"raw_file": AssetOut(),
"episode": AssetOut(),
"sample": AssetOut(),
})
def ingest(context):
result = run_spark_ingest(context.partition_key) # 一次 Spark 作业
yield Output(result["files"], output_name="raw_file")
yield Output(result["episode_id"], output_name="episode")
yield Output(result["sample_ids"], output_name="sample")
这里有一个容易误判的行为：multi_asset 的执行单位是整个函数。只要它的任何一个 out 被选中（被 job 圈到、或被上游触发），整个函数就会执行，所有 out 一起物化——除非显式声明 can_subset=True。因此"job 的 selection 里只列了部分资产，另一些资产也被物化了"是设计行为，不是 bug：被物化的那些资产是同一个 multi_asset 的兄弟输出。
1.3 条件物化：用 multi_asset 做路由#
分支不藏在函数内部，而是显式画在图上：两个 out 都标 is_required=False，一次运行只 yield 其中一个，另一个在 UI 上显示为 skipped。edp 的标注路由——预标后的样本，要么"预标直接转正”，要么"打包派给人工"：
@multi_asset(outs={
"annotation_auto": AssetOut(is_required=False), # 分支一：自动转正
"annotation_dispatch": AssetOut(is_required=False), # 分支二：打包派人工
})
def annotation_router(context, prelabel_annotation):
if lookup_profile(context.partition_key) == "auto_only":
yield Output(promote(prelabel_annotation), output_name="annotation_auto")
else:
pkg = package_for_labeling(prelabel_annotation)
yield Output(pkg, output_name="annotation_dispatch")
这也解释了 1.1 的坑的来源：正因为路由的两个 out 是条件物化的，它们共同的下游（qc_result）才必须用 deps。
这里有个地方需要注意assetsdefinition不等于asset
@asset / @multi_asset 装饰后的东西既不再是普通函数，也不等于"一个资产"，而是一个 **AssetsDefinition（资产定义对象）：@asset 产生的定义对象恰好定义一个资产，@multi_asset 产生的定义对象一次定义 N 个资产。这解释了顶层注册代码的一个表面矛盾——Definitions 的 assets= 列表里的元素不单纯是 asset**：
defs = Definitions(
assets=[
ingest_multi_asset, # 一个元素，带来 raw_file / episode / sample 三个资产
annotation_router, # 一个元素，带来 annotation_auto / annotation_dispatch 两个资产
qc_result, # 一个元素，一个资产
...
],
)
列表注册的是定义对象，所以列表长度和资产图上的节点数不相等。资产名（asset key）也和 Python 对象名解耦：asset key 由 name= / outs= 的键决定，ingest_multi_asset 这个变量名从不出现在图上。
2. job：asset的组合，DAG真正的入口#
2.1 Job是什么#
define_asset_job 把一组资产圈起来、起个名字。注意asset的生产逻辑、依赖关系全部在 @asset / @multi_asset 里定义完毕；job 里只有一份 asset key 清单（selection），Dagster 在加载时把这些 key 反解析回 1.4 节的那些 AssetsDefinition 单例，拼出这次 run 的可执行子图。job 也不定义执行顺序（顺序永远由资产依赖决定），它只回答"这次要物化哪些节点"，同时充当触发入口和运行历史里的标签：
from dagster import define_asset_job
ingest_job = define_asset_job(
name="ingest_job",
selection=["raw_file", "episode", "sample", # ingest 这个 multi_asset 的三个 out
"prelabel_annotation",
"annotation_auto", "annotation_dispatch",
"qc_result"],
)
围绕 selection 有四个容易与直觉冲突的规则：
- selection 写的是 asset key，不是函数名。
ingest、annotation_router这些函数名不会出现在 job 里，出现的是它们的 out。排查"某个函数为什么被触发"时，应该找的是它的输出资产名。 - 选中 multi_asset 的一个 out 等于选中整个函数（1.2 的规则在 job 里的体现）。
- 一个资产可以属于多个 job。 edp 的
qc_result既在agest_job里面（自动分支当场质检），也在手动触发的标注job里面（人工分支几天后质检）。 - 同一组 selection 可以定义多个 job。 edp 对同一条入湖链路建了
ingest_append_job和ingest_correct_job两个同构 job，唯一区别是被哪个 sensor 触发——运行历史里 job 名不同，一眼分辨这次 run 是"新增采集"还是"数据修正"。用 job 名表达结构性分支，比在函数里写 if/else 更利于排查。
还有一个更隐蔽的现象：不在任何 job 里的资产也会自己跑。edp 的汇总指标资产不属于任何 job，但挂了自动物化策略：
from dagster import asset, AssetKey, AutoMaterializePolicy
@asset(
deps=[AssetKey("episode"), AssetKey("sample"), AssetKey("dataset")],
auto_materialize_policy=AutoMaterializePolicy.eager(), # 上游一变就自动刷新
)
def analytics_summary(context):
return run_duckdb_summary()
daemon 检测到上游有新物化，就直接把它物化掉，不经过任何显式 job，运行历史里显示为隐式的 __ASSET_JOB run。定时报表、总览指标这类"上游变了就该刷新"的资产适合这种写法，省掉一条 cron。
job 也可以完全不含资产。维护性任务（比如 Iceberg 小文件合并）不值得建模成数据资产，用传统的 @op + @job：
from dagster import job, op
@op
def compact_tables(context):
for table in ["episode", "sample", "entity_tag"]:
spark.sql(f"CALL system.rewrite_data_files(table => '{table}')")
@job
def compaction_job():
compact_tables()
2.2 Job不是什么#
这一节是理解 Dagster 心智模型的关键，值得单独拿出来讲，因为最终用户是通过这张图来关联整个工作流的。
Airflow 的世界里，一个 DAG 就是一个工作流、一个调度单元，十个 DAG 就是十张互不相干的图。Dagster 不是这样：全局只有一张资产图。Definitions 把所有 AssetsDefinition 单例收集起来，按每个资产声明的 ins/deps 拼成一张图；job、sensor、schedule 都只是作用在这张图上的操作，图本身不属于任何 job。
这里有三个关键性质：
其一，跨 job 的依赖边存在于图上，但不存在于执行里。 Dagster 不支持"job A 跑完自动接着跑 job B"的跨 job 编排。edp 是最好的例子：annotation_dispatch 属于 ingest_append_job 的 selection，annotation_collect 属于完全独立的 annotation_collect_job，而 annotation_collect 的定义里声明了对 annotation_dispatch 的依赖。这两个 job 引用的是同一批模块级 Python 单例对象（1.4 节），所以这条依赖边在全局图上真实存在，UI 血缘里能看到 dispatch → collect 的连线。但执行层面这条边不通电：ingest_append_job 的 run 结束后，Dagster 既不等待也不触发 annotation_collect_job，后者必须由自己的触发器（webhook 或 sensor）独立发起。图上的连线表达的是"数据从哪来"，不是"接下来跑什么"。
其二，一个业务工作流通常横跨多个 run，靠三样东西缝合。 血缘边（deps，让图在 UI 上连续）、分区键（同一个 upload_id 作为 partition_key 串起两次 run 的物化历史）、触发器（sensor / webhook 决定下一段何时开跑）。在 UI 里点开某个分区的资产，沿血缘能看到这个 upload 在两个 job、间隔几天的两次 run 里的完整轨迹。这就是"工作流被最终用户关联"的方式：用户看的是资产图加分区历史，不是 run 列表。run 是一次性的执行痕迹，资产图才是稳定的业务视图。
其三，想让跨 job 的边自动"通电"，只有自动物化一条路。 AutoMaterializePolicy（新版叫 Declarative Automation）由 daemon 监听上游物化事件，是唯一能让"上游变了、下游自动跑"跨越 job 边界成立的机制——第 2 节的 analytics_summary 就是例子。没挂策略的边只是画在图上的血缘。
总结成一句话：图是全局的、静态的，由代码里的依赖声明决定；执行是局部的、一次性的，由 job selection 加触发器决定。读这类代码时先在脑中把全局资产图画出来，再看每个 job 圈了图上的哪一块、每个触发器在什么时机去圈，整个工作流就清楚了。
3. 触发：schedule、sensor、API#
3.1 sensor：轮询外部状态，发现活就发 RunRequest#
sensor 是一个每隔 N 秒执行一次的函数，“找活"的逻辑完全自定义——查数据库、扫对象存储目录都可以。找到活就返回 RunRequest：
from dagster import sensor, RunRequest, DefaultSensorStatus
@sensor(job=ingest_job, minimum_interval_seconds=15,
default_status=DefaultSensorStatus.RUNNING)
def ingest_sensor(context):
ready = db_query("SELECT upload_id FROM upload_session WHERE status='ready'")
return [RunRequest(run_key=f"ingest-{u}", partition_key=u) for u in ready]
run_key 是幂等键：同一个 run_key 只会真正执行一次，重复提交被 Dagster 静默跳过。因此 sensor 每次轮询把所有 ready 的会话都请求一遍也是安全的，已处理过的自动去重。
3.2 schedule：定时 + 兜底#
@schedule 处理 cron 场景。edp 用它做 T+1 兜底：sensor 守护进程万一故障，第二天早上的 cron 把漏掉的会话补上。兜底能成立的前提是——schedule 和 sensor 使用同一套 run_key 命名：
from dagster import schedule, RunRequest
@schedule(job=ingest_job, cron_schedule="0 9 * * *")
def ingest_fallback(context):
for u in pending_upload_ids():
# 与 sensor 相同的 run_key：sensor 已处理过的，这里发出去也会被跳过
yield RunRequest(run_key=f"ingest-{u}", partition_key=u)
run_key 的对齐是约定不是自动行为；两边命名不一致，兜底就变成双跑。
3.3 API 触发与人在环#
外部系统（网关、webhook、CI）通过 Dagster 的 GraphQL API 直接发起某个 job 的 run。edp 里两处在用：科研人员建数据集（带类型化 Config 参数），以及标注工具标完后的回调。
在DAG里面，如果需要等待人工参与，那么Dagster不会一直卡着等待。edp 的做法是拆成两次独立 run：“派活” run 打包、通知、把批次状态写进外部数据库后就结束；几天后标注完成，webhook 调 API 唤醒run。两次 run 用同一个分区键串起血缘。这里有一条硬约束：run 之间不共享内存，中间状态（批次的样本清单、包路径）必须落外部存储，Dagster 不会替 run 保留任何进程内状态。webhook 可能因网络问题丢失，所以再配一个低频 sensor 轮询批次状态做兜底——webhook 为主、轮询兜底是人工参与在DAG里工作流的标准组合。
3.4 daemon：所有触发的引擎#
sensor、schedule、自动物化策略全部由 dagster-daemon 进程驱动。dagster dev 会自动带起它；生产部署是三件套——webserver（UI/API）、daemon（触发）、code location（管道代码，独立进程/容器）——外加一个 PostgreSQL 存元数据。daemon 挂掉后所有触发静默停止且不报错，它本身必须被监控。
4. 分区：partition_key 是隐形的参数总线#
实际上分区的作用到底是什么？我认为这是一个非常好的问题，分区实际上是对于执行力度的把控，可以对着分区里面的数据显示出来到底哪些执行了检查之类的东西。这里的分区我认为是需要用户自己把控，自己定义的这是个非常值得深挖的问题**。我个人认为，分区提供了四个能力：**
- 管控粒度：没有分区的话，assert实际上是无状态，untracked的。
- 按分区追踪新鲜度/缺失情况：分区的管理，分区数据的追踪，很难讲
- 分区可以在初期就抓到问题，而不是那种一个动作
一句话总结：分区不是"划分渲染”，它在同一个 job/asset 内部提供的是实打实的独立重跑/新鲜度追踪能力；但在跨 job 场景下（比如本项目 annotation_dispatch → annotation_collect 这种），它意义确实不大。简单来说还是得看个人
分区也可以不用按天切数据。edp 用动态分区把"一次上传会话"建模成一个分区：
from dagster import DynamicPartitionsDefinition, SensorResult
uploads = DynamicPartitionsDefinition(name="uploads")
# sensor 里：发现新会话时先注册分区，再对该分区发起 run
@sensor(job=ingest_job)
def ingest_sensor(context):
ready = pending_upload_ids()
known = set(context.instance.get_dynamic_partitions("uploads"))
new = [u for u in ready if u not in known]
return SensorResult(
run_requests=[RunRequest(run_key=f"ingest-{u}", partition_key=u) for u in ready],
dynamic_partitions_requests=[uploads.build_add_request(new)] if new else [],
)
好处有两个。第一，链路上每个资产通过 context.partition_key 直接知道"这次处理的是哪个上传会话"，不需要把 upload_id 作为 config 层层传递——分区键成了整条链路的隐形参数总线。第二，UI 上可以按分区键圈选任意范围 backfill，“重刷某几个上传会话"变成图形界面上的框选操作。
5. IOManager：返回值的存取器#
ins/函数参数声明的上游返回值由 IOManager 负责持久化与注入：上游 Output(value=...) 被它写到存储，下游运行时再由它读出、作为参数传入。不配置时默认是 fs_io_manager——把值 pickle 到 DAGSTER_HOME 目录。
生产上的原则是编排器不搬数据本体。资产之间传递的应该是小引用——ID 列表、路径、版本号——数据本体放数据湖/数据库，下游拿引用自己去读。edp 的 sample 资产返回的是 sample_ids 列表而不是样本数据，样本本体在 Iceberg/Lance。遵守这条原则后，默认 IOManager 就够用，不需要为"传大数据"定制。
同理，数据库/对象存储的连接层也不必急着包成 Dagster Resource：无状态、懒加载的普通 Python 模块直接 import 即可。需要"同一份代码在测试/生产环境切换连接"时，再包一层 ConfigurableResource，资产函数签名不受影响。
6. 外部执行：K8s、Spark、Ray 怎么接#
前面所有例子都在 Dagster 进程里直接跑 Python 函数——本地开发够用，生产上重计算必须丢到外部集群。edp 的分层是：
orchestration/assets/ # 薄壳：声明依赖、传 run_id/partition_key、收 metadata
engines/spark/ # Spark 作业逻辑（入湖、freeze、导出）
engines/ray/ # Ray 作业逻辑（预标、质检、训练）
engines/duckdb/ # 轻量 SQL（指标汇总、打标签）
资产函数只做"调引擎、传引用、写 metadata”，一行 Spark/Ray 计算都不写在里面：
@asset(ins={"sample": AssetIn(key=AssetKey("sample"))})
def prelabel_annotation(context, sample: list[str]) -> Output[list[str]]:
from engines.ray.prelabel import run as prelabel_run # 延迟 import，方便单测 mock
result = prelabel_run(sample, run_id=context.run_id)
return Output(value=sample, metadata=result) # 只往下游传 ID，不传样本本体
engines/ 里才是"怎么起集群、怎么等完成"的地方。调用方（asset）不需要知道 Ray 是本地起的还是 K8s 上的——这正是"编排器只做控制面"在代码结构上的落点。
6.1 三种接法，按侵入性从低到高#
| 接法 | 做法 | 适合 |
|---|---|---|
| 提交-等待（edp 现用） | asset 调 engines/*.run()，引擎内部 spark-submit / ray job submit 后阻塞等结果 | Spark/Ray 作业已是独立镜像，改动最小 |
| Dagster K8s Executor | 配置 k8s_job_executor，Dagster 把每个 step 调度成一个 K8s Job Pod | 想让 Dagster 自己管 Pod 生命周期 |
| Dagster Pipes | 外部进程跑完后通过 Pipes 协议回传 metadata / 日志 | 作业镜像完全自治、不想被 Dagster 进程包住 |
大多数数据平台用第一种就够了：Dagster 进程极轻（只做参数准备和状态回写），真正的 JVM/GPU worker 全在外部集群里。
6.2 Spark on K8s：提交-等待模式#
engines/spark/ 里封装 session 构建和作业逻辑，生产环境在 build_spark_session() 里切到 K8s master：
# engines/spark/spark_session.py
def build_spark_session(app_name: str) -> SparkSession:
builder = SparkSession.builder.appName(app_name)
if settings.spark_master: # 生产：k8s://https://k8s-api:443
builder = builder.master(settings.spark_master)
return (
builder
.config("spark.kubernetes.container.image", settings.spark_image)
.config("spark.kubernetes.namespace", settings.spark_namespace)
.config("spark.executor.instances", settings.spark_executors)
# Iceberg catalog、S3 等配置照旧
.getOrCreate()
)
资产侧不变，还是一行 ingest_run(upload_id, run_id)。另一种更重的方式是用 Spark Operator（SparkApplication CRD）：Dagster asset 只负责 kubectl apply 一份 YAML 然后轮询 CR 状态——适合"一个作业就是一个独立 Spark 集群、跑完即销毁"的场景，和 edp 的"共享 session、函数内跑 SQL"是不同粒度，按作业体量选。
无论哪种，数据本体都直接写 Iceberg/OSS，函数返回值只带 episode_id、sample_ids 这类引用（呼应第 5 节）。
6.3 Ray on K8s：本地与远程同一入口#
edp 的 engines/ray/ray_utils.py 用环境变量切换：
import ray
def ensure_ray() -> None:
if ray.is_initialized():
return
if settings.ray_address: # 生产：ray://kuberay-head-svc:10001
ray.init(address=settings.ray_address, ignore_reinit_error=True)
else: # 本地：dagster dev 直接跑
ray.init(num_cpus=2, include_dashboard=False)
预标/质检/训练的具体逻辑写在 @ray.remote 函数里，run() 负责 ensure_ray() → 提交 task → ray.get() 等结果 → 写 Lance/Iceberg。GPU 作业在 K8s 侧用 KubeRay 起集群、Volcano/Kueue 做 gang scheduling 和配额——这些对 Dagster 透明，Dagster 只看到"这次 prelabel_run 花了 12 分钟、写了 800 条"。
也可以用 Ray Jobs API（ray job submit --address ...）把整份脚本当独立 Job 提交，模式和 Spark Operator 类似：Dagster 进程只等 Job 终态，不进 Ray 集群内部。
6.4 Dagster K8s Executor：让 step 本身跑在 Pod 里#
如果希望"Dagster 的每一次物化 step 都在独立 Pod 里执行"（而不是在 code location 进程里跑 Python 再内部提交），配置 dagster-k8s：
# dagster.yaml（实例级配置，不在 definitions.py 里）
run_launcher:
module: dagster_k8s.launcher
class: K8sRunLauncher
config:
job_image: my-registry/edp-code-location:latest # 含 orchestration + engines 代码
service_account_name: dagster
# definitions.py 或 dagster.yaml 的 execution 段
execution:
config:
executor:
config:
job_namespace: dagster
image_pull_policy: Always
这时每个 asset step 是一个 K8s Job Pod，Pod 里执行 asset 函数体（里面再调 engines.spark.run()）。注意：默认 FilesystemIOManager 是 Pod 本地盘，上下游 step 若落在不同 Pod 上会读不到 pickle 文件——必须换成 S3/MinIO 上的 IO Manager，或者坚持"Output 只传引用、数据走 Iceberg"让 IOManager 几乎不传大对象。
6.5 Dagster Pipes：完全外部的作业进程#
Pipes 适合"作业镜像和 Dagster 代码完全解耦"的场景：Dagster 拉起一个外部进程（K8s Job、Spark driver、Ray driver），外部进程通过环境变量里的 DAGSTER_PIPES_CONTEXT 把日志和 metadata 实时推回 UI，Dagster 侧用 PipesK8sClient 或 PipesSubprocessClient 等待：
from dagster import asset
from dagster_pipes import PipesK8sClient
@asset
def prelabel_annotation(context, sample: list[str]):
return (
PipesK8sClient()
.run(
context=context,
base_pod_spec={...}, # 预标镜像 + GPU 资源请求
command=["python", "prelabel_main.py"],
env={"SAMPLE_IDS": json.dumps(sample), "RUN_ID": context.run_id},
)
.get_results()
.get_materialize_result()
)
外部 prelabel_main.py 里用 open_dagster_pipes() 写进度和 metadata。好处是 GPU 镜像可以极大（PyTorch + CUDA），不必塞进 code location 镜像；代价是多一套 Pipes 协议适配。
6.6 路由规则：同一 asset，不同引擎#
edp 按数据量路由执行引擎（大批量 Spark、小批量 DuckDB、GPU 活 Ray），路由逻辑放在 engines/ 入口或热配置里，asset 签名不变：
# engines/router.py（示意）
def run_ingest(upload_id: str, run_id: str) -> dict:
manifest_size = estimate_manifest_bytes(upload_id)
if manifest_size < settings.duckdb_threshold:
from engines.duckdb.ingest import run
else:
from engines.spark.ingest_append import run
return run(upload_id, run_id)
Dagster 图上仍是一个 ingest 节点，换引擎不改编排定义——这正是"任务声明和执行引擎解耦"。
6.7 落地检查清单#
- run_id 贯穿审计：每个
engines.*.run()都收run_id=context.run_id，写进 Iceberg 审计列，方便从 Dagster UI 跳到数据湖里的对应批次。 - 幂等在引擎侧：Dagster 重跑 = 同一分区再物化一次，Spark/Ray 作业必须 MERGE/upsert，不能 append-only 裸写。
- IOManager 与部署方式对齐：本地
dagster dev用默认 fs 即可；K8s Executor 或跨节点 step 必须换共享 IO Manager，或坚持只传引用。 - GPU 资源不进 Dagster：Volcano/Kueue 配额在 K8s 层管，Dagster 只记耗时和产出 metadata。
7. asset check：质量与新鲜度#
@asset_check 给资产挂检查项，结果显示在资产页上；build_last_update_freshness_checks 是内置的新鲜度检查：
from datetime import timedelta
from dagster import asset_check, AssetCheckResult, AssetKey, build_last_update_freshness_checks
freshness = build_last_update_freshness_checks(
assets=[AssetKey("sample")],
lower_bound_delta=timedelta(hours=27), # T+1 窗口 24h + 3h 缓冲
)
@asset_check(asset=AssetKey("dataset"))
def dataset_quality_gate(context) -> AssetCheckResult:
meta = latest_materialization_metadata("dataset")
return AssetCheckResult(passed=meta["num_samples"] > 0, metadata=meta)
两者对应告警体系里两类不同的问题。freshness 管"该跑没跑"：失败的任务会报错，但没跑的任务不会——资产超过 27 小时没有新物化，UI 标红，这是兜住静默死亡的机制。asset check 管"跑了但对不对"。edp 里真正的质量门是引擎侧的硬性前置（不达标直接抛异常，根本不会写出数据集），asset check 只是把同一结论在 UI 上呈现为独立卡片——check 是可视化和告警面，不应是唯一防线。
8. 极小可运行示例#
把上面的概念收敛成一个最小项目：监听目录里出现的上传文件，每个文件是一个动态分区，入湖后自动刷新汇总报表。五个文件，可直接 dagster dev -f definitions.py 跑起来。
minidag/
assets.py # 资产 + 分区定义
jobs.py # job
sensors.py # 目录轮询 sensor
schedules.py # T+1 兜底
definitions.py # 入口
assets.py——一个 multi_asset（模拟一次入湖产出两份资产）+ 一个自动物化的报表：
import json
from pathlib import Path
from dagster import (AssetOut, AutoMaterializePolicy, DynamicPartitionsDefinition,
Output, asset, multi_asset)
INCOMING = Path("/tmp/minidag/incoming")
uploads = DynamicPartitionsDefinition(name="uploads")
@multi_asset(
partitions_def=uploads,
outs={"records": AssetOut(), "stats": AssetOut()},
)
def parse_upload(context):
upload_id = context.partition_key # 分区键即 upload_id
rows = json.loads((INCOMING / f"{upload_id}.json").read_text())
# 真实场景：这里把数据写进数据库/数据湖，只往下游传引用
yield Output([r["id"] for r in rows], output_name="records",
metadata={"num_rows": len(rows)})
yield Output({"num_rows": len(rows)}, output_name="stats")
@asset(deps=["records"], auto_materialize_policy=AutoMaterializePolicy.eager())
def report(context):
# 只连血缘不接参数：汇总数据自己去存储里查（这里简化为数文件）
n = len(list(INCOMING.glob("*.json")))
return Output(n, metadata={"total_uploads": n})
jobs.py——selection 写的是 out 的名字，函数名 parse_upload 不出现：
from dagster import define_asset_job
from assets import uploads
ingest_job = define_asset_job(
name="ingest_job",
selection=["records", "stats"],
partitions_def=uploads,
)
sensors.py——发现新文件：注册分区 + 发 RunRequest，run_key 保证幂等：
from dagster import DefaultSensorStatus, RunRequest, SensorResult, sensor
from assets import INCOMING, uploads
from jobs import ingest_job
@sensor(job=ingest_job, minimum_interval_seconds=10,
default_status=DefaultSensorStatus.RUNNING)
def upload_sensor(context):
seen = set(context.instance.get_dynamic_partitions("uploads"))
pending = [p.stem for p in INCOMING.glob("*.json")]
new = [u for u in pending if u not in seen]
return SensorResult(
run_requests=[RunRequest(run_key=f"ingest-{u}", partition_key=u) for u in pending],
dynamic_partitions_requests=[uploads.build_add_request(new)] if new else [],
)
schedules.py——兜底 cron，与 sensor 共用 run_key 命名：
from dagster import RunRequest, DefaultScheduleStatus, schedule
from assets import INCOMING
from jobs import ingest_job
@schedule(job=ingest_job, cron_schedule="0 9 * * *",
default_status=DefaultScheduleStatus.RUNNING)
def ingest_fallback(context):
for p in INCOMING.glob("*.json"):
yield RunRequest(run_key=f"ingest-{p.stem}", partition_key=p.stem)
definitions.py——唯一入口，全部对象注册进一个 Definitions：
from dagster import Definitions
from assets import parse_upload, report
from jobs import ingest_job
from schedules import ingest_fallback
from sensors import upload_sensor
defs = Definitions(
assets=[parse_upload, report],
jobs=[ingest_job],
sensors=[upload_sensor],
schedules=[ingest_fallback],
)
验证流程：mkdir -p /tmp/minidag/incoming && dagster dev -f definitions.py 启动后，往目录里丢一个文件：
echo '[{"id": 1}, {"id": 2}]' > /tmp/minidag/incoming/u-001.json
十秒内可以在 UI 上观察到一整条链路：sensor 注册分区 u-001 并发起 ingest_job run → records/stats 两个资产物化（metadata 里有行数）→ daemon 因为 report 的 eager 策略自动补一个隐式 run 把报表刷新。再丢一次同名文件不会重跑（run_key 去重）；丢新文件则新增分区。删掉 report 的自动物化策略后它就不再自己刷新——这个小实验能直观验证第 2 节"不在 job 里的资产为什么会跑"。
10. 生产要点#
- 单一入口：所有对象注册进一个
Definitions，开发与生产加载同一份定义。 - 部署三件套：webserver + daemon + code location，元数据放 PostgreSQL；daemon 必须被监控。
- 幂等三层：run_key（触发去重）、分区键（重跑定位）、引擎侧 MERGE/覆盖写（数据幂等）。
- 触发有兜底：sensor 为主、cron 兜底、webhook 配轮询，freshness check 兜"该跑没跑"。
- 编排器只做控制面：资产函数里只调
engines/模块、传引用不传数据本体，重计算提交到 Spark/Ray/K8s（见第 6 节）。 - 分支显式化：结构性分支用双 job 名或
is_required=False的条件物化表达，保持图和运行历史可读。
结尾#
唉，尴尬
狗头的赞赏码.jpg