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

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)

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

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前端。
# 这里仅为示例，表明节点内部的逻辑。

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",
            ),
        ]
    )

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