本文深入探讨了如何在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 stimport pandas as pdfrom kedro.io import DataCatalog, MemoryDataSetfrom kedro.framework.session import KedroSessionimport 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, Noneif 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 pddef 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, nodefrom .nodes import merge_and_process_datadef 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, Anyfrom kedro.pipeline import Pipelinefrom kedro.framework.project import find_pipelinesdef 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
以上就是Kedro与Streamlit集成：动态数据目录在Web应用中的高效实践的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1379770.html
微信扫一扫 
支付宝扫一扫 
