Dagster中SourceAsset参数传入IO Manager及相关问题排查

1. SourceAsset加载时Metadata为空 #

原因 #

SourceAsset的Metadata获取逻辑和普通Asset不同：普通Asset的Metadata可直接通过context.asset_metadata读取，但SourceAsset作为上游只读资产，其Metadata存储在资产定义本身中，context.asset_metadata仅对可写入的普通Asset生效。

解决方法 #

在IO Manager的load_input方法中，通过资产Key从仓库实例中获取对应的SourceAsset定义，进而读取其Metadata。示例代码：

from dagster import ConfigurableIOManager, AssetKey, SecretStr
import pandas as pd
import psycopg2  # 示例用PostgreSQL，可根据实际数据库替换

class DatabaseIOManager(ConfigurableIOManager):
    # 数据库凭证参数
    db_ip: str
    db_port: int
    db_user: str
    db_password: SecretStr

    def load_input(self, context):
        # 区分普通Asset和SourceAsset的Metadata获取逻辑
        if context.upstream_output:
            # 针对SourceAsset，从仓库中获取资产定义的Metadata
            asset_key = context.upstream_output.asset_key
            repo = context.instance.get_repository(context.repository_name)
            source_asset = repo.get_source_asset(asset_key)
            query_metadata = source_asset.metadata
        else:
            # 普通Asset直接读取上下文Metadata
            query_metadata = context.asset_metadata
        
        # 从Metadata中提取查询参数
        fields = query_metadata.get("fields", "*")
        filters = query_metadata.get("filters", "")
        table_name = query_metadata["table_name"]  # 假设Metadata中必传表名
        query = f"SELECT {fields} FROM {table_name} {filters}"
        
        # 连接数据库并返回DataFrame
        conn = psycopg2.connect(
            host=self.db_ip,
            port=self.db_port,
            user=self.db_user,
            password=self.db_password.get_secret_value()
        )
        df = pd.read_sql(query, conn)
        conn.close()
        return df

同时确保SourceAsset定义时正确传入Metadata：

from dagster import SourceAsset, AssetKey

my_source_asset = SourceAsset(
    key=AssetKey("user_data"),
    metadata={
        "table_name": "public.users",
        "fields": "id, username, email",
        "filters": "WHERE is_active = true"
    }
)

2. 自定义DatabaseConfig类显示不可序列化 #

原因 #

自定义配置类未继承Dagster官方的Config基类，或包含不可序列化的字段类型（如自定义对象、未加密的敏感数据）。

解决方法 #

1. 直接让IO Manager继承ConfigurableIOManager（内置配置序列化逻辑），无需单独定义Config类；敏感字段使用SecretStr类型，既保证序列化安全，又能正常获取值：

from dagster import ConfigurableIOManager, SecretStr

class DatabaseIOManager(ConfigurableIOManager):
    db_ip: str
    db_port: int
    db_user: str
    db_password: SecretStr  # 敏感字段用SecretStr处理

    # 如需自定义JSON序列化（可选，用于返回非敏感配置）
    def json(self):
        return {
            "db_ip": self.db_ip,
            "db_port": self.db_port,
            "db_user": self.db_user
        }

2. 若必须单独定义配置类，需继承dagster.Config：

from dagster import Config, SecretStr

class DatabaseConfig(Config):
    db_ip: str
    db_port: int
    db_user: str
    db_password: SecretStr

3. Dagit中IO Manager使用计数显示为0 #

原因 #

• IO Manager未通过io_manager_key正确关联到资产定义
• 关联的资产从未被执行过（计数基于实际运行的作业/资产实例）

解决方法 #

1. 确保资产定义时指定正确的io_manager_key，并在Definitions中注册资源：

from dagster import asset, Definitions

@asset(io_manager_key="database_io_manager")
def processed_user_data(user_data):
    # 资产处理逻辑
    return user_data.dropna()

defs = Definitions(
    assets=[processed_user_data, my_source_asset],
    resources={
        "database_io_manager": DatabaseIOManager(
            db_ip="127.0.0.1",
            db_port=5432,
            db_user="admin",
            db_password=SecretStr("your_password")
        )
    }
)

2. 运行一次关联的资产（通过Dagit手动触发或调度执行），执行完成后使用计数会自动更新。

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