Kedro与Streamlit集成：动态数据目录在Web应用中的高效实践

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

Kedro与Streamlit集成：动态数据目录在Web应用中的高效实践

2025-11-13

浏览次数：次

返回列表

Kedro与Streamlit集成：动态数据目录在Web应用中的高效实践

本文深入探讨了如何在streamlit应用中无缝集成kedro数据管道，并动态传递自定义datacatalog。我们将分析常见的集成误区，特别是关于kedrosession和kedrocontext中datacatalog和pipeline_registry属性的错误使用，并提供一个清晰、可操作的解决方案，以实现streamlit加载数据后，通过内存数据集高效运行kedro管道，从而构建灵活的数据处理web应用。

Kedro与Streamlit集成的核心挑战

在构建数据驱动的Web应用时，将Streamlit的交互式前端与Kedro的强大数据管道后端结合是一种常见且高效的模式。然而，当需要在Streamlit中动态加载数据（例如通过文件上传），并将其作为输入传递给Kedro管道进行处理时，会遇到一些挑战。核心问题在于如何将Streamlit中已加载的Pandas DataFrame等内存数据，封装成Kedro可识别的DataCatalog，并在不修改Kedro项目配置的情况下，将其注入到管道执行流程中。

常见的需求场景包括：

用户通过Streamlit上传CSV或Excel文件。
Streamlit将这些文件读取为Pandas DataFrame。
需要将这些DataFrame作为输入，运行一个或多个Kedro管道。
Kedro管道处理完成后，将结果返回给Streamlit进行展示或进一步操作。

常见错误与原因分析

在尝试将自定义DataCatalog传递给Kedro管道时，开发者可能会遇到以下几种AttributeError或TypeError：

AttributeError: can't set attribute 'catalog' 这个错误通常发生在尝试直接修改KedroSession或KedroContext的catalog属性时，例如 context.catalog = custom_catalog。 原因分析：在较新版本的Kedro中，KedroSession.catalog以及从session.load_context()获取到的context.catalog通常是只读的，或者不应通过直接赋值的方式来动态替换。Kedro设计其内部组件时，强调配置的不可变性，以确保管道执行的一致性和可预测性。
AttributeError: 'KedroContext' object has no attribute 'pipeline_registry' 这个错误表明尝试访问KedroContext对象上不存在的pipeline_registry属性。 原因分析：KedroContext对象从未直接拥有名为pipeline_registry的属性。管道的注册和获取通常通过KedroSession或项目内部的机制来管理，而不是直接暴露在KedroContext上。
TypeError: KedroSession.run() got an unexpected keyword argument 'extra_params' 这个错误通常与session.run()方法的参数变化有关。 原因分析：KedroSession.run()方法的签名可能会在不同Kedro版本中发生变化。extra_params参数在某些旧版本中可能用于传递额外的运行时参数，但在新版本中可能已被移除或替换为其他参数（如config_params）。因此，使用过时的参数会导致TypeError。

正确集成方案：动态DataCatalog的传递

解决上述问题的关键在于理解KedroSession.run()方法的设计意图，并利用其提供的data_catalog参数来注入自定义的DataCatalog。

核心思想：在Streamlit中加载数据后，将这些数据封装成Kedro的MemoryDataSet，然后组合成一个临时的DataCatalog实例。最后，在调用KedroSession.run()时，通过data_catalog参数将这个自定义的DataCatalog传递进去。这样，Kedro管道在执行时会优先使用这个自定义的DataCatalog来查找输入数据，而不是默认的conf/base/catalog.yml中定义的数据集。

步骤一：在Streamlit中加载数据并创建MemoryDataSet

首先，在Streamlit应用中实现文件上传和数据加载逻辑，然后将加载的Pandas DataFrame封装成MemoryDataSet。MemoryDataSet是Kedro提供的一种特殊数据集，用于处理内存中的数据，而无需将其写入磁盘。

import streamlit as st
import pandas as pd
from kedro.io import DataCatalog, MemoryDataSet
from kedro.framework.session import KedroSession
import os

# 假设Kedro项目根目录的路径
# 请根据实际情况修改此路径，确保Streamlit应用可以访问到Kedro项目
KEDRO_PROJECT_PATH = os.path.abspath("./my_kedro_project") 

st.set_page_config(layout="wide")
st.title("Kedro管道与动态数据集成示例")

st.markdown("""
本应用演示了如何通过Streamlit上传数据，并将其作为自定义DataCatalog传递给Kedro管道进行处理。
""")

# Streamlit文件上传器
st.header("1. 上传输入数据")
uploaded_file_1 = st.file_uploader("上传第一个CSV文件 (例如: reagentes_raw.csv)", type=["csv"])
uploaded_file_2 = st.file_uploader("上传第二个CSV文件 (例如: balanco_de_massas_raw.csv)", type=["csv"])

df1, df2 = None, None

if uploaded_file_1:
    df1 = pd.read_csv(uploaded_file_1)
    st.success("文件 'reagentes_raw' 加载成功！")
    st.subheader("reagentes_raw 数据预览:")
    st.dataframe(df1.head())

if uploaded_file_2:
    df2 = pd.read_csv(uploaded_file_2)
    st.success("文件 'balanco_de_massas_raw' 加载成功！")
    st.subheader("balanco_de_massas_raw 数据预览:")
    st.dataframe(df2.head())

# 运行Kedro管道的按钮
st.header("2. 运行Kedro管道")
if st.button('处理数据') and df1 is not None and df2 is not None:
    if not os.path.exists(KEDRO_PROJECT_PATH):
        st.error(f"错误：Kedro项目路径不存在或不正确。请检查路径: {KEDRO_PROJECT_PATH}")
        st.stop()

    with st.spinner('正在执行Kedro管道...'):
        try:
            # 1. 创建自定义DataCatalog，使用MemoryDataSet封装DataFrame
            # 这里的键名 (例如 "reagentes_raw", "balanco_de_massas_raw") 
            # 必须与你的Kedro管道中定义的输入数据集名称一致。
            custom_catalog = DataCatalog({
                "reagentes_raw": MemoryDataSet(df1),
                "balanco_de_massas_raw": MemoryDataSet(df2),
                # 如果有更多数据集，按此模式添加
            })

            # 2. 初始化KedroSession并运行指定的管道
            # 确保 'my_kedro_pipeline' 是你Kedro项目中实际定义的管道名称
            with KedroSession.create(project_path=KEDRO_PROJECT_PATH) as session:
                # 通过 data_catalog 参数传入自定义的 DataCatalog
                session.run(pipeline_name="my_kedro_pipeline", data_catalog=custom_catalog)

            st.success('Kedro管道执行成功！')

            # 3. 从自定义catalog中加载管道输出结果
            # 假设管道输出一个名为 "processed_output_data" 的数据集
            # 这个数据集也必须被定义为MemoryDataSet在custom_catalog中
            if "processed_output_data" in custom_catalog.list():
                processed_data = custom_catalog.load("processed_output_data")
                st.header('3. 管道处理结果:')
                st.dataframe(processed_data.head())
                st.download_button(
                    label="下载处理后的数据 (CSV)",
                    data=processed_data.to_csv(index=False).encode('utf-8'),
                    file_name="processed_output.csv",
                    mime="text/csv",
                )
            else:
                st.warning("Kedro管道未将 'processed_output_data' 存储回自定义DataCatalog。请检查管道配置。")

        except Exception as e:
            st.error(f"运行Kedro管道时发生错误: {e}")
            st.exception(e)

步骤二：Kedro项目配置（示例）

为了使上述Streamlit应用能够成功运行，你需要有一个相应的Kedro项目。以下是一个简化的Kedro项目结构和管道示例，以匹配Streamlit代码中的数据集名称：

网页制作与PHP语言应用

图书《网页制作与PHP语言应用》，由武汉大学出版社于2006出版，该书为普通高等院校网络传播系列教材之一，主要阐述了网页制作的基础知识与实践，以及PHP语言在网络传播中的应用。该书内容涉及：HTML基础知识、PHP的基本语法、PHP程序中的常用函数、数据库软件MySQL的基本操作、网页加密和身份验证、动态生成图像、MySQL与多媒体素材库的建设等。

460 查看详情网页制作与PHP语言应用

项目结构:

my_kedro_project/
├── conf/
│   └── base/
│       └── catalog.yml  # 可以为空或定义其他持久化数据集
│       └── parameters.yml
├── src/
│   └── my_kedro_project/
│       ├── __init__.py
│       ├── pipeline_registry.py
│       └── pipelines/
│           └── my_kedro_pipeline/
│               ├── __init__.py
│               ├── nodes.py
│               └── pipeline.py
└── pyproject.toml

src/my_kedro_project/pipelines/my_kedro_pipeline/nodes.py 示例：

import pandas as pd

def merge_and_process_data(df_reagentes: pd.DataFrame, df_balanco: pd.DataFrame) -> pd.DataFrame:
    """
    一个简单的节点函数，用于合并并处理两个输入DataFrame。
    """
    st.write("Kedro节点：正在合并数据...")
    # 假设这两个DataFrame有一个共同的键 'id' 用于合并
    # 实际项目中，你需要根据数据结构调整合并逻辑
    merged_df = pd.merge(df_reagentes, df_balanco, on='id', how='inner', suffixes=('_reag', '_bal'))

    # 进行一些简单的处理
    merged_df['calculated_value'] = merged_df['value_reag'] * merged_df['value_bal']

    return merged_df

# 注意：为了让Streamlit的st.write在Kedro节点中可见，你可能需要一些高级的日志捕获或回调机制。
# 在标准的Kedro执行中，st.write不会直接输出到Streamlit前端。
# 这里仅为示例，表明节点内部的逻辑。

src/my_kedro_project/pipelines/my_kedro_pipeline/pipeline.py 示例：

from kedro.pipeline import Pipeline, node
from .nodes import merge_and_process_data

def create_pipeline(**kwargs) -> Pipeline:
    """
    创建并注册 'my_kedro_pipeline'。
    输入数据集名称 ('reagentes_raw', 'balanco_de_massas_raw') 
    必须与Streamlit中自定义DataCatalog的键名一致。
    输出数据集名称 ('processed_output_data') 
    也应在自定义DataCatalog中被预期。
    """
    return Pipeline(
        [
            node(
                func=merge_and_process_data,
                inputs=["reagentes_raw", "balanco_de_massas_raw"],
                outputs="processed_output_data",
                name="merge_and_process_node",
            ),
        ]
    )

src/my_kedro_project/pipeline_registry.py 示例：

from typing import Dict, Any
from kedro.pipeline import Pipeline
from kedro.framework.project import find_pipelines

def register_pipelines() -> Dict[str, Pipeline]:
    """
    注册项目的管道。
    """
    pipelines = find_pipelines()
    # 注册你的管道，并将其设置为默认管道
    pipelines["__default__"] = pipelines["my_kedro_pipeline"]
    return pipelines

注意事项与最佳实践

Kedro版本兼容性：KedroSession.run()方法的参数可能随Kedro版本更新而变化。始终查阅你所使用的Kedro版本的官方文档，以确保参数的正确性。
数据持久化：
- 输入数据：如果Streamlit上传的数据量较大，或者需要在多次运行之间保持不变，可以考虑先将上传数据保存到临时文件（如Parquet），然后在Kedro管道的catalog.yml中定义相应的ParquetDataSet。但对于动态、一次性的内存数据，MemoryDataSet是更简洁的选择。
- 输出数据：如果管道的输出需要持久化，可以在Kedro管道内部定义一个ParquetDataSet或CSVDataSet来将结果写入磁盘。或者，如示例所示，从自定义DataCatalog中加载结果MemoryDataSet后，在Streamlit中手动将其保存或提供下载。
错误处理与日志：在Streamlit应用中，务必使用try-except块捕获KedroSession.run()可能抛出的异常，并使用st.error()和st.exception()向用户提供友好的错误提示和详细的堆栈信息。
Kedro项目路径：确保Streamlit应用能够正确找到KEDRO_PROJECT_PATH。在部署时，这可能意味着需要将Kedro项目作为子目录包含在Streamlit应用中，或者通过环境变量动态配置路径。
管道设计：设计Kedro管道时，输入数据集的名称（例如reagentes_raw）应与你在Streamlit中创建DataCatalog时使用的键名保持一致。这样，`session.run(data