Kedro Pipeline 对象完全指南从节点编排到依赖解析的原理与实践【免费下载链接】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 将数据任务抽象为Node节点而Pipeline管道则负责把一组节点组织成有依赖关系的可执行工作流。本文以 Kedro 仓库中 Understanding the pipeline object 一文为主体结合 Pipeline 类源码 与 测试用例系统讲解如何构建管道、合并管道、查看执行顺序与输入输出、打标签以及如何规避坏管道读完你可以独立编写、排查并组织 Kedro 项目中的管道代码。什么是 Pipeline 对象Pipeline是 Kedro 中用于组织节点依赖关系与执行顺序的核心数据结构。节点指南 将节点定义为表示任务的构建块管道则把一组节点组合成完整工作流它负责编排节点的依赖和执行顺序连接输入与输出同时保持代码的模块化。Kedro 的核心特性是自动依赖解析管道根据每个节点声明的输入、输出变量推导节点间的先后关系而不一定按照节点传入列表的顺序执行。要做到这一点只需把节点链入一个Pipeline对象——它本质上是一个共享同一组变量的节点列表可以通过Pipeline构造函数基于节点或其他管道实例化传入管道时其全部节点会被展开并入新管道。Pipeline类定义在 kedro/pipeline/pipeline.py其模块注释将其描述为可作为有向无环图DAG顺序或并行执行的Node对象集合并提供对输入依赖、产出输出与执行顺序的快速访问。在 kedro/pipeline/init.py 中Pipeline与函数式 APIpipeline一起被导出两者的构造参数完全一致pipeline()本质上是Pipeline()的薄封装见 pipeline.py#L1218-L1284。从源码看Pipeline构造函数支持以下参数见 pipeline.py#L133-L143参数类型作用nodesIterable[Node \| Pipeline] \| Pipeline管道包含的节点或子管道传入管道会展开为节点列表inputsstr \| set[str] \| dict[str, str]暴露给上游管道连接点的输入名称可选不传则从管道结构自动推断outputsstr \| set[str] \| dict[str, str]暴露给下游管道连接点的输出名称可选不传则自动推断parametersstr \| set[str] \| dict[str, str]需要命名空间化的参数名可省略params:前缀tagsstr \| Iterable[str]应用于管道内所有节点的标签集合namespacestr \| None除显式指定的inputs/outputs与params:/parameters之外所有数据集名称的前缀prefix_datasets_with_namespacebool是否用命名空间前缀节点数据集默认True构造时管道会执行一系列校验节点名必须唯一_validate_duplicate_nodes、输出不得重复_validate_unique_outputs、confirms 不得重复_validate_unique_confirms并使用TopologicalSorter检测环见下文循环依赖一节。如何构建一个管道以下示例构造了一个计算一组数字方差的小管道。实际项目中节点的函数定义往往更复杂涉及的变量通常对应整个数据集from kedro.pipeline import Pipeline, Node def mean(xs, n): return sum(xs) / n def mean_sos(xs, n): return sum(x**2 for x in xs) / n def variance(m, m2): return m2 - m * m variance_pipeline Pipeline( [ Node(len, xs, n), Node(mean, [xs, n], m, namemean_node), Node(mean_sos, [xs, n], m2, namemean_sos), Node(variance, [m, m2], v, namevariance_node), ] )Kedro 根据每个节点声明的输入、输出确定执行顺序。本例中第一个节点计算xs的长度输出为n第二个节点用xs与n计算均值输出为m第三个节点用xs与n计算均方和mean_sos输出为m2第四个节点用均值m与均方和m2计算方差输出为v。Kedro 的依赖解析算法保证每个节点在其所需输入由前置节点产出之后才执行因此节点自动按正确顺序运行无需人工排序。Node的输入、输出既可以是单个字符串也可以是列表、字典乃至None无输入或无输出详见 node.py 中 Node 构造签名。字典形式允许把数据集名映射为函数参数名例如Node(func, {arg_name: dataset_name}, ...)。使用describe查看管道执行顺序describe()方法返回管道节点的概览及其执行顺序print(variance_pipeline.describe())输出如下#### Pipeline execution order #### Name: None Inputs: xs len([xs]) - [n] mean_node mean_sos variance_node Outputs: v ##################################在默认的names_onlyTrue模式下输出只显示节点名称传入describe(names_onlyFalse)则显示完整的函数名([输入]) - [输出]形式。从 describe 的实现 可见其节点顺序取自拓扑排序结果self.nodesInputs/Outputs则分别来自inputs()与outputs()排序后的set_to_string会按字母序输出空集显示为None。该字符串通常配合日志使用例如logger.info(pipeline.describe())源码 docstring 示例。如何合并多个管道多个管道可以合并pipeline_de和pipeline_ds会被展开为其底层节点列表再合并在一起pipeline_de Pipeline([Node(len, xs, n), Node(mean, [xs, n], m)]) pipeline_ds Pipeline( [Node(mean_sos, [xs, n], m2), Node(variance, [m, m2], v)] ) last_node Node(print, v, None) pipeline_all Pipeline([pipeline_de, pipeline_ds, last_node]) print(pipeline_all.describe())输出如下#### Pipeline execution order #### Name: None Inputs: xs len([xs]) - [n] mean([n,xs]) - [m] mean_sos([n,xs]) - [m2] variance([m,m2]) - [v] print([v]) - None Outputs: None注意此例合并后所有节点都没有name因此describe按节点字符串函数签名形式展示在Pipeline构造函数中传入子Pipeline对象会被展开见 pipeline.py#L247-L251 的chain.from_iterable展开逻辑。除构造函数合并外Pipeline还重载了集合运算符见 pipeline.py#L372-L395与|合并两个管道的节点等价于Pipeline(set(self._nodes other._nodes))-返回self中有而other中没有的节点构成的新管道返回两个管道共有节点构成的新管道。这些运算符同样支持将管道作为集合元素参与运算便于按需裁剪工作流。获取管道中的节点信息管道以拓扑顺序暴露其节点列表供管道可视化等自定义功能使用每个节点都携带输入、输出信息nodes variance_pipeline.nodes nodes输出如下[ Node(len, xs, n, None), Node(mean, [xs, n], m, mean_node), Node(mean_sos, [xs, n], m2, mean_sos), Node(variance, [m, m2], v, variance node), ]查看单个节点的输入nodes[0].inputs输出[xs]从 nodes 属性的实现 看拓扑顺序由grouped_nodespipeline.py#L548-L565逐层取出就绪节点生成并缓存若节点 A 必须先于节点 B 运行则 A 一定出现在 B 之前。测试 tests/pipeline/test_pipeline.py#L287 亦验证了节点以拓扑顺序返回。获取管道的输入与输出与上节类似可以用inputs()和outputs()查询管道的自由输入与最终输出variance_pipeline.inputs()输出Out[7]: {xs}variance_pipeline.outputs()输出Out[8]: {v}从实现看inputs()返回运行管道必须提供的自由输入不含管道内部节点产生又被内部消费的中间输入outputs()返回整个管道运行后产出的最终输出不含被其他节点消费的中间输出二者都会解析转码transcoding后的名称见 pipeline.py#L397-L452。另外all_inputs()/all_outputs()返回所有节点输入输出的并集datasets()返回管道用到的全部数据集。如何为管道打标签可以通过tags参数为管道打标签该标签会应用到管道内每一个节点pipeline Pipeline( [Node(..., namenode1), Node(..., namenode2)], tagspipeline_tag )管道标签也可以与节点标签叠加使用。下面的例子中node1和node2都带有pipeline_tag而node2额外带有node_tagpipeline Pipeline( [Node(..., namenode1), Node(..., namenode2, tagsnode_tag)], tagspipeline_tag, )从 构造函数实现 看传入的tags会被归一化为集合_to_list(tags)随后对每个节点调用n.tag(_tags)完成标签传播Pipeline.tag()方法pipeline.py#L1049-L1059也提供了对已构造管道整体打标签的能力。标签的实战价值在于筛选only_nodes_with_tagspipeline.py#L941-L956可按任意一个标签命中筛选节点而filter()pipeline.py#L958-L1047支持将tags、from_nodes、to_nodes、node_names、from_inputs、to_outputs、node_namespaces等条件求交集一次性生成满足全部条件的子管道。如何避免创建坏管道管道通常能够解析其依赖但在某些情况下解析无法完成此时管道被视为构造不良not well-formed。构造阶段Pipeline.__init__会执行多种校验从 pipeline.py#L239-L290 可见依次包括nodes为空、节点名重复、转码输入输出非法、输出重复、confirms 重复以及拓扑排序环检测。坏节点bad nodes假设管道中只有一个既无输入也无输出的节点try: Pipeline([Node(lambda: print(!), None, None)]) except Exception as e: print(e)输出Invalid Node definition: it must have some inputs or outputs. Format should be: Node(function, inputs, outputs)Node要求函数至少有一个输入或输出见 node.py 的校验逻辑 及后续检查否则无法建立任何依赖关系。循环依赖circular dependencies对任意两个变量若第一个依赖第二个则第二个不能反过来依赖第一个否则循环依赖会阻止管道编译。下面第一个节点表达了由x计算y第二个节点表达了由y计算x二者无法共存于同一管道try: Pipeline( [ Node(lambda x: x 1, x, y, namefirst node), Node(lambda y: y - 1, y, x, namesecond node), ] ) except Exception as e: print(e)输出Circular dependencies exist among these items: [first node: lambda([x]) - [y], second node: lambda([y]) - [x]]检测原理上Pipeline在构造时基于node_dependenciespipeline.py#L516-L531节点→其直接父节点的映射构建graphlib.TopologicalSorter并立即调用prepare()试探排序一旦CycleError被抛出就抛出CircularDependencyErrorpipeline.py#L277-L285。对应测试 tests/pipeline/test_pipeline.py#L500-L523 构造了A→B→C→A的闭环并断言抛出CircularDependencyError。使用点号命名节点的隐患点号命名的节点可能产生怪异行为Pipeline([Node(lambda x: x, inputsinput1kedro, outputsoutput1.kedro)])凡输入或输出名称中含有.的节点都有造成管道断裂disconnected或 Kedro 结构格式错误的风险。原因是.在 Kedro 内部有特殊含义——它表示命名空间管道namespace pipeline。在上例中输出段应被视为断裂名称暗示存在一个output1命名空间管道。输入没有命名空间而输出通过点号被命名空间化导致 Kedro 分别处理二者。更稳妥的写法是input1_kedro与output1_kedro。Kedro 建议使用_而非.作为名称分隔符。如何在 Kedro 项目中组织管道代码管理 Kedro 项目时建议将相关任务分组到各自独立的管道中以达成模块化。一个项目通常包含多个任务把频繁执行的任务分别组织成独立管道有助于维持秩序与效率。每个管道最好放在自己的文件夹中便于在项目内复制与复用——简言之一个管道一个文件夹。为此Kedro 引入了 模块化管道Modular Pipelines 的概念详细介绍见下一节。管道定义好后通常在项目的pipeline_registry.py中登记供运行时按名称发现与执行参见 pipeline_registry.md 与 运行管道指南管道内部的节点命名空间与合并策略可参考 namespaces.md 与 slice_a_pipeline.md。小结Pipeline是 Kedro 工作流编排的核心对象依据节点输入输出自动进行依赖解析与拓扑排序节点执行顺序由依赖决定而非传入顺序。describe()、nodes、inputs()、outputs()提供执行顺序、节点明细与管道边界自由输入/最终输出的完整视图。管道可互相合并构造函数展开或/|/-/运算符tags标签可传播到全部节点并配合filter()、only_nodes_with_tags等进行子管道筛选。构造时即校验无输入输出的坏节点、循环依赖、重复节点名/输出、点号命名等都会在创建阶段直接报错杜绝运行时才暴露的结构性问题。【免费下载链接】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），仅供参考