Kedro与Streamlit集成：动态数据目录下的管道运行实践

---

本文旨在指导开发者如何在streamlit应用中集成并运行kedro数据管道，重点解决如何动态创建并传递自定义`datacatalog`以处理streamlit加载的数据。文章将阐明常见的错误尝试及其原因，并提供一种健壮的方法，通过`kedrosession.run()`的`data_catalog`参数正确地将运行时数据注入kedro管道，从而实现数据处理的无缝衔接。

在构建交互式数据应用时，将强大的数据管道框架（如Kedro）与灵活的Web应用框架（如Streamlit）结合是一种常见的需求。特别是当数据源是动态的，例如用户通过Streamlit界面上传文件时，我们需要一种机制来将这些运行时加载的数据作为输入，传递给Kedro管道进行处理。本文将详细介绍如何实现这一目标，并纠正在此过程中可能遇到的常见误区。

理解Kedro与Streamlit集成中的数据流

Kedro的核心概念之一是DataCatalog，它定义了数据加载和保存的方式。通常，DataCatalog在conf/base/catalog.yml中静态定义。然而，在Streamlit应用中，数据通常在运行时由用户上传，这意味着我们需要一个动态的DataCatalog来封装这些内存中的DataFrame。

目标是将Streamlit中加载的DataFrame包装成MemoryDataSet，然后构建一个临时的DataCatalog，并将其传递给Kedro管道执行。

常见错误尝试与原因分析

在尝试将自定义DataCatalog传递给Kedro管道时，开发者可能会遇到一些AttributeError。理解这些错误的原因对于正确实现集成至关重要。

错误尝试一：直接修改KedroContext或KedroSession的catalog属性

# 错误的代码示例
from kedro.framework.session import KedroSession
from kedro.io import DataCatalog, MemoryDataSet
import pandas as pd
import streamlit as st

# 假设 df1, df2, ... 是在Streamlit中加载的DataFrame
df1 = pd.DataFrame({'col1': [1, 2]})
df2 = pd.DataFrame({'col2': [3, 4]})

if st.button('Processar Dados de Entrada'):
    with KedroSession.create(project_path="./my_kedro_project") as session:
        context = session.load_context()
        # 尝试直接设置context.catalog，这将导致AttributeError
        # context.catalog = DataCatalog({"my_data": MemoryDataSet(df1)}) # AttributeError: can't set attribute 'catalog'

        # 尝试直接设置session.catalog，同样会导致AttributeError
        # session.catalog = DataCatalog({"my_data": MemoryDataSet(df1)}) # AttributeError: can't set attribute 'catalog'

        # ...后续管道运行代码

原因分析：AttributeError: can't set attribute 'catalog'

KedroSession和KedroContext的catalog属性在Kedro内部被设计为只读。这意味着您不能在会话或上下文创建之后，通过直接赋值的方式来修改它们引用的DataCatalog对象。DataCatalog在KedroSession初始化时被加载并冻结，以确保管道执行的一致性和可预测性。尝试直接修改它会违反这一设计原则，从而引发AttributeError。

错误尝试二：通过KedroContext访问pipeline_registry

# 错误的代码示例
from kedro.framework.session import KedroSession
from kedro.runner import SequentialRunner
import streamlit as st

# ... (数据加载和catalog创建) ...

if st.button('Processar Dados de Entrada'):
    with KedroSession.create(project_path="./my_kedro_project") as session:
        context = session.load_context()
        runner = SequentialRunner()
        # 尝试通过context.pipeline_registry获取管道，这将导致AttributeError
        # runner.run(pipeline=context.pipeline_registry.get("tag_web_app"), catalog=custom_catalog) # AttributeError: 'KedroContext' object has no attribute 'pipeline_registry'

原因分析：AttributeError: 'KedroContext' object has no attribute 'pipeline_registry'

KedroContext对象并不直接拥有pipeline_registry属性。管道注册是KedroSession负责管理的一部分。当您通过KedroSession.run()方法执行管道时，会话会自动处理管道的查找和注册。直接从KedroContext中访问pipeline_registry是不符合Kedro设计模式的。

标贝悦读AI配音

在线文字转语音软件-专业的配音网站

66 查看详情

正确的方法：通过KedroSession.run()传递自定义DataCatalog

Kedro提供了一种简洁且推荐的方式来在运行时注入自定义DataCatalog，即使用KedroSession.run()方法的data_catalog（或旧版本中的catalog）参数。

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

首先，在Streamlit应用中，您需要使用文件上传器或其他方式加载数据，并将其转换为Pandas DataFrame。然后，将这些DataFrame包装成Kedro的MemoryDataSet对象。MemoryDataSet是Kedro提供的一种数据集类型，用于处理内存中的数据，非常适合这种动态场景。

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

# 假设你的Kedro项目路径
project_path = Path(__file__).parent / "my_kedro_project" # 根据实际项目结构调整

st.title("Kedro与Streamlit数据处理应用")

uploaded_file1 = st.file_uploader("上传 Reagentes CSV", type=["csv"])
uploaded_file2 = st.file_uploader("上传 Balanço de Massas CSV", type=["csv"])
# ... 可以根据需要添加更多文件上传器

df1, df2, df3, df4, df5, df6 = None, None, None, None, None, None

if uploaded_file1:
    df1 = pd.read_csv(uploaded_file1)
    st.write("Reagentes 数据加载成功:")
    st.dataframe(df1.head())

if uploaded_file2:
    df2 = pd.read_csv(uploaded_file2)
    st.write("Balanço de Massas 数据加载成功:")
    st.dataframe(df2.head())

# 假设还有其他文件加载，这里简化
# df3 = pd.DataFrame(...)
# df4 = pd.DataFrame(...)
# df5 = pd.DataFrame(...)
# df6 = pd.DataFrame(...)

步骤二：构建自定义DataCatalog

当所有必需的DataFrame都加载完毕后，您可以创建一个新的DataCatalog实例，并将这些MemoryDataSet对象作为键值对添加到其中。键名应与您的Kedro管道中期望的数据集名称相匹配。

# ... (承接上一步的代码) ...

if st.button('Processar Dados de Entrada'):
    if df1 is not None and df2 is not None: # 确保所有必要数据都已加载
        # 创建自定义DataCatalog
        custom_catalog = DataCatalog({
            "reagentes_raw": MemoryDataSet(df1),
            "balanco_de_massas_raw": MemoryDataSet(df2),
            # 根据需要添加更多数据集
            # "laboratorio_raw": MemoryDataSet(df3),
            # "laboratorio_raiox_raw": MemoryDataSet(df4),
            # "carta_controle_pims_raw": MemoryDataSet(df5),
            # "blend_raw": MemoryDataSet(df6)
        })

        st.info("正在执行Kedro管道...")

        try:
            # 步骤三：通过KedroSession.run()传递自定义DataCatalog
            with KedroSession.create(project_path=project_path) as session:
                session.run(data_catalog=custom_catalog, pipeline_name="tag_web_app")

            st.success('数据处理成功！')

            # 步骤四：从自定义DataCatalog中加载处理后的结果
            # 假设管道输出一个名为 "merged_raw_data_process" 的数据集
            if "merged_raw_data_process" in custom_catalog.list():
                merged_data = custom_catalog.load("merged_raw_data_process")
                st.header('结果数据预览')
                st.dataframe(merged_data.head())

                # 假设结果数据中有一个时间戳列
                if 'timestamp_column' in merged_data.columns: # 请替换为实际的时间戳列名
                    last_update = merged_data['timestamp_column'].max()
                    st.write(f"最新数据时间: {last_update.strftime('%d/%m/%Y %H:%M')}")
            else:
                st.warning("管道未生成预期的 'merged_raw_data_process' 数据集。")

        except Exception as e:
            st.error(f"Kedro管道执行失败: {e}")
    else:
        st.warning("请上传所有必要的数据文件。")

步骤三：通过KedroSession.run()传递自定义DataCatalog

这是解决问题的关键步骤。KedroSession.run()方法接受一个data_catalog（或旧版本中的catalog）参数，允许您传入一个临时的DataCatalog实例。这个传入的DataCatalog会与项目默认的DataCatalog合并，或者在某些情况下完全覆盖默认的同名数据集定义，从而将您的内存数据注入到管道执行中。

# ... (代码片段已包含在步骤二中) ...
with KedroSession.create(project_path=project_path) as session:
    session.run(data_catalog=custom_catalog, pipeline_name="tag_web_app")

请注意，pipeline_name参数用于指定要运行的特定管道。如果您的Kedro项目只有一个默认管道，可以省略此参数。

步骤四：从自定义DataCatalog中加载处理后的结果

管道执行完成后，如果您的管道配置为将结果保存到MemoryDataSet中（例如，通过在catalog.yml中将输出数据集定义为MemoryDataSet，或者在运行时通过custom_catalog覆盖），您可以直接从传入的custom_catalog中加载这些结果。

# ... (代码片段已包含在步骤二中) ...
merged_data = custom_catalog.load("merged_raw_data_process")

注意事项与最佳实践

1. Kedro项目结构： 确保您的Streamlit应用能够正确找到Kedro项目的根目录（project_path）。通常，Streamlit应用可以放在Kedro项目之外，但需要正确指定project_path。
2. MemoryDataSet的使用： MemoryDataSet非常适合处理临时数据。如果需要将处理结果持久化，您可以在DataCatalog中定义其他类型的数据集（如ParquetDataSet、CSVDataSet等），或者在管道的末尾手动保存DataFrame。
3. 管道定义： 确保您的Kedro管道中的节点能够接收和处理MemoryDataSet提供的数据。输入数据集的名称必须与custom_catalog中定义的键名匹配。
4. 错误处理： 在实际应用中，务必添加适当的错误处理机制，以捕获Kedro管道执行过程中可能出现的异常，并向用户提供友好的反馈。
5. 性能考虑： 对于大型数据集，MemoryDataSet可能会消耗大量内存。根据您的数据规模和性能需求，可能需要考虑更高效的数据处理策略，例如分块处理或使用更适合大数据的Kedro数据集类型。
6. Kedro版本兼容性： 本文中的data_catalog参数适用于较新版本的Kedro。如果您使用的是旧版本，可能需要使用catalog参数。请查阅您的Kedro版本文档以确认正确的参数名称。

总结

通过在KedroSession.run()方法中利用data_catalog参数，我们可以优雅地将Streamlit中加载的动态数据注入到Kedro管道中进行处理。这种方法避免了直接修改Kedro内部只读属性的错误，提供了一种符合Kedro设计哲学且易于维护的集成方案。遵循本文介绍的步骤和最佳实践，您将能够构建出功能强大、交互性强的数据处理应用。

以上就是Kedro与Streamlit集成：动态数据目录下的管道运行实践的详细内容，更多请关注其它相关文章！

# app  # session  # csv  # 大数据  # 鼓楼区网页seo服务费  # 淘宝电商站外营销推广  # 鹿寨高效网站建设有哪些  # dw网站建设模板代码  # 龙华网站建设必备知识  # 焦作短视频seo推广  # 浏阳软文营销推广中心  # 北京seo外包服务  # 端口推广都有什么网站  # seo优化服务费用  # 解决问题  # 旧版本  # 道中  # 键值  # 上传  # 您可以  # 数据处理  # 自定义  # 您的  # 加载  # blend  # 键值对  # stream  # ai 

相关栏目： 【 Google疑问12 】 【 Facebook疑问10 】 【 优化推广96088 】 【 技术知识133117 】 【 IDC资讯59369 】 【 网络运营7196 】 【 IT资讯61894 】

相关推荐： 谷歌浏览器怎么把网页翻译成中文_Chrome网页翻译功能使用方法  创建您的便携版VS Code：让配置随身携带  VS Code源代码管理(SCM)视图的进阶使用技巧  如何在CSS中使用伪类选择器_hover实现悬停效果  Python定时发送QQ消息  解决PHP MySQL数据库更新无响应：SQL查询语法错误解析  QQ邮箱PC端登录页面_QQ邮箱网页版登录界面  WPS长文档分栏排版不乱方法_WPS分栏+分节符报纸排版教程  win11如何诊断DirectX问题 Win11运行dxdiag工具排查显卡故障【排错】  漫蛙app官方版手机正版入口-漫蛙漫画manwa在线漫画正版入口  苹果11如何更换iCloud账号_苹果11账号切换的具体步骤  《知到》打卡课程方法  c++20的指定初始化(Designated Initializers)怎么用_c++ C风格结构体初始化  获取WooCommerce产品在后台编辑页面的分类ID  汽水音乐在线听歌网页版 汽水音乐在线听歌网页版入口  口腔诊所管理软件推荐  如何用mysql开发用户注册登录功能_mysql用户注册登录数据库设计  济南公交卡手机充值指南  海外搜索引擎推广效果怎么样,怎么分析效果！  使用 .htaccess 正确配置 WordPress 子目录重定向与路径保留  键盘保修需要什么_键盘售后维修流程  微信客户端怎么查看二维码_微信客户端个人二维码查看方法  Lar*el 关联查询：同时筛选父表与子表数据的高效策略  qq邮箱格式填写示例 qq邮箱标准填写规范  Linux如何优化系统启动流程_Linux启动项优化方案  小红书网页版首页入口 小红书网页版电脑端官方登录链接  HTML中多图片上传与预览：解决ID冲突的专业指南  如何使用 composer 和 aop-php 实现 AOP 编程？  手机自动关机是怎么回事？如何修复？手机异常关机的原因排查与修复技巧  电脑视频号|直播|如何分享屏幕  利用Flexbox实现图片元素的二维布局：2x2网格排列指南  Win10截图远程协助 Win10远程桌面截屏法【场景应用】  鸿蒙单条备忘录如何加密  使用 J*aScript 随机化 CSS Grid 布局中的元素顺序  惠普电脑BIOS界面看不懂怎么办_HP电脑BIOS功能选项解读与设置  Word 2003字体大小设置方法  TikTok视频播放不流畅怎么办 TikTok视频播放优化方法  CSS动画如何实现图标旋转并放大_transform rotate scale @keyframes实现  ToDesk远程摄像头功能使用方法_ToDesk远程视频画面查看设置教程  苹果手机缓存怎么清除_苹果手机缓存如何清除iphone各版本操作步骤  批改网网页版登录 批改网电脑版学生登录入口  韩小圈网页版PC端入口 韩小圈网页版官方网站入口  谷歌浏览器如何查找和删除恶意软件 谷歌浏览器内置安全清理工具使用教程  Win10通知横幅停留时间修改 Win10自定义通知显示时长【技巧】  邦丰播放器频道搜索设置  在React中正确处理HTML input type="number"的数值类型  《下一站江湖2》风神腿获取攻略  优化Leaflet弹出层图片显示：条件渲染策略  研招网官方网站正版登录网址_中国研究生招生信息网官网首页  TikTok视频播放中断怎么办 TikTok播放异常修复方法 

 2025-11-14