Kedro 会话生命周期管理KedroSession与KedroServiceSession实战指南【免费下载链接】kedroKedro is a toolbox for production-ready data science. It uses software engineering best practices to help you create data engineering and data science pipelines that are reproducible, maintainable, and modular.项目地址: https://gitcode.com/GitHub_Trending/ke/kedroKedro 的会话Session机制负责管理一次或多次 Kedro 运行的完整生命周期将库组件由KedroContext管理与运行时的静态、动态数据解耦。本文以 docs/extend/session.md 为骨架结合kedro/framework/session/源码与测试讲解KedroSession单次运行与KedroServiceSession服务化多运行的创建参数、调用方式以及bootstrap_project/configure_project的项目初始化流程。读完本文你将能在脚本、Notebook 或 Web 服务中正确创建会话并按需执行管道。概览会话为什么存在一次 Kedro 运行涉及两类截然不同的对象库组件由KedroContext统一管理负责配置加载、Catalog 构建、钩子调度等会话数据包括静态数据项目路径、环境名、运行时参数与动态数据CLI 上下文、Git 信息、用户名、异常信息等。会话Session将二者解耦使 Kedro 组件与插件可以无需导入KedroContext对象即可访问会话数据。Kedro 提供两个基于抽象基类AbstractSession定义于 kedro/framework/session/abstract_session.py实现的会话类特性KedroSessionKedroServiceSession定位单次运行single-run多运行multi-run服务场景生命周期一次会话对应一次管道运行一个会话内可多次运行管道每次可传不同运行时参数典型场景CLIkedro run、脚本化单次执行Web 服务 / API 中保持会话存活、按需触发运行状态持久化运行时参数与对应 session ID不落盘 session 数据close()仅记录日志KedroSession会捕获会话元数据包括CLI 上下文、环境信息、运行时参数、用户名以及项目 Git 状态而KedroServiceSession处于活跃开发中可能偶发破坏性变更官方鼓励试用并反馈意见该提示同时出现在 service_session.py 的类 docstring 中。两个类共有的核心方法create()携带会话数据创建会话实例load_context()实例化KedroContext。对于KedroServiceSession此方法接受可选参数runtime_params用于更新该次运行的KedroContext参数close()关闭当前会话。KedroSession在save_on_closeTrue时还会把会话数据保存到磁盘run()以给定参数运行管道详见 Running pipelines。在 abstract_session.py 中AbstractSession实现了上下文管理器协议__enter__返回自身__exit__调用close()这就是两个会话类都能直接配合with语句使用的原因。创建KedroSession单次运行以下代码以上下文管理器方式创建KedroSession并在上下文内运行管道。脚本可以从 Kedro 项目中的任意位置调用——find_kedro_project会从当前目录向上逐级寻找含pyproject.toml且带[tool.kedro]配置的项目根目录因此无需硬编码路径退出with块时会话自动关闭from pathlib import Path from kedro.framework.session import KedroSession from kedro.framework.startup import bootstrap_project from kedro.utils import find_kedro_project # Get project root current_dir Path(__file__).resolve().parent project_root find_kedro_project(current_dir) bootstrap_project(Path(project_root)) # Create and use the session with KedroSession.create(project_pathproject_root) as session: session.run()KedroSession.create()的可选参数如下参数类型/默认值说明project_pathPath \| str \| None项目根目录路径默认取Path.cwd()源码见 session.pysave_on_closebool默认Truecreate()内默认会话关闭时是否将 session 数据保存到磁盘envstr \| NoneKedroContext使用的环境名不传时读取环境变量KEDRO_ENVruntime_paramsdict \| None传递给底层KedroContext的运行时项目参数指定后会更新并优先于项目配置中读取的参数conf_sourcestr \| NoneKedroContext的配置来源目录默认取settings.CONF_SOURCE即项目的conf/从源码看create()内部流程如下session.py调用validate_settings()校验项目配置就绪以generate_timestamp()生成时间戳作为 session ID构造会话并初始化 session store组装session_dataproject_path、session_id、CLI 上下文_jsonify_cli_context、env、runtime_params、当前用户名getpass.getuser()、项目 Git 状态_describe_git含commit_sha与dirty标志将全部数据写入session._store。_describe_git与_jsonify_cli_context的实现可分别参见 session.py 与 session.py它们是“捕获会话元数据”这一能力的落地。会话数据持久化与 session storeKedroSession的close()在save_on_closeTrue时调用self._store.save()session.py。store 由_init_store()根据settings.SESSION_STORE_CLASS实例化默认是 store.py 中的BaseSessionStore默认存储路径为project_root/sessions。注意BaseSessionStore本身是不落盘的临时实现——read()返回空 dict、save()仅记录调试日志真正持久化由自定义 store 子类完成可通过项目settings.py中的SESSION_STORE_CLASS与SESSION_STORE_ARGS替换对应 settings 定义见 kedro/framework/project/init.py。一次会话只允许一次运行KedroSession.run()内部通过self._run_called标志强制“会话与运行 1:1 映射”——若同一会话被调用第二次会抛出KedroSessionErrorsession.py。这正是需要KedroServiceSession的原因服务化场景必须能在同一会话中多次运行。创建KedroServiceSession服务化多运行以下代码创建KedroServiceSession在同一会话中先后以不同运行时参数运行两次管道最后手动关闭会话from pathlib import Path from kedro.framework.session import KedroServiceSession from kedro.framework.startup import bootstrap_project from kedro.utils import find_kedro_project # Get project root current_dir Path(__file__).resolve().parent project_root find_kedro_project(current_dir) bootstrap_project(Path(project_root)) # Create and use the session session KedroServiceSession.create(project_pathproject_root) # first run session.run(runtime_params{param1: value1}) # second run with different runtime parameters session.run(runtime_params{param1: value2}) # close the session when done session.close()KedroServiceSession.create()的可选参数参数类型/默认值说明session_idstr \| None会话标识符不传时自动生成 UUIDstr(uuid.uuid4())见 service_session.pyproject_pathPath \| str \| None项目根目录路径envstr \| NoneKedroContext使用的环境名不传时读取KEDRO_ENVconf_sourcestr \| None配置来源目录默认conf/serving_modebool默认False会话是否处理并发run()调用为True时在create()阶段急切预加载全部管道以避免并发竞争条件create()与KedroSession的差异KedroServiceSession没有save_on_close参数——服务会话不落盘其close()仅输出关闭日志service_session.pyKedroServiceSession没有create()级runtime_params参数——运行时参数改由每次run()传入从而为每一次具体运行独立更新KedroContext参数。与之对应run()也接受独立的run_id参数默认用generate_timestamp()生成每次运行都有独立标识。KedroServiceSession.load_context(runtime_paramsNone)会构造一个全新的 config loader 并注入该次运行的参数service_session.py测试 tests/framework/session/test_service_session.py 验证了load_context(runtime_params{param1: value1})后context.config_loader.runtime_params即为该值。serving_mode并发安全的服务化运行serving_modeTrue的语义在源码中有精细实现_enable_serving_mode()先调用_preload_pipelines()成功之后才把_serving_mode置位保证预加载失败时会话仍停留在普通 CLI 模式而非不一致状态service_session.py预加载时通过pipelines.set_requested(None)list(pipelines)将pipeline_registry.py中注册的全部管道填入共享单例pipelines._contentservice_session.py运行阶段serving 模式跳过set_requested()避免并发请求间互相改写共享管道注册表而普通模式下set_requested()会告诉懒加载器只导入需要的管道加快 CLI 启动速度service_session.py安全加固serving 模式下 config loader 被强制设置restrict_runtime_params_type_selectionTrue——因为运行时参数可能来自不可信的 HTTP 请求体项目自身的CONFIG_LOADER_ARGS也无法削弱该限制service_session.py对应 issue kedro-org/kedro#5706。这些行为均有测试覆盖test_serving_mode_preloads_all_pipelines、test_serving_mode_flag_stays_false_when_preload_fails 以及用 20 线程ThreadPoolExecutor压测并发run()的 test_serving_mode_concurrent_runs_do_not_raise。bootstrap_project与configure_project运行前的项目初始化Kedro 的 CLI 在kedro run启动时已自动执行这两个函数因此日常命令行使用无需手动调用只有以编程方式脚本、服务、Notebook与 Kedro 项目交互时才需要。若想在 Notebook 等交互式环境中加载 Kedro 项目也可以直接使用%reload_kedroline magic参见 Kedro and Notebooks。整体启动流程如下两个函数都负责 Kedro 项目的初始化区别在于bootstrap_project项目模式内部调用configure_project并额外读取pyproject.toml的[tool.kedro]段获得package_name、project_name、kedro_init_version等元数据把项目源码目录默认src/加入sys.path并写入PYTHONPATH使项目可作为 Python 包被导入。实现见 startup.py 与_add_src_to_pathstartup.py。适合直接操作项目源码的开发场景。configure_project打包模式读取项目的settings.py与pipeline_registry.py在 Kedro 运行前注册配置与管道并记录全局PACKAGE_NAMEkedro/framework/project/init.py。如果你的 Kedro 项目是打包安装的执行管道前需调用configure_project。ValueError: package name not foundValueError: Package name not found. Make sure you have configured the project using bootstrap_project. This should happen automatically if you are using Kedro command line interface.该错误由validate_settings()抛出kedro/framework/project/init.py当全局PACKAGE_NAME为None即项目尚未被bootstrap_project/configure_project配置时即触发。使用 CLI 时无需担心但使用multiprocessing时需特别注意。取决于操作系统Python 有不同的多进程启动方式fork/spawn/forkserver。若进程以spawn方式启动Python 会在每个子进程中重新导入所有模块此时必须在新进程启动处再次调用configure_project。Kedro 在ParallelRunner中正是这样处理的例如 kedro/runner/task.pyif ( multiprocessing.get_start_method() in (spawn, forkserver) and package_name ): Task._bootstrap_subprocess(package_name, logging_config)_bootstrap_subprocess在子进程中重新执行configure_project(package_name)并可选重放日志配置task.py从而保证子进程也能正确解析项目包。关于各启动方式的取舍可参考 Run a pipeline 中对fork/forkserver/spawn的对比说明。实战选择建议CLI / 批处理 / 单次脚本默认使用kedro run由 CLI 自动完成初始化与KedroSession生命周期管理需要编程式控制时用KedroSession.create()with上下文管理器并注意一次会话仅一次run()。Web 服务 / API / 按需触发使用KedroServiceSession在同一会话中多次run()并每次注入不同runtime_params若服务会并发处理请求务必设置serving_modeTrue让管道在创建期一次性预加载同时获得运行时参数驱动 Catalog 类型选择的强制安全限制。多进程/分布式凡使用spawn启动方式子进程中必须重新调用configure_project或bootstrap_project否则会遇到 “Package name not found” 错误。【免费下载链接】kedroKedro is a toolbox for production-ready data science. It uses software engineering best practices to help you create data engineering and data science pipelines that are reproducible, maintainable, and modular.项目地址: https://gitcode.com/GitHub_Trending/ke/kedro创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考