Dagster资产间数据传递与用户配置管理教程

from dagster import Config

class FruitConfig(Config):
    fruit_select: str

# 错误示例：不正确的资产定义
import pandas as pd
# ... 其他导入 ...
from dagster import asset, Config

# ... generate_dataset 资产定义 (与正确示例相同，略) ...

class fruit_config(Config):
    fruit_select: str 

@asset(deps=[generate_dataset]) # deps在这里用于数据传递是错误的
def filter_data(config: fruit_config):
    # 错误：不应在此处调用上游资产来获取数据
    df = generate_dataset() 
    df2 = df[df['fruit'] == config.fruit_select]
    print(df2)
    return df2

@asset(deps=[filter_data]) # deps在这里用于数据传递是错误的
def filter_again():
    # 错误：不应在此处调用上游资产来获取数据
    df2 = filter_data() 
    df3 = df2[df2['units'] > 5]
    print(df3)
    return df3

import pandas as pd
import random
from datetime import datetime, timedelta
from dagster import asset, Config, materialize # materialize用于本地测试

# 1. 定义生成原始数据集的资产
@asset 
def generate_dataset() -> pd.DataFrame:
    """
    生成一个包含水果、单位和日期的随机数据集。
    """
    def random_dates(start_date, end_date, n=10):
        date_range = end_date - start_date
        return [start_date + timedelta(days=random.randint(0, date_range.days)) for _ in range(n)]

    random.seed(42) # 保证可复现性
    num_rows = 100
    fruits = ['Apple', 'Banana', 'Orange', 'Grapes', 'Kiwi']

    df = pd.DataFrame({
        'fruit': [random.choice(fruits) for _ in range(num_rows)],
        'units': [random.randint(1, 10) for _ in range(num_rows)],
        'date': random_dates(datetime(2022, 1, 1), datetime(2022, 12, 31), num_rows)
    })
    print("Generated Dataset:")
    print(df.head())
    return df

# 2. 定义用户配置类
class FruitConfig(Config):
    """
    用户自定义配置，用于选择要筛选的水果。
    """
    fruit_select: str 

# 3. 定义筛选数据的资产，接收上游资产输出和用户配置
@asset 
def filter_data(generate_dataset: pd.DataFrame, config: FruitConfig) -> pd.DataFrame:
    """
    根据用户配置的fruit_select筛选数据集。

    Args:
        generate_dataset (pd.DataFrame): 上游资产generate_dataset的输出。
        config (FruitConfig): 用户提供的配置对象。

    Returns:
        pd.DataFrame: 筛选后的数据集。
    """
    # 直接使用传入的generate_dataset参数
    df_filtered = generate_dataset[generate_dataset['fruit'] == config.fruit_select]
    print(f"\nFiltered Data for '{config.fruit_select}':")
    print(df_filtered.head())
    return df_filtered

# 4. 定义再次筛选的资产，接收上游资产输出
@asset
def filter_again(filter_data: pd.DataFrame) -> pd.DataFrame:
    """
    对上游filter_data资产的输出进行二次筛选（单位大于5）。

    Args:
        filter_data (pd.DataFrame): 上游资产filter_data的输出。

    Returns:
        pd.DataFrame: 再次筛选后的数据集。
    """
    # 直接使用传入的filter_data参数
    df_final = filter_data[filter_data['units'] > 5]
    print("\nFinal Filtered Data (units > 5):")
    print(df_final.head())
    return df_final

# 示例：如何在本地运行包含配置的资产
if __name__ == "__main__":
    # 使用materialize函数在本地运行资产
    # 传递Config的方式是嵌套在'ops'字典中，对应资产名和'config'键
    result = materialize(
        [generate_dataset, filter_data, filter_again],
        run_config={
            "ops": {
                "filter_data": {
                    "config": {
                        "fruit_select": "Banana" # 用户在此处定义参数
                    }
                }
            }
        }
    )
    assert result.success
    print("\nPipeline executed successfully!")