Kedro 0.17.5中如何编程加载数据目录及完整项目上下文

针对你在src/tests/pipelines/my_pipeline/test_my_dataframe.py中编写单元测试的需求，分两种场景提供可直接运行的实现：

场景1：仅需加载已持久化的数据帧做校验（轻量方案） #

不需要启动完整项目上下文，直接从conf路径读取配置初始化DataCatalog即可，启动速度快，适合纯数据完整性校验类测试。

from pathlib import Path
from kedro.config import ConfigLoader
from kedro.io import DataCatalog

# 定位项目根目录：测试文件位于src/tests/pipelines/my_pipeline/下，向上3级即为项目根
PROJECT_ROOT = Path(__file__).parents[3]

# 初始化配置加载器，自动合并conf/base、conf/local下的所有catalog配置
conf_loader = ConfigLoader(conf_source=PROJECT_ROOT / "conf")
catalog_conf = conf_loader.get("catalog*", "catalog/**")

# 生成可直接使用的DataCatalog实例
catalog = DataCatalog.from_config(catalog_conf)

# 加载目标数据帧，替换为你在catalog.yml中定义的数据集名称
# target_df = catalog.load("your_dataset_name")

注意：如果你的配置存放在其他自定义环境路径，在conf_loader.get()中补充对应路径的glob规则即可。

场景2：需要加载完整项目上下文（支持节点执行、流水线运行） #

如果测试过程中需要运行节点、调用项目自定义钩子、读取参数/凭证配置、使用项目自定义数据集，通过KedroSession加载完整上下文即可，环境和kedro run运行时完全一致。

from pathlib import Path
from kedro.framework.startup import bootstrap_project
from kedro.framework.session import KedroSession

# 定位项目根目录
PROJECT_ROOT = Path(__file__).parents[3]
# 初始化项目元数据、注册自定义入口
bootstrap_project(PROJECT_ROOT)

# 创建会话，替换package_name为你src目录下的项目包名（可在src/.kedro.yml的project_package_name字段查到）
with KedroSession.create(
    package_name="your_project_package_name",
    project_path=PROJECT_ROOT,
    env="base" # 可选，指定加载的配置环境，默认加载local环境
) as session:
    context = session.load_context()
    # 直接取上下文内初始化完成的catalog，无需手动加载配置
    catalog = context.catalog
    # 加载目标数据帧
    # target_df = catalog.load("your_dataset_name")
    # 执行节点/流水线
    # context.run(node_names=["your_target_node_name"])
    # 读取项目参数
    # params = context.params

测试用例参考写法 #

可以把catalog初始化封装成pytest fixture复用：

import pytest
from pathlib import Path
from kedro.framework.startup import bootstrap_project
from kedro.framework.session import KedroSession

@pytest.fixture(scope="module")
def catalog():
    PROJECT_ROOT = Path(__file__).parents[3]
    bootstrap_project(PROJECT_ROOT)
    with KedroSession.create(
        package_name="my_project",
        project_path=PROJECT_ROOT
    ) as session:
        yield session.load_context().catalog

def test_dataframe_integrity(catalog):
    df = catalog.load("my_processed_df")
    # 编写校验逻辑
    assert not df.empty
    assert df["user_id"].is_unique
    assert {"order_amount", "pay_time"}.issubset(df.columns)

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