Dagster教程：解决资产间数据传递与配置参数化错误

dagster._core.errors.DagsterInvalidConfigError: Error in config for op Error 1: Missing required config entry "config" at the root. Sample config for missing entry: {'config': {'fruit_select': '...'}}

import pandas as pd
from dagster import asset, Config

@asset
def generate_dataset() -> pd.DataFrame:
    # ... (生成DataFrame的逻辑)
    df = pd.DataFrame(...)
    return df

class fruit_config(Config):
    fruit_select: str

@asset
def filter_data(generate_dataset: pd.DataFrame, config: fruit_config) -> pd.DataFrame:
    # generate_dataset 参数会自动接收上游同名资产的输出
    # config 参数会自动接收运行时提供的 fruit_config 配置
    filtered_df = generate_dataset[generate_dataset['fruit'] == config.fruit_select]
    print(f"Filtered data by fruit '{config.fruit_select}':\n{filtered_df}")
    return filtered_df

@asset
def filter_again(filter_data: pd.DataFrame) -> pd.DataFrame:
    # filter_data 参数会自动接收上游同名资产的输出
    final_df = filter_data[filter_data['units'] > 5]
    print(f"Further filtered data by units > 5:\n{final_df}")
    return final_df

import pandas as pd
import random
from datetime import datetime, timedelta
from dagster import asset, Config, materialize, Definitions

# 辅助函数：生成随机日期
def random_dates(start_date, end_date, n=10):
    date_range = end_date - start_date
    random_dates_list = [start_date + timedelta(days=random.randint(0, date_range.days)) for _ in range(n)]
    return random_dates_list

@asset
def generate_dataset() -> pd.DataFrame:
    """
    生成一个包含水果、单位和日期的随机数据集。
    """
    random.seed(42) # 设置种子以保证可复现性
    num_rows = 100
    fruits = ['Apple', 'Banana', 'Orange', 'Grapes', 'Kiwi']
    fruit_column = [random.choice(fruits) for _ in range(num_rows)]
    units_column = [random.randint(1, 10) for _ in range(num_rows)]
    start_date = datetime(2022, 1, 1)
    end_date = datetime(2022, 12, 31)
    date_column = random_dates(start_date, end_date, num_rows)

    df = pd.DataFrame({
        'fruit': fruit_column,
        'units': units_column,
        'date': date_column
    })
    print("Generated Dataset Head:\n", df.head())
    return df

class FruitConfig(Config):
    """
    用于筛选水果的配置类。
    """
    fruit_select: str

@asset
def filter_data(generate_dataset: pd.DataFrame, config: FruitConfig) -> pd.DataFrame:
    """
    根据用户配置的水果类型筛选数据集。
    """
    print(f"Filtering data for fruit: '{config.fruit_select}'")
    filtered_df = generate_dataset[generate_dataset['fruit'] == config.fruit_select]
    print(f"Filtered Data Head (fruit='{config.fruit_select}'):\n", filtered_df.head())
    return filtered_df

@asset
def filter_again(filter_data: pd.DataFrame) -> pd.DataFrame:
    """
    进一步筛选单位大于5的数据。
    """
    print("Further filtering data for units > 5")
    final_df = filter_data[filter_data['units'] > 5]
    print("Final Filtered Data Head (units > 5):\n", final_df.head())
    return final_df

# 将资产组织到 Definitions 中，以便在Dagster UI中加载
defs = Definitions(
    assets=[generate_dataset, filter_data, filter_again]
)

# 如果需要在本地直接运行（不通过UI），可以这样调用：
# if __name__ == "__main__":
#     # 示例运行配置
#     run_config = {
#         "ops": {
#             "filter_data": {
#                 "config": {
#                     "fruit_select": "Banana"
#                 }
#             }
#         }
#     }
#     # 注意：对于资产，配置应该在 resources 或 ops 级别提供，
#     # 如果是单个资产，通常通过 `asset_name: { config: {...} }` 结构
#     # 在 Dagster 2.0+ 中，推荐使用 `Definitions` 和 `materialize` 函数
#     # 或者通过 `dagster dev` 在 UI 中运行并提供配置
#     # 对于本地测试，需要构建一个 Job 或使用更高级的测试模式
#
#     # 简单起见，这里不再提供直接的 materialize 示例，
#     # 因为主要目的是展示资产定义，并在Dagster UI中运行。
#     # 在UI中，当运行包含 filter_data 的资产组时，会提示输入 fruit_select。