3、核心概念（下）：深入理解 I/O Manager、Resource（资源）、Config（配置）与 Launchpad

上一章我们聊了 Asset、Op 和 Graph，算是把 Dagster 的骨架搭起来了。但光有骨架还不够——数据从哪读、写到哪去？外部系统怎么连？参数怎么调？这些才是真正让管道跑起来的血肉。

这一章，我们就来啃这三个硬核概念：I/O Manager、Resource 和 Config。再加上一个很实用的调试工具 Launchpad。嗯，内容不少，但都是日常开发高频接触的东西。

3.1 I/O Manager：数据读写的统一管家

先问一个问题：你的数据存在哪？CSV 文件？Parquet？数据库？S3？

传统做法是在每个 Op 里写死读写逻辑。今天用本地 CSV，明天换 Hive，得改一堆代码。我早期做数据管道时就被这事坑过——一个项目从开发环境迁移到生产环境，光改数据路径就改了一整天，还漏了两个地方，导致线上数据写错了位置。

I/O Manager 就是来解决这个问题的。它把数据的「读」和「写」抽象成统一接口。你的 Op 只需要说「我要读 asset_A」「我要写 asset_B」，至于数据到底存在哪、用什么格式，全由 I/O Manager 决定。

3.1.1 内置 I/O Manager

Dagster 自带了好几个开箱即用的 I/O Manager：

名称	存储后端	适用场景
FilesystemIOManager	本地文件系统	开发调试、单机运行
S3IOManager	AWS S3	云上生产环境
GCSIOManager	Google Cloud Storage	GCP 生态
ADLS2IOManager	Azure Data Lake	Azure 生态
DBIOManager	关系型数据库	需要 SQL 查询的场景

用法很简单。在 Definitions 里配置一下：

from dagster import Definitions, FilesystemIOManager, asset

@asset
def my_data():
    return [1, 2, 3]

defs = Definitions(
    assets=[my_data],
    resources={
        "io_manager": FilesystemIOManager(base_dir="/data/pipeline")
    }
)

这样，my_data 这个 asset 的输出会自动写到 /data/pipeline/my_data 下。下次再读它，I/O Manager 自动从同一路径加载回来。

3.1.2 自定义 I/O Manager

内置的满足不了你？那就自己写一个。我有个项目需要把数据写到 Elasticsearch 里做实时检索，内置的显然不行。

自定义 I/O Manager 需要继承 IOManager 并实现两个方法：

from dagster import IOManager, io_manager, OutputContext, InputContext

class MyElasticsearchIOManager(IOManager):
    def __init__(self, host: str, port: int):
        self.host = host
        self.port = port
        # 实际项目中在这里初始化 ES 客户端

    def handle_output(self, context: OutputContext, obj):
        # 把 obj 写入 Elasticsearch
        index = context.asset_key.path[-1]
        # 写入逻辑...
        pass

    def load_input(self, context: InputContext):
        # 从 Elasticsearch 读取数据
        index = context.asset_key.path[-1]
        # 读取逻辑...
        return []

@io_manager(required_resource_keys={"es_config"})
def my_es_io_manager(init_context):
    host = init_context.resources.es_config["host"]
    port = init_context.resources.es_config["port"]
    return MyElasticsearchIOManager(host, port)

3.2 Resource：管理外部系统的连接

Resource 是什么？说白了就是你的管道需要的外部资源——数据库连接、API 客户端、SparkSession、Redis 连接池……这些东西。

你可能会问：这不就是 I/O Manager 吗？

不一样。I/O Manager 管的是「数据读写」，Resource 管的是「外部连接」。一个 Resource 可以被多个 I/O Manager 或 Op 共享。举个例子：你的管道同时读写 PostgreSQL 和 Elasticsearch，它们共用一个 logging_client Resource 来记录操作日志。

3.2.1 定义和使用 Resource

定义一个 Resource 很简单：

from dagster import resource

@resource
def my_db_connection(init_context):
    # 初始化连接
    conn = create_database_connection(
        host=init_context.resources.config["host"],
        port=init_context.resources.config["port"]
    )
    yield conn
    # 管道结束后自动关闭
    conn.close()

注意这里用了 yield 而不是 return。这是 Python 的生成器模式——yield 之前的代码在资源初始化时执行，yield 之后的代码在资源销毁时执行。非常适合做「获取连接 → 使用 → 释放」这个流程。

在 Op 里怎么用？通过 context.resources：

@op(required_resource_keys={"my_db"})
def query_users(context):
    conn = context.resources.my_db
    return conn.execute("SELECT * FROM users").fetchall()

3.2.2 Resource 的配置与依赖

Resource 可以嵌套。一个 Resource 依赖另一个 Resource，这在真实项目中很常见：

@resource
def db_pool(init_context):
    # 依赖一个配置 Resource 来获取连接参数
    config = init_context.resources.db_config
    pool = create_pool(config["host"], config["port"])
    yield pool
    pool.close()

defs = Definitions(
    assets=[...],
    resources={
        "db_config": {"host": "localhost", "port": 5432},
        "my_db": db_pool,
    }
)

3.3 Config：让管道参数化

写死的管道是没有灵魂的。同一个管道，今天跑昨天的数据，明天跑今天的数据，参数得能调才行。

Config 就是 Dagster 的参数化方案。它分两种：Op Config 和 Asset Config。

3.3.1 Op Config

给 Op 加配置，用 @op(config_schema=...)：

from dagster import op, OpExecutionContext

@op(config_schema={"date": str, "limit": int})
def fetch_orders(context: OpExecutionContext):
    date = context.op_config["date"]
    limit = context.op_config["limit"]
    # 用 date 和 limit 去查数据
    return get_orders(date, limit)

运行的时候传配置：

job = JobDefinition(graph=graph, ...)
result = job.execute_in_process(
    run_config={
        "ops": {
            "fetch_orders": {
                "config": {
                    "date": "2024-01-01",
                    "limit": 1000
                }
            }
        }
    }
)

3.3.2 Asset Config

Asset 的配置方式类似，但更推荐用 Config 类来定义，类型安全：

from dagster import asset, Config

class OrderFetchConfig(Config):
    date: str
    limit: int = 1000  # 默认值

@asset
def daily_orders(config: OrderFetchConfig):
    return fetch_orders(config.date, config.limit)

我个人更偏爱这种写法。用类定义配置，IDE 能自动补全，类型检查也能提前发现问题。曾经有个同事把日期格式传错了，如果是字典配置，得等到运行时才能发现；用 Config 类的话，Python 类型检查直接就报错了。

3.4 Launchpad：Dagster 的配置调试利器

说到配置，就不得不提 Launchpad。它是 Dagster UI 里的一个交互式配置编辑器。

你想想看，每次调试管道都要在代码里改 run_config，然后重新运行，多麻烦。Launchpad 让你在 UI 里直接编辑配置，实时生效。

打开方式：在 Dagster UI 里点击某个 Job 或 Asset，右侧面板就能看到 Launchpad 标签。

它会自动根据你的 config_schema 生成一个 JSON 编辑器，并且会做校验。配置写错了？它会标红提示你。我经常用它来快速测试不同的参数组合，比改代码再运行快太多了。

3.5 三者如何协同工作

最后，我们用一个完整的例子把 I/O Manager、Resource 和 Config 串起来：

from dagster import asset, Config, resource, IOManager, Definitions

# 1. 定义配置
class DataProcessConfig(Config):
    source_table: str
    target_table: str

# 2. 定义 Resource（数据库连接池）
@resource
def pg_pool(init_context):
    pool = create_pool(
        host=init_context.resources.db_config["host"],
        port=init_context.resources.db_config["port"]
    )
    yield pool
    pool.close()

# 3. 自定义 I/O Manager（读写 PostgreSQL）
class PostgresIOManager(IOManager):
    def __init__(self, pool):
        self.pool = pool

    def handle_output(self, context, obj):
        table = context.asset_key.path[-1]
        with self.pool.getconn() as conn:
            conn.execute(f"INSERT INTO {table} VALUES ...", obj)

    def load_input(self, context):
        table = context.asset_key.path[-1]
        with self.pool.getconn() as conn:
            return conn.execute(f"SELECT * FROM {table}").fetchall()

@io_manager(required_resource_keys={"pg_pool"})
def pg_io_manager(init_context):
    return PostgresIOManager(init_context.resources.pg_pool)

# 4. Asset 使用配置
@asset
def processed_data(config: DataProcessConfig):
    # 业务逻辑：从 source_table 读，处理后写到 target_table
    # 读写由 I/O Manager 自动处理
    return transform(raw_data)

defs = Definitions(
    assets=[processed_data],
    resources={
        "db_config": {"host": "localhost", "port": 5432},
        "pg_pool": pg_pool,
        "io_manager": pg_io_manager,
    }
)

这个例子里：

• Config 告诉管道要处理哪张表
• Resource 管理数据库连接池
• I/O Manager 自动处理数据的读写
• Asset 只关心业务逻辑，其他一概不管

嗯，这就是 Dagster 的设计哲学——关注点分离。每个组件只做一件事，但组合起来就能构建出健壮、可维护的数据管道。