本文深入探讨了如何在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代码中的数据集名称：
项目结构:
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
