在数据工程和机器学习工作流中，​​资产（Assets）​​是Dagster的核心构建块，而资产间的数据传递直接影响管道的可维护性和扩展性。本文深入探讨了Dagster中三种主流的数据传递方式，结合代码示例和优缺点分析，帮助开发者根据场景选择最优策略，构建高效、可扩展的数据管道。

在数据工程和机器学习工作流中，**资产（Assets）**是Dagster的核心构建块，而资产间的数据传递是构建高效、可维护管道的关键挑战。本文深入探讨了Dagster中三种主流的数据传递方式：

1. 显式管理数据（通过外部存储如SQLite）；
2. 隐式管理数据（利用I/O管理器如DuckDBPandasIOManager）；
3. 避免跨资产传递数据（通过合并任务为单一资产）。

通过具体代码示例和优缺点分析，本文旨在帮助开发者根据场景选择最适合的数据传递策略，从而提升管道的可扩展性和可维护性。

1. 显式管理数据：外部存储的直接控制

核心思想

通过外部数据库（如SQLite、PostgreSQL）或文件系统直接读写数据，资产间通过共享存储位置实现数据传递。依赖关系通过deps参数显式声明。

代码示例

import sqlite3
import dagster as dg

@dg.asset
def asset1():
    with sqlite3.connect("database.sqlite") as conn:
        conn.execute("CREATE OR REPLACE TABLE test (i INTEGER)")
        conn.execute("INSERT INTO test VALUES (42)")

@dg.asset(deps=[asset1])  # 显式声明依赖
def asset2(context: dg.AssetExecutionContext):
    with sqlite3.connect("database.sqlite") as conn:
        result = conn.execute("SELECT * FROM test").fetchall()
        context.log.info(result)

defs = dg.Definitions(assets=[asset1, asset2])

优势

• 灵活性高：可自由选择存储介质（如本地文件、云数据库），适应不同环境需求。
• 透明性强：数据存储和读取逻辑清晰可见，便于调试。

局限性

• 手动管理成本高：需自行处理连接、事务及错误（如数据库宕机）。
• 潜在一致性问题：多资产并发访问时需额外设计锁机制。

适用场景：需要强控制权或依赖特定存储系统的场景（如合规要求本地存储）。

2. 隐式管理数据：I/O管理器的自动化

核心思想

通过I/O管理器抽象数据读写逻辑，资产仅需关注业务计算，无需关心底层存储细节。Dagster自动处理依赖和数据传递。

代码示例

import pandas as pd
from dagster_duckdb_pandas import DuckDBPandasIOManager
import dagster as dg

duckdb_io_manager = DuckDBPandasIOManager(database="my_database.duckdb")

@dg.asset
def people():
    return pd.DataFrame({"id": [1, 2, 3], "name": ["Alice", "Bob", "Charlie"]})

@dg.asset
def birds():
    return pd.DataFrame({"id": [1, 2, 3], "name": ["Bluebird", "Robin", "Eagle"]})

@dg.asset
def combined_data(people, birds):  # 自动注入依赖资产的数据
    return pd.concat([people, birds])

defs = dg.Definitions(
    assets=[people, birds, combined_data],
    resources={"io_manager": duckdb_io_manager}
)

优势

• 低代码冗余：无需手动编写读写逻辑，聚焦核心计算。
• 环境适配性强：通过更换I/O管理器（如S3、BigQuery）轻松切换存储后端。

局限性

• 灵活性受限：默认行为可能无法满足定制化需求（如特殊命名规则）。
• 调试复杂度：数据流隐藏在I/O管理器中，问题排查需深入配置。

适用场景：标准化数据存储需求（如团队统一使用DuckDB或Snowflake）。

3. 避免跨资产传递：合并任务的单一资产化

核心思想

将多个关联任务（如下载→解压→加载）合并为一个原子资产，通过函数内变量传递数据，而非依赖外部存储。

代码示例

from typing import List
import dagster as dg

def download_files() -> str: ...
def unzip_files(zipfile: str) -> List[str]: ...
def load_data(files: List[str]): ...

@dg.asset
def my_dataset():
    zipped_files = download_files()
    files = unzip_files(zipped_files)
    load_data(files)

优势

• 性能优化：减少I/O开销，适合内存可容纳的小规模数据。
• 简化架构：逻辑内聚，降低跨资产耦合度。

局限性

• 可复用性差：函数绑定于单一资产，难以在其他管道中复用。
• 资源依赖性强：假设本地有足够存储和内存，不适合大数据场景。

适用场景：原型开发或确定性强的轻量级流程（如每日小规模ETL）。

总结与选型建议

策略	适用场景	关键权衡
显式外部存储	需要强控制或合规性要求的复杂存储系统	灵活性↑，维护成本↑
I/O管理器	标准化存储需求，追求开发效率	简洁性↑，定制性↓
合并资产	小数据量、低延迟要求的端到端流程	性能↑，可复用性↓

实践启示：

• 混合使用策略：例如用I/O管理器处理主体数据流，显式存储处理临时中间结果。
• 监控与测试：无论哪种策略，均需通过Dagster的日志和测试工具验证数据一致性。

通过合理选择数据传递方式，开发者可以构建出既高效又易于维护的Dagster管道，从容应对从简单ETL到复杂机器学习工作流的挑战。