4、Op深入：Op的输入与输出定义，使用@op装饰器，多输入/多输出Op，Op的配置与上下文

好，咱们进入第四讲。前面我们搭了个简单的管道，知道了 Op 是 Dagster 里的最小执行单元。但说实话，那个 Hello World 级别的 Op 太简单了，根本不够看。这一章，咱们把 Op 的里里外外扒个干净——输入输出怎么定义、多进多出怎么玩、配置怎么传、上下文怎么用。这些都是你写真实管道时每天都要面对的东西。

4.1 输入与输出：Op 的“接口契约”

每个 Op 本质上就是一个函数。但和普通函数不同，Dagster 要求你明确声明这个 Op 要吃什么（输入）、要吐什么（输出）。为什么？因为只有这样，Dagster 才能帮你做依赖分析、并行调度、缓存复用。说白了，这是给框架看的“接口契约”。

核心原则：输入输出定义决定了 Op 在 DAG 中的位置。一个 Op 的输出名，必须和下游 Op 的输入名匹配，Dagster 才能把它们连起来。

4.1.1 基本定义方式

最简单的写法，就是直接在函数参数里写类型注解。Dagster 会自动推断输入输出。

from dagster import op, Out, In

@op
def hello_world():
    return "Hello"

@op
def greet(name: str) -> str:
    return f"Hello, {name}!"

嗯，这里要注意：@op 装饰器默认把函数参数当作输入，把返回值当作一个名为 result 的输出。如果你只有一个返回值，这样写完全够用。

但如果你想要更明确的控制，可以用 In 和 Out 对象来定义。我个人习惯在复杂管道里用这种方式，因为可读性更好。

@op(
    ins={"name": In(dagster_type=str)},
    out=Out(dagster_type=str)
)
def greet_explicit(name: str) -> str:
    return f"Hello, {name}!"

4.1.2 输入输出的类型系统

Dagster 内置了一套类型系统。除了 Python 原生类型，你还可以用 DagsterType 自定义校验逻辑。我在项目中遇到过一个问题：有个同事传了个空列表进来，下游直接崩了。后来加了个自定义类型做非空校验，问题就解决了。

类型 说明 使用场景
str 字符串 文本数据、文件路径
int 整数 计数、ID
float 浮点数 指标、阈值
list 列表 批量数据
dict 字典 结构化配置
pd.DataFrame Pandas DataFrame 数据分析管道

小技巧：如果你用 Pandas 做数据处理，建议在 Op 的输入输出上明确标注 pd.DataFrame 类型。这样 Dagster 会在执行前做类型检查，避免运行时才发现类型不匹配。

4.2 多输入/多输出 Op：一个 Op 搞定复杂逻辑

现实中的数据管道很少是“一个输入一个输出”的简单链条。更多时候，一个 Op 需要从多个源头拉数据，或者一个 Op 要产出多个结果。这时候就需要多输入/多输出 Op 了。

4.2.1 多输入 Op

定义多输入 Op 很简单，直接在函数参数里加就行。Dagster 会自动把参数名映射为输入名。

@op
def merge_data(users: list, orders: list) -> dict:
    # 合并用户和订单数据
    result = {"user_count": len(users), "order_count": len(orders)}
    return result

如果你想显式控制输入名（比如参数名和输入名不一致时），可以用 In 的 input_name 参数。不过说实话，我很少用这个，保持命名一致更省心。

4.2.2 多输出 Op

多输出稍微复杂一点。你需要用 Out 定义多个输出，然后在函数里返回一个字典，key 对应输出名。

from dagster import op, Out, Output

@op(
    out={
        "valid_records": Out(dagster_type=list),
        "invalid_records": Out(dagster_type=list),
        "summary": Out(dagster_type=dict)
    }
)
def split_records(records: list):
    valid = []
    invalid = []
    for r in records:
        if r.get("is_valid"):
            valid.append(r)
        else:
            invalid.append(r)
    
    yield Output(valid, "valid_records")
    yield Output(invalid, "invalid_records")
    yield Output({
        "total": len(records),
        "valid_count": len(valid),
        "invalid_count": len(invalid)
    }, "summary")

这里用了 yield Output() 的写法。为什么不用 return？因为多输出需要逐个声明。我曾经踩过这个坑——一开始用 return 返回字典，结果下游只拿到了最后一个输出。记住：多输出必须用 yield Output()。

避坑指南：多输出 Op 里，yield 的顺序不重要，Dagster 是根据 Output 的第二个参数（输出名）来匹配的。但为了代码可读性，建议保持和 out 定义一致的顺序。

4.3 Op 的配置：让 Op 更灵活

写死的 Op 没什么用。真实场景下，同一个 Op 在不同环境（开发/测试/生产）可能需要不同的参数。比如数据库连接串、文件路径、阈值等。Dagster 的配置系统就是干这个的。

4.3.1 使用 Config 定义配置

Dagster 推荐用 Config 类来定义配置结构。这样既有类型检查，又有自动补全。

from dagster import op, Config

class FileProcessConfig(Config):
    input_path: str
    output_path: str
    batch_size: int = 1000  # 默认值
    encoding: str = "utf-8"

@op
def process_file(config: FileProcessConfig):
    print(f"读取文件: {config.input_path}")
    print(f"写入文件: {config.output_path}")
    print(f"批次大小: {config.batch_size}")
    # 实际处理逻辑...

调用时，通过 config 参数传入配置值：

from dagster import job

@job
def my_job():
    process_file()

# 运行时传入配置
result = my_job.execute_in_process(
    run_config={
        "ops": {
            "process_file": {
                "config": {
                    "input_path": "/data/input.csv",
                    "output_path": "/data/output.csv",
                    "batch_size": 500
                }
            }
        }
    }
)

我个人习惯把配置定义放在 Op 所在的文件顶部，或者单独一个 configs.py 文件里。这样配置和逻辑分离，维护起来清爽很多。

4.3.2 配置的校验与默认值

Config 类继承自 Pydantic 的 BaseModel，所以天然支持类型校验和默认值。你还可以加自定义校验逻辑。

from dagster import op, Config
from pydantic import validator

class DatabaseConfig(Config):
    host: str
    port: int = 5432
    database: str
    username: str
    password: str
    
    @validator("port")
    def check_port(cls, v):
        if v < 1024 or v > 65535:
            raise ValueError("端口号必须在 1024-65535 之间")
        return v

小技巧：敏感信息（如密码）不要硬编码在配置里。可以用环境变量或 Dagster 的 EnvVar 来注入。我在生产环境就是这么做的，安全又灵活。

4.4 Op 上下文：OpExecutionContext 的妙用

每个 Op 在执行时，Dagster 都会给它一个上下文对象。这个对象里装着运行时的各种信息——日志、资源、分区信息、运行 ID 等等。你只需要在函数参数里声明 context 参数，Dagster 就会自动注入。

4.4.1 上下文的核心能力

上下文属性 类型 说明
context.log Logger 结构化日志，支持 info/warning/error
context.resources Resources 访问外部资源（数据库、API 等）
context.op_config Config 当前 Op 的配置对象
context.run_id str 当前运行的唯一 ID
context.partition_key str 当前分区的 key（如果启用了分区）

4.4.2 实战：用上下文做日志和监控

我在项目中遇到过一个问题：管道跑了好几个小时，突然崩了，但不知道崩在哪一步。后来我养成了一个习惯——每个 Op 都用 context.log 记录关键步骤。

from dagster import op, OpExecutionContext

@op
def data_quality_check(context: OpExecutionContext, data: list) -> dict:
    context.log.info(f"开始数据质量检查，共 {len(data)} 条记录")
    
    issues = []
    for i, record in enumerate(data):
        if not record.get("id"):
            issues.append(f"第 {i} 条记录缺少 ID")
    
    if issues:
        context.log.warning(f"发现 {len(issues)} 个数据质量问题")
        for issue in issues[:5]:  # 只记录前5个
            context.log.warning(f"  问题详情: {issue}")
    else:
        context.log.info("数据质量检查通过")
    
    return {
        "total": len(data),
        "issues": len(issues),
        "run_id": context.run_id
    }

你想想看，有了这些日志，排查问题是不是方便多了？Dagster 的日志还会自动关联到对应的 Op 和运行 ID，你可以在 UI 上直接按运行 ID 过滤。

4.4.3 上下文与配置的结合

上下文和配置经常一起用。通过 context.op_config 可以拿到当前 Op 的配置，这样你就不用把配置参数传来传去了。

@op
def export_report(context: OpExecutionContext, data: list):
    config = context.op_config  # 类型就是你在 @op 里定义的 Config 类
    context.log.info(f"导出报告到 {config.output_path}")
    # 导出逻辑...

避坑指南：不要在 Op 内部修改 context.op_config 的值。配置应该是只读的。我曾经手贱改了一次，结果同一个 Op 在不同分片执行时配置乱掉了。记住：配置是输入，不是状态。

4.5 本章知识体系总览

说了这么多，咱们用一张图把 Op 的核心概念串起来。这张图涵盖了输入输出、配置、上下文的关系，以及它们如何协同工作。

Op 核心概念关系图 @op 装饰器 输入定义 In() / 参数注解 输出定义 Out() / yield Output() 配置系统 Config 类 / run_config 执行上下文 OpExecutionContext 数据流入 数据流出 配置注入 运行时信息 关键设计原则 1. 输入输出定义接口契约，决定 Op 在 DAG 中的位置 2. 配置让 Op 可复用，上下文让 Op 可观测 3. 多输出用 yield Output()，配置用 Config 类

这张图把 Op 的四个核心维度都串起来了。你写 Op 的时候，脑子里要有这张图：输入从哪里来、输出到哪里去、配置怎么传、上下文怎么用。想清楚这四点，你的 Op 设计就不会跑偏。

← 上一章 📖 返回目录 下一章 →