企业官网建设流程全解析

dlt dltHub 实战：构建、本地运行与云端部署 Streamlit 数据仪表盘

【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy 🛠️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt

本文基于 dlt 仓库中的 dltHub 官方指南 Build and deploy a Streamlit app 展开，完整走通"在 dltHub workspace 中用 Streamlit 构建一个读取 dlt 数据集的仪表盘、本地调试、配置accessprofile、部署到 dltHub 并生成公开分享链接"的全流程。读完之后，你可以掌握dlthub local serve/dlthub deploy/dlthub serve/dlthub job publish这条命令链的实际用法，并理解 dlt 的 workspace 框架是如何自动识别 Streamlit 模块、生成 HTTP 触发的交互式任务（interactive job）并以受控参数启动streamlit run进程的。

Streamlit 应用在 dltHub 中的定位：一个普通 .py 文件就是一个交互式任务

dltHub 对 Streamlit 的定义非常简单：一个导入了streamlit的普通.py文件就是一个 Streamlit 应用，运行时（runtime）会把它当作交互式仪表盘（dashboard）来提供（serve）。你不需要注册路由、不需要声明框架实例，只要模块命名空间里能"看见"streamlit模块对象即可。

这一约定在源码中有直接实现。dlt 的框架检测器按 marimo → MCP → Streamlit 的顺序探测每个被 manifest 导入的模块，见 detectors.py 中的_detect_streamlit：

def _detect_streamlit(module: ModuleType) -> Optional[TJobDefinition]: """Detect streamlit usage by finding the streamlit module in namespace.""" for obj in module.__dict__.values(): if isinstance(obj, ModuleType) and getattr(obj, "__name__", "") == "streamlit": break else: return None # ... 生成 job 定义

一旦命中，检测器会为该模块生成如下任务定义：

• job_ref形如jobs.<module_name>（由 _module_job_ref 构造）；
• entry_point.job_type = "interactive"，launcher 指向 launchers/streamlit.py；
• expose = {interface: "gui", category: "dashboard"}，即对外暴露为一个 GUI 类仪表盘；
• triggers = [http()]，表示它由 HTTP 请求触发而非调度触发；
• execute = {concurrency: 1}，交互式任务同一时刻只允许一个运行实例（见 _INTERACTIVE_EXECUTION）；
• 模块 docstring 会被直接取作任务描述，所以仪表盘脚本首行注释（如"""Sample shop dashboard."""）会出现在 workspace 的任务列表中。

从源码结构看有两点值得注意：一是检测基于"模块对象存在于命名空间"，所以import streamlit as st、from streamlit import ...等任何别名写法都能被识别，这由测试用例 test_detectors.py 及其用例文件 streamlit_standard.py、streamlit_aliased.py 验证；二是当同一模块中同时存在 FastMCP 与 Streamlit 时 MCP 检测优先（见 test_mcp_wins_over_streamlit）。

前置条件：安装依赖并先加载数据

第一步是把 Streamlit 加入 workspace 依赖：

uv add streamlit

文档中的示例仪表盘读取sample_shop_pipeline加载的数据，该 pipeline 由uvx dlthub-start@latest脚手架生成（详见 Deploy your first pipeline）。关键点：仪表盘只能展示已经加载过的数据，且数据必须落在仪表盘将要读取的同一个 destination 上。跳过这一步的后果是应用能正常启动，但每次读取都返回 "table not found"。

两种数据准备方式，取决于你当前的调试目标：

# 方式一：本地加载（dev profile，DuckDB），让仪表盘在 `dlthub local serve` 下可用 uv run dlthub local run load_sample_shop # 方式二：在远程部署前，先针对 dltHub 上的 prod destination 加载 uv run dlthub run load_sample_shop

local run对应 CLI 的dlthub local命令组中"在本地运行单个 batch 任务"的操作（见 LocalWorkspaceCommand），它解析本地 profile 配置后启动 pipeline；而dlthub run则是面向 dltHub 远端的运行命令。两条路径的区别本质上是 profile（dev/prod 指向的 destination 不同）的区别。

编写仪表盘：用 dlt.dataset 直连已加载数据

在 workspace 根目录创建sample_shop_dashboard.py：

"""Sample shop dashboard.""" import dlt import streamlit as st st.set_page_config(page_title="Sample shop", layout="wide") st.title("Sample shop orders") @st.cache_data def load_data(): dataset = dlt.dataset(destination="warehouse", dataset_name="sample_shop") return dataset["orders"].df(), dataset["customers"].df() orders, customers = load_data() col1, col2, col3, col4 = st.columns(4) col1.metric("Orders", f"{len(orders):,}") col2.metric("Customers", f"{len(customers):,}") col3.metric("Total revenue", f"${orders['order_total'].sum():,.2f}") col4.metric("Avg order value", f"${orders['order_total'].mean():.2f}") st.subheader("Top customers by spend") by_customer = ( orders.merge(customers, left_on="customer_id", right_on="id", suffixes=("_o", "_c")) .groupby("name", as_index=False)["order_total"].sum() .sort_values("order_total", ascending=False) .head(10) ) st.dataframe( by_customer.rename(columns={"name": "Customer", "order_total": "Total spend ($)"}), width="stretch", hide_index=True, ) st.subheader("Orders by store") by_store = orders.groupby("store_id").size().rename("orders").sort_values(ascending=False) st.bar_chart(by_store)

代码要点：

1. dlt.dataset(destination="warehouse", dataset_name="sample_shop")是核心数据入口。它不重新执行 pipeline，而是直连一个"已经加载完成"的数据集（dataset 访问机制见 Dataset Access），destination="warehouse"指向前面配置中的[destination.warehouse]，dataset_name则是 pipeline 加载数据时产生的数据集名。dataset["orders"].df()把表转成 pandas DataFrame 供 Streamlit 组件消费。
2. @st.cache_data包裹取数函数：Streamlit 每次 widget 交互都会重跑脚本，缓存可以避免每次点击都重新查询 destination。
3. 其余部分是标准 Streamlit 用法：4 列 metric 卡片展示订单数、客户数、总收入与客单价；merge + groupby算出消费 Top 10 客户；st.bar_chart展示各门店订单量。

本地运行：dlthub local serve

uv run dlthub local serve sample_shop_dashboard.py

serve子命令的 CLI 帮助文本是 "Serve an interactive workspace job locally (notebook, dashboard, app)"（见 commands.py），它与dlthub local run共用同一套选择器参数（见 _add_common_run_args），常用参数包括：

文档中说明的行为是：该命令在 workspace 当前激活的本地 profile（默认dev，读取.dlt/config.toml）下启动仪表盘并在浏览器中打开。这也解释了为什么前置步骤要用local run加载到 DuckDB——本地 serve 走 dev profile，读的是本地 destination。

源码级原理：dlt 如何启动你的 Streamlit 进程

dlthub serve（本地或远端）最终都通过同一个 launcher 启动 Streamlit。launchers/streamlit.py 的run()做了四件事：

1. 应用 job 配置与环境准备：apply_job_configuration从jobs.<section>.<name>配置节解析任务行为设置，prepare_run_env在用户代码运行前把 profile、interval、时区写入环境变量（见 _launcher.py）；

2. 解析运行时参数：端口port必须由运行时通过run_args.port下发（缺失会抛ValueError），可选的base_path会转成--server.baseUrlPath参数，用于应用挂载在 URL 子路径下；

3. 解析框架专属配置：从JOBS与任务所在 section 两个配置节解析 StreamlitConfiguration：

   配置项	默认值	含义
   enable_cors	False	是否启用 CORS，默认在代理后关闭
   enable_xsrf_protection	False	是否启用 XSRF 防护，默认在代理后关闭
   gather_usage_stats	False	是否向 Streamlit 上报使用统计

4. 拼装命令行并 exec：最终执行形如：

   streamlit run <script_path> \ --server.address=0.0.0.0 \ --server.port=<port> \ --server.headless=true \ --server.enableCORS=false \ --server.enableXsrfProtection=false \ --server.enableWebsocketCompression=false \ --browser.gatherUsageStats=false \ --browser.serverAddress=0.0.0.0 \ --browser.serverPort=<port> \ [--server.baseUrlPath=<base_path>]

   在 POSIX 上通过os.execvp用 Streamlit 进程替换当前进程（Windows 上回退为 spawn + wait 以正确传播退出码，见 exec_process）。

从源码结构看，这套"检测器产出 job 定义 → launcher 拼装框架命令 → exec 进程"的模式对 marimo、FastMCP、Streamlit 三种框架统一适用（launcher 注册表见 launchers/init.py），Streamlit 是其中唯一的纯"模块导入即识别"的框架。

配置 access profile

dlthub serve默认在accessprofile 下运行交互式任务。最简脚手架不预置accessprofile，因此部署前需要自己创建。在.dlt/access.config.toml中声明 destination 类型：

[destination.warehouse] destination_type = "motherduck"

在.dlt/access.secrets.toml中提供凭据。注意对 MotherDuck 来说，database和password必须同时写在 secrets 文件中：

[destination.warehouse.credentials] database = "dlt_test" password = "<read-only motherduck JWT>"

这里使用只读（read-only）JWT 是一个有意的安全实践：仪表盘是只读消费端，access profile 的凭据不应具备写权限。完整的 profile 模型（local/prod/access等如何组合 config 与 secrets 文件）可参阅 Profiles in dltHub。

部署到 dltHub：manifest、deploy、serve 与公开链接

第一步，把仪表盘模块加入__deployment__.pymanifest，让 workspace 知晓它的存在：

"""Minimal dltHub workspace.""" from pipeline import load_sample_shop import sample_shop_dashboard # module-import → one job __all__ = ["load_sample_shop", "sample_shop_dashboard"]

import sample_shop_dashboard这种模块级导入会经过上面的框架检测链，被识别为一个jobs.sample_shop_dashboard交互式任务。随后部署并远程启动：

uv run dlthub deploy # 发布 manifest + 上传代码 uv run dlthub serve sample_shop_dashboard.py # 远程启动应用并打开 URL

访问控制方面，dlthub serve启动的应用运行在 workspace 认证之后——只有你的账号能打开该链接。需要对外分享时，用job publish生成公开 URL，用job unpublish撤销：

uv run dlthub job publish sample_shop_dashboard.py # 生成公开 URL uv run dlthub job unpublish sample_shop_dashboard.py # 撤销

小结与延伸阅读

整条链路可以概括为：uv add streamlit→ 先加载数据（local run 或 run）→ 编写一个dlt.dataset(...)取数的 Streamlit 脚本 →dlthub local serve本地验证 → 配置accessprofile 的 config/secrets → 在__deployment__.py中 import 该模块 →dlthub deploy+dlthub serve（或job publish公开化）。仓库中可进一步深入的文件：

• Streamlit launcher 实现：最终拼出的streamlit run命令行与配置注入；
• 框架检测器：marimo/MCP/Streamlit 的识别顺序与 job 定义生成，以及模块级__trigger__/__expose__/__require__dunder 的覆盖机制；
• Streamlit 检测测试 与用例脚本：验证不同 import 写法均能被识别；
• dltHub 命令行接口文档 与 dlt.dataset 文档：命令全集与数据集访问 API。

【免费下载链接】dltdata load tool (dlt) is an open source Python library that makes data loading easy 🛠️项目地址: https://gitcode.com/GitHub_Trending/dl/dlt