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

# 错误的代码示例
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'

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

# 错误的代码示例
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'

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(...)

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

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("请上传所有必要的数据文件。")

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

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