Streamlit深度解析：2025年数据科学家的开发神器

┌────────────────┐    ┌────────────────┐    ┌────────────────┐
│ Python脚本     │ -> │ Streamlit运行时 │ -> │ 交互式Web应用  │
│(数据分析/模型)  │    │  (Python/JS)   │    │  (浏览器)      │
└────────────────┘    └────────────────┘    └────────────────┘
         ↑                     ↑                     ↑
         │                     │                     │
┌────────────────┐    ┌────────────────┐    ┌────────────────┐
│ 数据源         │    │ 组件库         │    │ 用户交互       │
│(文件/数据库/API)│    │ (UI元素)       │    │ (输入/操作)    │
└────────────────┘    └────────────────┘    └────────────────┘

# Streamlit核心模块结构示例
import streamlit as st

# 创建应用
def create_app():
    # 设置页面配置
    st.set_page_config(
        page_title="数据分析应用",
        page_icon="📊",
        layout="wide"
    )
    
    # 添加标题和说明
    st.title("2025年销售数据分析")
    st.markdown("这是一个基于Streamlit的交互式数据分析应用")
    
    # 添加交互组件
    department = st.selectbox("选择部门", ["销售部", "市场部", "研发部"])
    date_range = st.date_input("选择日期范围", [])
    
    # 加载和处理数据
    data = load_data(department, date_range)
    
    # 展示数据和可视化
    st.dataframe(data)
    st.line_chart(data["销售额"])
    
    # 添加侧边栏
    with st.sidebar:
        st.header("配置选项")
        show_details = st.checkbox("显示详细数据")
        if show_details:
            st.write("正在显示详细数据...")

# 应用入口
if __name__ == "__main__":
    create_app()

# 基本文本元素
st.title("这是标题")
st.header("这是一级标题")
st.subheader("这是二级标题")
st.text("这是普通文本")
st.markdown("**这是Markdown文本**")

# 媒体展示
st.image("path/to/image.jpg", caption="图片标题")
st.audio("path/to/audio.mp3")
st.video("path/to/video.mp4")

# 按钮和开关
if st.button("点击我"):
    st.write("按钮被点击了!")
    
agree = st.checkbox("我同意")
if agree:
    st.write("感谢您的同意!")
    
# 选择器
option = st.selectbox(
    "您喜欢哪种动物?",
    ["猫", "狗", "鸟", "鱼"]
)
st.write("您选择了:", option)

# 滑块和输入框
age = st.slider("请选择您的年龄", 0, 100, 25)
name = st.text_input("请输入您的姓名")

# 列布局
col1, col2, col3 = st.columns(3)
with col1:
    st.header("第一列")
    st.image("path/to/image1.jpg")
with col2:
    st.header("第二列")
    st.image("path/to/image2.jpg")
with col3:
    st.header("第三列")
    st.image("path/to/image3.jpg")
    
# 容器和展开区域
with st.container():
    st.write("这是一个容器")
    st.bar_chart({"数据": [1, 5, 2, 6, 3]})
    
with st.expander("点击展开更多"):
    st.write("这里是展开的内容")
    st.image("path/to/detailed_image.jpg")
    
# 侧边栏
with st.sidebar:
    st.header("侧边栏菜单")
    st.radio("选择一个选项", ["选项1", "选项2", "选项3"])

# 页面配置
st.set_page_config(
    page_title="我的应用",
    page_icon="🚀",
    layout="wide",
    initial_sidebar_state="expanded",
    menu_items={
        'Get Help': 'https://www.streamlit.io/help',
        'Report a bug': "https://github.com/streamlit/streamlit/issues",
        'About': "这是我的第一个Streamlit应用"
    }
)

# 折线图
st.line_chart({
    'data1': [1, 2, 3, 4, 5],
    'data2': [5, 4, 3, 2, 1]
})

# 条形图
st.bar_chart({
    'data1': [1, 2, 3, 4, 5],
    'data2': [5, 4, 3, 2, 1]
})

# 面积图
st.area_chart({
    'data1': [1, 2, 3, 4, 5],
    'data2': [5, 4, 3, 2, 1]
})

# 地图可视化
import pandas as pd
df = pd.DataFrame(
    {
        'lat': [34.0522, 40.7128, 37.7749],
        'lon': [-118.2437, -74.0060, -122.4194],
        'city': ['洛杉矶', '纽约', '旧金山']
    }
)
st.map(df)

# Matplotlib集成
import matplotlib.pyplot as plt
fig, ax = plt.subplots()
ax.hist([1, 2, 3, 4, 5])
st.pyplot(fig)

# Plotly集成
import plotly.express as px
df = px.data.iris()
fig = px.scatter(df, x="sepal_width", y="sepal_length", color="species")
st.plotly_chart(fig)

# Altair集成
import altair as alt
df = pd.DataFrame({
    'x': [1, 2, 3, 4, 5],
    'y': [10, 20, 30, 40, 50]
})
chart = alt.Chart(df).mark_point().encode(
    x='x',
    y='y'
)
st.altair_chart(chart)

# Bokeh集成
from bokeh.plotting import figure
p = figure(width=400, height=400)
p.circle([1, 2, 3, 4, 5], [6, 7, 2, 4, 5])
st.bokeh_chart(p)

# 数据框展示
import pandas as pd
df = pd.DataFrame({
    '姓名': ['张三', '李四', '王五'],
    '年龄': [25, 30, 35],
    '工资': [8000, 10000, 12000]
})
st.dataframe(df)  # 交互式数据框

# 静态表格
st.table(df)

# 指标卡片
st.metric("当前温度", "25°C", "2°C")

# 数据编辑器
edited_df = st.data_editor(df)

# 缓存函数结果
@st.cache_data
def load_data(file_path):
    # 模拟耗时操作
    import time
    time.sleep(2)
    import pandas as pd
    return pd.read_csv(file_path)

# 使用缓存的数据
data = load_data("large_dataset.csv")
st.dataframe(data)

# 缓存机器学习模型
@st.cache_resource
def train_model(data):
    # 模拟耗时的模型训练
    import time
    time.sleep(5)
    from sklearn.ensemble import RandomForestClassifier
    model = RandomForestClassifier()
    # 假设data中有特征和标签
    X = data.drop('label', axis=1)
    y = data['label']
    model.fit(X, y)
    return model

# 使用缓存的模型
model = train_model(data)

# 配置缓存过期时间
@st.cache_data(ttl="1h")  # 1小时后过期
def fetch_data_from_api():
    import requests
    response = requests.get("https://api.example.com/data")
    return response.json()
    
# 配置缓存键
@st.cache_data(key_func=lambda params: str(params))
def process_data_with_params(params):
    # 处理数据的代码
    return processed_data
    
# 禁用哈希
@st.cache_data(allow_output_mutation=True)
def get_dynamic_object():
    # 返回可能变化的对象
    return {'dynamic': 'object'}

# 清除所有缓存
if st.button("清除缓存"):
    st.cache_data.clear()
    st.cache_resource.clear()
    st.success("缓存已清除!")
    
# 查看缓存统计信息
with st.expander("查看缓存统计"):
    cache_info = st.cache_data.get_stats()
    st.write(cache_info)

# 初始化和使用会话状态
if 'count' not in st.session_state:
    st.session_state.count = 0
    
# 增加计数器
if st.button('增加'):
    st.session_state.count += 1
    
# 减少计数器
if st.button('减少'):
    st.session_state.count -= 1
    
# 显示当前计数
st.write(f"当前计数: {st.session_state.count}")

# 创建表单
with st.form("用户信息表单"):
    name = st.text_input("姓名")
    age = st.number_input("年龄", min_value=0, max_value=120)
    email = st.text_input("电子邮箱")
    
    # 表单提交按钮
    submitted = st.form_submit_button("提交")
    
    if submitted:
        # 处理表单数据
        st.success(f"感谢您的提交，{name}!")
        st.write(f"您的年龄是: {age}")
        st.write(f"您的电子邮箱是: {email}")

# 定义回调函数
def update_counter():
    st.session_state.count += 1
    
# 使用回调函数
st.button("增加计数", on_click=update_counter)
st.write(f"当前计数: {st.session_state.get('count', 0)}")

# 带参数的回调
def update_value(key, value):
    st.session_state[key] = value
    
st.button("设置值", on_click=update_value, args=('my_key', 'my_value'))
st.write(f"当前值: {st.session_state.get('my_key', '未设置')}")

# 多页面应用结构
import streamlit as st

# 页面配置
st.set_page_config(page_title="多页面应用")

# 侧边栏导航
page = st.sidebar.selectbox(
    "选择页面",
    ["首页", "数据展示", "模型预测"]
)

# 根据选择显示不同页面
if page == "首页":
    st.title("欢迎来到多页面应用")
    st.write("这是首页内容")
    
elif page == "数据展示":
    st.title("数据展示页面")
    st.write("这里展示数据分析结果")
    # 数据展示代码
    
elif page == "模型预测":
    st.title("模型预测页面")
    st.write("这里进行模型预测")
    # 模型预测代码

pip install streamlit

# app.py
import streamlit as st

st.title("我的第一个Streamlit应用")
st.write("欢迎使用Streamlit!")

name = st.text_input("请输入您的姓名")
if name:
    st.write(f"您好，{name}!")

streamlit run app.py

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows
venv\Scripts\activate
# macOS/Linux
source venv/bin/activate

# 安装依赖
pip install streamlit pandas matplotlib

# 导出依赖
pip freeze > requirements.txt

# 部署步骤
# 1. 将应用推送到GitHub仓库
# 2. 访问 https://share.streamlit.io/ 并连接GitHub账号
# 3. 选择要部署的仓库、分支和文件
# 4. 配置环境变量（如果需要）
# 5. 点击"Deploy"开始部署

# Dockerfile示例
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8501

HEALTHCHECK CMD curl --fail http://localhost:8501/_stcore/health

ENTRYPOINT ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

# 构建和运行Docker容器
docker build -t my-streamlit-app .
docker run -p 8501:8501 my-streamlit-app

# Nginx配置示例
server {
    listen 80;
    server_name example.com;
    
    location / {
        proxy_pass http://localhost:8501;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

streamlit export --format html app.py > app.html

# .streamlit/config.toml示例
[server]
port = 8501
enableCORS = true
enableXsrfProtection = true

[browser]
serverAddress = "localhost"
serverPort = 8501

[runner]
magicEnabled = true
fastReruns = true

[theme]
primaryColor = "#F39C12"
backgroundColor = "#2E3440"
secondaryBackgroundColor = "#3B4252"
textColor = "#ECEFF4"
font = "sans serif"

# 设置环境变量
export STREAMLIT_SERVER_PORT=8501
export STREAMLIT_SERVER_ENABLECORS=true
export STREAMLIT_SERVER_HEADLESS=true

# 运行应用
streamlit run app.py

# 启用基本身份验证
STREAMLIT_SERVER_AUTHENTICATE=1 \
STREAMLIT_SERVER_USERNAME=admin \
STREAMLIT_SERVER_PASSWORD=secret \
streamlit run app.py

# 异步处理示例
import asyncio
import streamlit as st

async def long_running_task():
    # 模拟耗时操作
    await asyncio.sleep(5)
    return "任务完成"
    
if st.button("开始长任务"):
    with st.spinner("处理中..."):
        result = await long_running_task()
    st.success(result)

# 自定义错误处理
import streamlit as st

try:
    # 可能出错的代码
    data = pd.read_csv("non_existent_file.csv")
except Exception as e:
    st.error(f"发生错误: {str(e)}")
    # 可选的错误日志记录
    import logging
    logging.error(f"应用错误: {str(e)}")

# 机器学习模型展示示例
import streamlit as st
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split

# 加载数据
@st.cache_data
def load_data():
    from sklearn.datasets import load_iris
    iris = load_iris()
    df = pd.DataFrame(iris.data, columns=iris.feature_names)
    df['target'] = iris.target
    return df, iris.target_names

data, target_names = load_data()

# 训练模型
@st.cache_resource
def train_model():
    X = data.drop('target', axis=1)
    y = data['target']
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
    model = RandomForestClassifier()
    model.fit(X_train, y_train)
    accuracy = model.score(X_test, y_test)
    return model, accuracy

model, accuracy = train_model()
st.write(f"模型准确率: {accuracy:.2f}")

# 模型预测
st.header("进行预测")
sepal_length = st.slider("花萼长度", float(data['sepal length (cm)'].min()), float(data['sepal length (cm)'].max()))
sepal_width = st.slider("花萼宽度", float(data['sepal width (cm)'].min()), float(data['sepal width (cm)'].max()))
petal_length = st.slider("花瓣长度", float(data['petal length (cm)'].min()), float(data['petal length (cm)'].max()))
petal_width = st.slider("花瓣宽度", float(data['petal width (cm)'].min()), float(data['petal width (cm)'].max()))

if st.button("预测"):
    input_data = [[sepal_length, sepal_width, petal_length, petal_width]]
    prediction = model.predict(input_data)
    prediction_proba = model.predict_proba(input_data)
    
    st.write(f"预测结果: {target_names[prediction[0]]}")
    st.write("预测概率:")
    for i, name in enumerate(target_names):
        st.write(f"{name}: {prediction_proba[0][i]:.2f}")