Kedro Data Catalog 完全指南从 catalog.yml 配置到源码级运行原理【免费下载链接】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 中的Data Catalog是项目所有数据源的注册中心它以catalog.yml文件为入口将节点的输入输出名称映射为DataCatalog类中的具体数据集对象。本文以 Data Catalog 介绍 为主线完整梳理 catalog 的核心概念、catalog.yml的全部配置维度类型、路径、参数、凭据、版本化、多环境覆盖、数据集工厂、懒加载机制并结合 DataCatalog 源码 解释这些配置在底层是如何被解析和执行的。读完本文你将能够独立完成 Kedro 项目数据层的注册、定制与调试。Data Catalog 是什么在 Kedro 项目中Data Catalog 是所有可供项目使用的数据源的注册表。它由一个 YAML 文件catalog.yml承载把节点输入和输出的名称作为DataCatalog类的键key与具体的数据集实现一一对应。节点通过名称引用数据Kedro 再通过 Catalog 把名称解析为实际的文件、表或对象存储资源。两个与版本相关的关键事实需要先了解从 Kedro0.19.0起具体的数据集实现如CSVDataset不再包含在 Kedro 核心包中需要从kedro-datasets包导入从kedro-datasets2.0.0起所有数据集名称中的大写 SDataSet改为小写 sDataset例如CSVDataSet现在写作CSVDataset。kedro-datasets为常见文件类型与文件系统提供了开箱即用的数据集实现是 catalog 配置中type字段的主要来源。catalog.yml基础注册第一批数据集Data Catalog 的入口文件是catalog.yml通常位于conf/base/目录下。以下示例来自 set_up_data 教程注册了两个 CSV 数据集和一个 Excel 数据集companies: type: pandas.CSVDataset filepath: data/01_raw/companies.csv reviews: type: pandas.CSVDataset filepath: data/01_raw/reviews.csv shuttles: type: pandas.ExcelDataset filepath: data/01_raw/shuttles.xlsx load_args: engine: openpyxl # Use modern Excel engine (the default since Kedro 0.18.0)在本地文件系统读取或保存文件时每个条目需要三个要素数据集名称key顶层键作为节点输入输出引用的标识符type声明使用的数据集类filepath文件位置。配置数据集参数catalog.yml中数据集配置的层次结构为顶层键是数据集名称例如shuttles、weather下一层包含多个键第一个必填键是type声明数据集类型其余键是数据集参数因实现而异部分数据集参数可以进一步由底层库决定例如shuttles的load_args由 pandas 加载 CSV 的选项定义而weather的save_args由 Snowpark 的saveAsTable方法定义。shuttles: # Dataset name type: pandas.ExcelDataset # Dataset type filepath: data/01_raw/shuttles.xlsx # pandas.ExcelDataset parameter load_args: # pandas.ExcelDataset parameter engine: openpyxl # Pandas option for loading CSV files weather: # Dataset name type: snowflake.SnowparkTableDataset # Dataset type table_name: weather_data database: meteorology schema: observations credentials: snowflake_client save_args: # snowflake.SnowparkTableDataset parameter mode: overwrite # Snowpark saveAsTable input option column_order: name table_type: 注意Kedro 数据集会把load_args/save_args直接委托给底层实现。完整参数列表请查阅kedro-datasets文档中对应数据集类的__init__方法其中会给出对底层库 API如pandas.read_excel的参数引用。数据集type与filepath连接任意数据存储type支持的数据集类型Kedro 支持连接 CSV、Excel、Parquet、Feather、HDF5、JSON、pickle 对象、SQL 表、SQL 查询等多种数据形态底层依赖 pandas、PySpark、NetworkX、Matplotlib 等库。完整列表见kedro-datasets文档。filepath基于 fsspec 的协议寻址Kedro 依赖fsspec从各类数据存储读取和写入数据。filepath应使用protocol://path/to/data的通用形式若不写协议默认按本地文件系统处理等价于file://。可用协议包括协议说明file://本地或网络文件系统默认协议允许相对路径hdfs://userserver:port/path/to/dataHadoop 分布式文件系统HDFS面向集群内高可靠、副本文件s3://my-bucket-name/path/to/dataAmazon S3 远程二进制存储常与 EC2 搭配基于 s3fs 库s3://my-bucket-name/path/to/dataS3 兼容存储如 MinIO同样基于 s3fsgcs://Google Cloud Storage基于 gcsfsabfs://Azure Blob Storage / Azure Data Lake Storage Gen2http:///https://直接从 HTTP Web 服务器读取数据fsspec还提供 SSH、FTP、WebHDFS 等其他文件系统实现。深入配置load_args / save_args / fs_args / validator除type和filepath外catalog 还接受几组影响数据加载、保存和访问方式的设置load_args和save_args控制底层第三方库如何加载/保存数据。例如pandas.CSVDataset的load_args会作为关键字参数传给pd.read_csvsave_args传给pd.DataFrame.to_csvcars: type: pandas.CSVDataset filepath: data/01_raw/company/cars.csv load_args: sep: , save_args: index: False date_format: %Y-%m-%d %H:%M decimal: .validator在每次加载或保存时按 schema 或自定义规则校验数据详见 数据集校验。fs_args控制 Kedro 与文件系统本身的交互。顶层键传给底层文件系统类例如 GCS 的GCSFileSystem而open_args_load和open_args_save传给文件系统的open方法控制文件在加载/保存时如何被打开test_dataset: type: ... fs_args: project: test_project # 传给 GCSFileSystem open_args_load: mode: r encoding: utf-8 open_args_save: mode: a # 追加模式保存每个数据集实现的默认加载、保存和文件系统参数定义在实现类中的DEFAULT_LOAD_ARGS、DEFAULT_SAVE_ARGS、DEFAULT_FS_ARGS中可在kedro-datasets文档中查询。凭据credentials安全地访问远程存储Kedro 在实例化DataCatalog前会先从项目配置中读取credentials.yml通常位于conf/local/中的凭据将结果字典通过credentials参数传入DataCatalog.from_config()。catalog.yml条目通过顶层的credentials:键按名称引用凭据块。假设conf/local/credentials.yml中包含dev_s3: client_kwargs: aws_access_key_id: key aws_secret_access_key: secretcatalog 条目即可引用motorbikes: type: pandas.CSVDataset filepath: s3://your_bucket/data/02_intermediate/company/motorbikes.csv credentials: dev_s3 load_args: sep: ,Catalog 会在凭据字典中查找dev_s3将其值作为credentials参数传入数据集构造函数。数据集版本化versioning在 catalog 条目中添加versioned: True即可启用版本化cars: type: pandas.CSVDataset filepath: data/01_raw/company/cars.csv versioned: True启用后filepath成为存储各版本的目录基础。每次流水线运行产生新版本时会存储在filepath/version/filename中其中version是格式为YYYY-MM-DDThh.mm.ss.sssZ的时间戳。默认情况下kedro run加载最新版本要加载指定版本使用--load-versions参数以数据集名:版本时间戳的格式传入kedro run --load-versionscars:YYYY-MM-DDThh.mm.ss.sssZ版本化数据集提供list_versions()方法列出所有可用版本# In a Kedro session or notebook dataset catalog._datasets[cars] versions dataset.list_versions(full_pathFalse) print(versions) # [2024-01-15T10.30.00.000Z, 2024-01-14T09.15.00.000Z, ...]full_pathTrue默认返回各版本的完整文件路径full_pathFalse返回版本字符串时间戳版本按时间倒序返回最新的在前。版本化支持的前提数据集必须继承kedro.io.AbstractVersionedDataset类以接受构造参数version并在_save/_load方法中通过_get_save_path和_get_load_path使用版本化路径。要验证数据集是否支持版本化可检查其类继承关系例如CSVDataset(AbstractVersionedDataset[pd.DataFrame, pd.DataFrame])即支持版本化。HTTP(S) 是数据集实现支持的文件系统但不能与版本化组合使用。多环境配置conf/base与conf/localKedro 通过配置加载器扫描conf文件夹下的配置文件先扫描conf/base再扫描conf/local指定的覆盖环境合并后返回配置字典。因此可以通过在不同环境放置不同版本的catalog.yml来区分开发、测试与生产配置。例如conf/base/catalog.yml中定义生产环境的 S3 位置cars: filepath: s3://my_bucket/cars.csv type: pandas.CSVDataset在conf/local/catalog.yml中覆盖为本地文件cars: filepath: data/01_raw/cars.csv type: pandas.CSVDataset当流水线代码引用cars数据集时本地运行使用conf/local的条目而conf/local无覆盖时使用生产条目。完整的合并规则见 配置基础文档。数据集工厂dataset factories数据集工厂Kedro0.18.12引入让你用模式pattern批量注册配置相似的数据集大幅减少 catalog 条目。例如以下两个条目factory_data: type: pandas.CSVDataset filepath: data/01_raw/factory_data.csv process_data: type: pandas.CSVDataset filepath: data/01_raw/process_data.csv可以重写为一个工厂模式{name}_data: type: pandas.CSVDataset filepath: data/01_raw/{name}_data.csv运行时模式会与流水线节点inputs/outputs中定义的数据集名称进行匹配。工厂模式的行为类似正则表达式可理解为反向的f-string输入数据集factory_data匹配模式{name}_data时name被解析为factory输出数据集process_data则解析为process。注意工厂模式必须用引号包裹否则会触发 YAML 解析错误。三类模式数据集模式dataset patterns在catalog.yml中用{name}_data这类占位符显式定义任何符合命名规则的数据集都会被动态解析用户 catch-all 模式当没有数据集模式匹配时作为兜底使用{default_dataset}占位符每个 catalog 只允许一个指定多个会抛出DatasetError默认运行时模式default runtime patternsKedro 内置的模式当数据集未在 catalog 中定义时通常是流水线运行中产生的中间数据集自动使用例如DataCatalog的{{default}: {type: kedro.io.MemoryDataset}}和SharedMemoryDataCatalog的{{default}: {type: kedro.io.SharedMemoryDataset}}。模式解析顺序数据集模式最明确优先匹配用户 catch-all 模式无数据集模式匹配时的回退默认运行时模式上述均不匹配时由 Kedro 在运行时自动创建MemoryDataset或SharedMemoryDataset。默认情况下catalog.get()不启用运行时模式除非显式设置fallback_to_runtime_patternTruekedro run执行时则自动启用。Pipeline 感知的 catalog 命令DataCatalog通过CatalogCommandsMixin暴露一组检查模式解析结果的命令Kedro 在初始化 session 时自动装配kedro catalog describe-datasets描述流水线中用到的数据集按解析方式显式条目、工厂匹配、默认兜底分组kedro catalog list-patterns按优先级列出 catalog 中所有工厂模式kedro catalog resolve-patterns将流水线数据集对全部模式进行解析返回完整配置。懒加载Lazy loading从 Kedro0.19.10起DataCatalog引入_LazyDataset辅助类以优化性能。它先存储数据集的配置与版本化信息而不立即实例化数据集对象将真正的创建materialisation推迟到数据集被首次访问时。从catalog.yml实例化DataCatalog时Kedro 不会一次性创建所有底层数据集对象而是把每个数据集包装成_LazyDataset注册进 catalog首次访问时直接访问或流水线执行期间再自动实例化In [1]: catalog Out[1]: { shuttles: kedro_datasets.pandas.excel_dataset.ExcelDataset } # 此时 shuttles 尚未完全实例化——只注册了其配置 In [2]: catalog[shuttles] Out[2]: kedro_datasets.pandas.excel_dataset.ExcelDataset( filepathPurePosixPath(/Projects/default/data/01_raw/shuttles.xlsx), protocolfile, load_args{engine: openpyxl}, save_args{index: False}, writer_args{engine: openpyxl} ) # 访问数据集即触发实例化 In [3]: catalog Out[3]: { shuttles: kedro_datasets.pandas.excel_dataset.ExcelDataset( filepathPurePosixPath(/Projects/default/data/01_raw/shuttles.xlsx), ... ) }该机制对大型 catalog 的启动开销有明显改善。在流水线预热阶段可以提前强制实例化所有数据集从而尽早发现配置或导入错误、校验外部依赖、确保执行前所有数据集可创建。虽然_LazyDataset不对最终用户暴露但理解它有助于调试 catalog 行为与数据集实例化问题。在代码中以编程方式使用 DataCatalog除 YAML 配置外也可以通过kedro.io.DataCatalog以代码方式定义数据源适合在catalog.py或 notebook 中构建 IO 层from kedro.io import DataCatalog from kedro_datasets.pandas import ( CSVDataset, SQLTableDataset, SQLQueryDataset, ParquetDataset, ) catalog DataCatalog( { bikes: CSVDataset(filepath../data/01_raw/bikes.csv), cars: CSVDataset(filepath../data/01_raw/cars.csv, load_argsdict(sep,)), cars_table: SQLTableDataset( table_namecars, credentialsdict(consqlite:///kedro.db) ), scooters_query: SQLQueryDataset( sqlselect * from cars where gear4, credentialsdict(consqlite:///kedro.db), ), ranked: ParquetDataset(filepathranked.parquet), } )DataCatalog提供完整的映射式 APIkeys()、values()、items()、__iter__、__getitem__、__contains__、__len__以及load(name, versionNone)、save(name, data)、exists(name)、release(name)、confirm(name)等方法详见 如何在代码中访问 Data Catalog。实用 YAML 配方精选Data Catalog YAML 示例页 收录了大量可直接复用的catalog.yml配方以下为几个典型场景读取带压缩的 CSVboats: type: pandas.CSVDataset filepath: data/01_raw/company/boats.csv.gz load_args: sep: , compression: gzip fs_args: open_args_load: mode: rb从 S3 加载 CSV带凭据与加载参数motorbikes: type: pandas.CSVDataset filepath: s3://your_bucket/data/02_intermediate/company/motorbikes.csv credentials: dev_s3 load_args: sep: , skiprows: 5 skipfooter: 1 na_values: [#NA, NA]从 GCS 加载 Excel带文件系统参数rockets: type: pandas.ExcelDataset filepath: gcs://your_bucket/data/02_intermediate/company/motorbikes.xlsx fs_args: project: my-project credentials: my_gcp_credentials save_args: sheet_name: Sheet1SQL 表与查询凭据块须包含 SQLAlchemy 兼容的连接串conscooters: type: pandas.SQLTableDataset credentials: scooters_credentials table_name: scooters load_args: index_col: [name] columns: [name, gear] save_args: if_exists: replace scooters_query: type: pandas.SQLQueryDataset credentials: scooters_credentials sql: select * from cars where gear4用 YAML 锚点复用公共配置csv命名模板块、: *csv插入模板内容模板条目名必须以_开头Kedro 才不会将其实例化为数据集_csv: csv type: spark.SparkDataset file_format: csv load_args: sep: , na_values: [#NA, NA] header: True inferSchema: False cars: : *csv filepath: s3a://data/01_raw/cars.csv bikes: : *csv filepath: s3a://data/01_raw/bikes.csv load_args: header: False # 局部声明的键会覆盖插入的键源码印证DataCatalog 如何从配置构建DataCatalog的构建入口是类方法 from_config()它接收配置字典、凭据字典、加载版本与保存版本等参数。从源码结构看其完整流程为实例化时通过CatalogConfigResolver提取并排序 catalog 中的工厂模式见 catalog_config_resolver.py 中的_extract_patterns、_sort_patterns、match_dataset_pattern同时从credentials.yml读取凭据并注入条目每个配置条目经parse_dataset_definition位于 core.py解析出数据集类与构造参数数据集以_LazyDataset占位data_catalog.py 第 42 行注册首次访问时才调用materialize()完成实例化版本化相关的路径解析、时间戳生成generate_timestamp与版本列表查询list_versions由core.py中的版本辅助逻辑实现。这种配置解析 → 凭据注入 → 模式匹配 → 懒加载实例化的分层设计正是 Kedro 能把 YAML 配置、工厂模式和运行时流水线无缝衔接起来的原因。进阶主题与相关文档分区与增量数据集处理跨多文件的数据见 分区与增量数据集概念 与 使用方法数据集校验加载/保存时按 schema 校验见 dataset_validation.md自定义数据集创建自己的数据集实现见 自定义数据集教程凭据与参数见 parameters_and_credentials.md。【免费下载链接】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），仅供参考