量化自我：使用 DuckDB 和 Streamlit 构建超高速健康仪表盘

Source: Dev.to

要进行翻译，我需要您提供文章的完整文本（除代码块和链接外的内容）。请粘贴您希望翻译的 Markdown 文本，我会按照要求将其翻译成简体中文并保留原有的格式。

为什么选择此技术栈？

如果你一直在关注现代数据栈，你会知道 OLAP for wearables 正在成为热点话题。传统的 Python 库如 Pandas 很强大，但在处理大型嵌套 XML 结构时常常会受到内存开销的限制。

Architecture 🏗️

可穿戴设备数据最大的挑战是 ETL（抽取‑转换‑加载）过程。我们需要将层次化的 XML 文件转换为扁平化、可查询的 Parquet 格式，以便 DuckDB 进行处理。

graph TD
    A[Apple Health export.xml] --> B[Python XML Parser]
    B --> C[PyArrow Table]
    C --> D[Parquet Storage]
    D --> E[DuckDB Engine]
    E --> F[Streamlit Dashboard]
    F --> G[Millisecond Insights 🚀]

前置条件

在开始之前，请确保您已经准备好 export.xml 并安装了所需的工具：

pip install duckdb streamlit pandas pyarrow

第一步 – 从 XML 混乱到 Parquet 有序

Apple 的 XML 格式…“独一无二”。我们使用 PyArrow 来定义模式，并将这些记录转换为压缩的 Parquet 文件。这样可以将文件大小降低最多 90%，并优化列式读取。

import xml.etree.ElementTree as ET
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

def parse_health_data(xml_path: str, parquet_path: str = "health_data.parquet"):
    """Parse Apple Health export XML and write a Parquet file."""
    tree = ET.parse(xml_path)
    root = tree.getroot()

    # Keep only Record elements for this dashboard
    records = [
        {
            "type": rec.get("type"),
            "value": rec.get("value"),
            "unit": rec.get("unit"),
            "creationDate": rec.get("creationDate"),
            "startDate": rec.get("startDate"),
        }
        for rec in root.findall("Record")
    ]

    df = pd.DataFrame(records)

    # Convert dates and numeric values
    df["startDate"] = pd.to_datetime(df["startDate"])
    df["value"] = pd.to_numeric(df["value"], errors="coerce")

    # Arrow Table → Parquet
    table = pa.Table.from_pandas(df)
    pq.write_table(table, parquet_path, compression="snappy")
    print("✅ Transformation complete!")

# parse_health_data("export.xml")

第 2 步 – 秘密武器：DuckDB 逻辑

我们不再使用 Pandas 将整个 Parquet 文件加载到内存中，而是直接使用 DuckDB 进行查询。这使我们能够在毫秒级别完成复杂的聚合（例如，心率变异性与睡眠质量的对比）。

import duckdb
import pandas as pd

def get_heart_rate_summary(parquet_path: str = "health_data.parquet") -> pd.DataFrame:
    """Return daily average & max heart‑rate from the Parquet file."""
    con = duckdb.connect(database=":memory:")
    query = f"""
        SELECT
            date_trunc('day', startDate) AS day,
            AVG(value) AS avg_heart_rate,
            MAX(value) AS max_heart_rate
        FROM '{parquet_path}'
        WHERE type = 'HKQuantityTypeIdentifierHeartRate'
        GROUP BY 1
        ORDER BY 1 DESC
    """
    return con.execute(query).df()

第3步 – 构建 Streamlit UI 🎨

Streamlit 让可视化这些 SQL 结果变得极其简便。只需几行代码即可添加滑块、日期选择器和交互式图表。

import streamlit as st
import plotly.express as px

# ----------------------------------------------------------------------
# Page configuration
# ----------------------------------------------------------------------
st.set_page_config(page_title="Quantified Self Dashboard", layout="wide")

st.title("🏃‍♂️ My Quantified Self Dashboard")
st.markdown(
    "Analyzing millions of health records with **DuckDB** speed."
)

# ----------------------------------------------------------------------
# Load data
# ----------------------------------------------------------------------
df_hr = get_heart_rate_summary()

# ----------------------------------------------------------------------
# Layout
# ----------------------------------------------------------------------
col1, col2 = st.columns(2)

with col1:
    st.subheader("Heart Rate Trends")
    fig = px.line(
        df_hr,
        x="day",
        y="avg_heart_rate",
        title="Average Daily Heart Rate",
        markers=True,
    )
    st.plotly_chart(fig, use_container_width=True)

with col2:
    st.subheader("Raw DuckDB Query Speed")
    st.code(
        """SELECT avg(value) FROM 'health_data.parquet' 
WHERE type = 'HKQuantityTypeIdentifierHeartRate'"""
    )
    st.success("Query executed in ~0.002 s")

运行应用：

streamlit run your_script.py

“官方”扩展方式 🥑

在本地构建固然有趣，但生产级数据工程需要更稳健的模式：多用户环境、自动化摄取以及高级机器学习流水线。想深入了解这些主题，请查看 WellAlly 博客。他们提供了出色的架构模式和可直接用于生产的示例，远超本地脚本的范畴。

结论

通过将原始 XML 解析切换为 DuckDB + Parquet 工作流，我们把一个缓慢的数据问题转变为高性能的分析工具。你不再需要庞大的集群来分析数百万条健康记录——只需一台笔记本电脑、几个 Python 库，以及一点好奇心。量化愉快！

你在追踪什么？
无论是步数、睡眠还是编码时间，欢迎在评论区告诉我你是如何可视化自己的生活的！👇