本教程详细阐述了如何在Streamlit应用中有效集成Kedro数据管道，实现动态数据加载与处理。核心在于通过KedroSession.run()方法的data_catalog参数传递自定义的DataCatalog，以管理Streamlit中加载的DataFrame数据。文章还深入分析了常见的集成误区，如直接修改KedroContext属性，并提供了正确的代码示例和最佳实践，确保数据流的顺畅与高效。
引言：Kedro与Streamlit的强大结合
在现代数据应用开发中，数据管道的自动化与交互式界面的结合变得日益重要。Kedro作为一个生产级的数据管道框架，能够帮助我们构建可维护、可测试和可重用的数据处理逻辑。而Streamlit则以其简洁的API，使得Python开发者能够快速构建美观的数据应用。将Kedro管道集成到Streamlit应用中，可以实现用户通过Web界面上传数据，并实时触发复杂的Kedro数据处理流程，从而构建功能强大且用户友好的数据产品。
本教程的目标是指导您如何在Streamlit应用中运行特定的Kedro管道，并向该管道传递在Streamlit中动态加载的数据，通过自定义的DataCatalog进行管理。
理解Kedro的数据流管理核心：DataCatalog与KedroSession
在深入集成之前，理解Kedro的两个核心概念至关重要：
DataCatalog (数据目录): Kedro的DataCatalog是管理所有数据源的中心枢纽。它定义了数据集的名称、类型以及加载/保存数据的方式。在动态数据场景中，MemoryDataSet特别有用，它允许我们将Python对象（如Pandas DataFrame）作为数据集在内存中传递，而无需写入磁盘。KedroSession (会话): KedroSession是Kedro项目的入口点，负责加载项目上下文、配置以及运行管道。它是执行Kedro操作的主要接口。KedroSession.run()方法是启动管道执行的关键。
正确姿势：通过KedroSession.run()传递自定义DataCatalog
在Streamlit中运行Kedro管道并传递动态数据，最核心且正确的方法是利用KedroSession.run()方法的data_catalog参数。这个参数允许您在运行时提供一个临时的、自定义的DataCatalog，它将覆盖或扩展Kedro项目默认的catalog.yml中定义的同名数据集。
实现步骤：
在Streamlit中加载数据： 使用Streamlit的文件上传器或其他输入组件获取用户数据，并将其转换为Pandas DataFrame。创建MemoryDataSet： 将这些DataFrame封装成MemoryDataSet实例。MemoryDataSet允许Kedro管道在内存中处理这些数据。构建自定义DataCatalog： 创建一个新的DataCatalog实例，并将您的MemoryDataSets添加到其中，使用它们在Kedro管道中对应的输入数据集名称作为键。创建KedroSession并运行管道： 使用KedroSession.create()初始化一个会话，然后调用session.run()，并将自定义的DataCatalog通过data_catalog参数传递进去，同时指定要运行的pipeline_name。
示例代码：
以下是一个在Streamlit中集成Kedro管道的完整示例，展示了如何动态加载数据并传递给Kedro：
import streamlit as stimport pandas as pdfrom kedro.framework.session import KedroSessionfrom kedro.io import DataCatalog, MemoryDataSetimport os# 假设您的Kedro项目位于当前工作目录下的 'my_kedro_project'# 请根据实际情况调整 project_pathproject_path = os.path.join(os.getcwd(), 'my_kedro_project')st.title("Kedro与Streamlit数据处理应用")st.header("上传您的数据")# 模拟Streamlit文件上传和DataFrame创建# 在实际应用中，这里会是 st.file_uploader 和 pd.read_csv/excel 等uploaded_file1 = st.file_uploader("上传 reagentes_raw.csv", type=['csv'])uploaded_file2 = st.file_uploader("上传 balanco_de_massas_raw.csv", type=['csv'])# ... 更多文件上传器df1, df2, df3, df4, df5, df6 = None, None, None, None, None, Noneif uploaded_file1: df1 = pd.read_csv(uploaded_file1) st.write("reagentes_raw 数据预览:") st.dataframe(df1.head())if uploaded_file2: df2 = pd.read_csv(uploaded_file2) st.write("balanco_de_massas_raw 数据预览:") st.dataframe(df2.head())# ... 处理其他上传文件# 确保所有必需的DataFrame都已加载 (这里仅为演示，实际应根据管道输入进行检查)if st.button('处理输入数据') and df1 is not None and df2 is not None: # 简化检查 st.info('正在执行Kedro管道...') # 模拟其他DataFrame，实际应通过上传获取 if df3 is None: df3 = pd.DataFrame({'colA': [1,2], 'colB': ['x','y']}) if df4 is None: df4 = pd.DataFrame({'colC': [3,4], 'colD': ['a','b']}) if df5 is None: df5 = pd.DataFrame({'colE': [5,6], 'colF': ['m','n']}) if df6 is None: df6 = pd.DataFrame({'colG': [7,8], 'colH': ['p','q']}) try: # 创建自定义DataCatalog，包含MemoryDataSet 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) # 示例数据 }) # 创建KedroSession并运行管道 with KedroSession.create(project_path=project_path) as session: # 关键：通过 data_catalog 参数传递自定义目录 session.run(data_catalog=custom_catalog, pipeline_name="tag_web_app") st.success('数据处理成功！') # 从自定义的catalog中加载管道的输出数据 # 假设管道的输出数据集名为 "merged_raw_data_process" merged_data = custom_catalog.load("merged_raw_data_process") st.header('处理结果预览') st.dataframe(merged_data.head()) # 示例：显示最后更新时间，假设输出数据包含 'timestamp' 列 if 'timestamp' in merged_data.columns: last_update = pd.to_datetime(merged_data['timestamp']).max() st.write(f"数据集中最新信息的时间: {last_update.strftime('%Y/%m/%d %H:%M:%S')}") else: st.write("输出数据中未找到 'timestamp' 列。") except Exception as e: st.error(f"Kedro管道执行失败: {e}") st.exception(e)
注意：
请将project_path替换为您的Kedro项目的实际路径。my_kedro_project目录下需要有一个名为tag_web_app的Kedro管道，并且该管道的输入数据集名称（例如reagentes_raw）需要与custom_catalog中定义的键名一致。管道的输出数据集（例如merged_raw_data_process）也需要在custom_catalog中定义或由管道生成，以便后续加载。
结果的获取与处理
当Kedro管道通过session.run(data_catalog=custom_catalog, …)执行完毕后，管道的输出数据集（如果它们被定义为写入catalog）将存储在您传入的custom_catalog对象中。这意味着，您可以直接从该custom_catalog实例中加载管道处理后的结果，并在Streamlit应用中进行展示或进一步处理。
如上例所示：
merged_data = custom_catalog.load("merged_raw_data_process")st.dataframe(merged_data.head())
这行代码从之前传入的custom_catalog中加载了名为merged_raw_data_process的数据集，该数据集是Kedro管道的最终输出。
常见错误与解决方案
在集成Kedro与Streamlit时，开发者可能会遇到一些常见的AttributeError。这些错误通常源于尝试以不正确的方式修改Kedro的内部状态。
错误1: AttributeError: can’t set attribute ‘catalog’
问题描述: 尝试直接对KedroSession或KedroContext的catalog属性进行赋值操作，例如 context.catalog = custom_catalog。
错误原因: KedroSession.catalog和KedroContext.catalog属性在Kedro的设计中是只读的。它们在会话或上下文创建时被初始化，并且不应该在运行时被直接外部修改。Kedro通过配置（catalog.yml）和session.run()方法的参数来管理数据目录的生命周期和内容。
解决方案: 绝对不要尝试直接设置context.catalog。正确的做法是，在调用session.run()时，通过data_catalog参数传递您自定义的DataCatalog。如前文示例所示：
with KedroSession.create(project_path=project_path) as session: session.run(data_catalog=custom_catalog, pipeline_name="tag_web_app")
这种方式是Kedro官方推荐且唯一支持的在运行时注入自定义数据目录的方法。
错误2: AttributeError: ‘KedroContext’ object has no attribute ‘pipeline_registry’
问题描述: 尝试从KedroContext对象中访问一个名为pipeline_registry的属性，例如 context.pipeline_registry.get(“tag_web_app”)。
错误原因: KedroContext对象本身不直接暴露pipeline_registry属性。管道的注册和管理是Kedro内部框架的一部分，通常通过KedroSession的run()方法或context.pipelines属性来间接访问和执行。尝试直接访问pipeline_registry是错误的API使用方式。
解决方案: 避免直接操作pipeline_registry。如果您需要运行特定的管道，只需在session.run()方法中通过pipeline_name参数指定即可：
with KedroSession.create(project_path=project_path) as session: session.run(pipeline_name="tag_web_app", data_catalog=custom_catalog)
如果您确实需要获取管道对象（例如用于更高级的调试或自定义运行），可以通过context.pipelines字典来访问，例如 context.pipelines[“tag_web_app”]，但通常情况下，直接使用session.run()更为简洁和推荐。
最佳实践与注意事项
Kedro项目结构清晰： 确保您的Kedro项目结构良好，管道定义清晰，输入输出数据集命名规范，这有助于Streamlit应用与Kedro管道的顺利对接。数据隔离： 使用MemoryDataSet确保每次Streamlit触发的Kedro管道运行时，数据都是独立的，不会相互干扰。错误处理： 在Streamlit应用中加入健壮的try-except块，捕获Kedro管道执行过程中可能出现的错误，并向用户提供友好的反馈。异步处理（高级）： 对于长时间运行的Kedro管道，考虑在Streamlit中使用异步任务队列（如Celery）来避免UI阻塞，提升用户体验。环境管理： 确保Streamlit应用运行的环境与Kedro项目所需的依赖一致，避免版本冲突。安全性： 如果涉及到敏感数据，请确保数据上传、处理和存储过程符合安全规范。
总结
将Kedro的强大数据管道能力与Streamlit的便捷交互界面相结合，能够为数据科学家和工程师提供一个高效且灵活的开发环境。本教程强调了在Streamlit应用中通过KedroSession.run()方法的data_catalog参数传递自定义DataCatalog的正确方法，这是处理动态数据的核心。同时，通过深入解析常见的AttributeError，我们明确了Kedro的API设计原则，即避免直接修改只读属性或访问不存在的内部组件。遵循这些指导原则和最佳实践，您将能够构建稳定、高效且易于维护的Kedro-Streamlit集成应用。
以上就是Kedro与Streamlit集成：构建动态数据管道的实践指南的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1379594.html
微信扫一扫 
支付宝扫一扫 
