📚 课程目录第4章 Dagster环境搭建

4、Dagster环境搭建：安装Dagster、启动Dagit UI、创建第一个Pipeline、理解Dagit界面

好，咱们正式开始动手了。

前面聊了那么多数据血缘的理论，说实话，没有工具落地，一切都是纸上谈兵。Dagster 这个框架，我用了快两年，从 0.9.x 版本一路跟到现在，看着它从一个小众工具变成数据工程圈的明星项目。今天我们就把它装起来，跑通第一个 Pipeline，顺便把 Dagit 那个界面摸透。

4.1 环境准备与安装

先说说我的个人习惯。我一般会在项目根目录下创建一个虚拟环境，专门给 Dagster 用。为什么？因为 Dagster 的依赖管理其实挺干净的，但如果你跟 Airflow 或者 Prefect 混装在一个环境里，偶尔会有版本冲突。我曾经在一个客户现场遇到过 dagster 和 apache-airflow 的依赖打架，折腾了半小时才定位到问题。从那以后，我每个数据编排项目都单独建虚拟环境。

推荐环境： Python 3.8 以上，建议 3.10 或 3.11。Dagster 1.3+ 对 Python 3.11 支持已经很稳定了。

安装命令非常简单，一行搞定：

pip install dagster dagster-webserver

这里我解释一下为什么是 dagster-webserver 而不是 dagit。在 1.0 版本之前，那个 Web UI 叫 Dagit。1.0 之后，官方把它重命名为 dagster-webserver，但大家还是习惯叫它 Dagit。嗯，我也改不过来，后面我就统称 Dagit 了，你心里知道就行。

安装完成后，验证一下：

dagster --version

如果看到类似 dagster, version 1.5.4 的输出，说明安装成功了。

4.2 启动 Dagit UI

安装完别急着写代码。我们先启动 Dagit，看看它长什么样。你想想看，一个连界面都没见过的工具，你怎么知道它好不好用？

dagster dev

这个命令会启动一个开发服务器，默认监听在 localhost:3000。打开浏览器，输入 http://localhost:3000，你应该能看到一个空白的界面——因为还没有任何 Pipeline 加载进来。

注意： 如果你用的是 dagster-webserver 命令，默认端口是 3000。但如果你在 Docker 或者远程服务器上跑，记得检查端口映射和防火墙设置。我有个同事曾经在 AWS EC2 上折腾了半天，最后发现是安全组没开端口。

看到那个空白的 Dagit 界面了吗？别慌，接下来我们就往里面塞东西。

4.3 创建第一个 Pipeline

现在我们来写第一个 Pipeline。我习惯在项目根目录下建一个 pipelines 文件夹，里面放各个 Pipeline 的代码。但今天简单点，直接在根目录下建一个 Python 文件。

创建一个 hello_dagster.py，写入以下代码：

from dagster import job, op

@op
def say_hello():
    return "Hello, Dagster!"

@op
def greet(context, name: str):
    context.log.info(f"Greeting {name}")
    return f"Hello, {name}!"

@job
def hello_pipeline():
    greet(say_hello())

这段代码干了三件事：

定义了两个 op：say_hello 和 greet。op 是 Dagster 里的最小执行单元，你可以把它理解成一个函数，但它比函数多了上下文、日志、资源注入等能力。
定义了一个 job：hello_pipeline。job 就是一组 op 的编排。这里我们把 say_hello 的输出传给 greet。
注意那个 context 参数：这是 Dagster 自动注入的上下文对象，里面包含了日志、配置、资源等信息。我第一次用的时候还纳闷，这个参数哪来的？后来才明白，Dagster 会在运行时自动注入，你只需要在 op 函数签名里声明它就行。

保存文件后，回到终端，用这个命令运行：

dagster job execute -f hello_dagster.py -j hello_pipeline

你应该能看到类似这样的输出：

2024-01-15 10:23:45 - dagster - INFO - hello_pipeline - 9c8f2a1b-... - say_hello - STEP_START
2024-01-15 10:23:45 - dagster - INFO - hello_pipeline - 9c8f2a1b-... - say_hello - STEP_SUCCESS
2024-01-15 10:23:45 - dagster - INFO - hello_pipeline - 9c8f2a1b-... - greet - STEP_START
2024-01-15 10:23:45 - dagster - INFO - Greeting Hello, Dagster!
2024-01-15 10:23:45 - dagster - INFO - hello_pipeline - 9c8f2a1b-... - greet - STEP_SUCCESS

看到 STEP_SUCCESS 了吗？恭喜，你的第一个 Dagster Pipeline 跑通了。

4.4 理解 Dagit 界面

现在我们把刚才的 Pipeline 加载到 Dagit 里看看。重新启动 dagster dev，这次它会自动扫描当前目录下的 Python 文件，找到所有 @job 装饰的 Pipeline。

打开浏览器，你应该能看到 hello_pipeline 出现在左侧的 Pipeline 列表里。点击它，进入详情页。我们来拆解一下 Dagit 的核心界面：

区域	名称	作用
左侧栏	Pipeline 列表	显示所有已加载的 Pipeline 和 Schedule/Sensor
中央画布	DAG 可视化	以有向无环图的形式展示 op 之间的依赖关系
右侧面板	详情面板	点击任意 op 后，显示其输入、输出、配置、日志等信息
顶部工具栏	运行控制	手动触发运行、查看运行历史、切换 Launchpad 等

我个人最喜欢的是那个 DAG 可视化画布。你想想看，以前用 Airflow 的时候，DAG 图是静态的，点一下节点只能看个日志。Dagit 的交互感强很多——你可以拖动画布、缩放、点击节点查看详情，甚至直接在界面上修改配置然后重新运行。

  核心概念： Dagit 不仅仅是一个监控面板，它是一个完整的开发环境。你可以在 Launchpad 里直接编辑 Pipeline 的配置，然后一键运行。这对于调试阶段来说，效率提升非常明显。

我们来看一张流程图，帮你把 Dagster 的核心组件关系理清楚：

回到 Dagit 界面，有几个地方我建议你重点看看：

Launchpad 面板：在顶部工具栏点击 "Launchpad"，你会看到一个 JSON 编辑器。这里可以手动传入配置，覆盖 op 的默认参数。我在调试数据源连接时经常用这个功能，不用改代码就能换数据库地址。
运行历史：每次运行 Pipeline，Dagit 都会记录完整的执行日志。你可以看到每个 op 的启动时间、结束时间、输出结果，甚至 CPU 和内存使用情况。这对于排查性能瓶颈非常有用。
资产视图：如果你用了 @asset 装饰器（后面章节会讲），Dagit 会生成一个资产目录，展示所有数据产物的血缘关系。这是数据血缘追踪的核心入口。

小技巧： 在 Dagit 的 DAG 画布上，按住 Ctrl 键并滚动鼠标滚轮可以缩放。按住 Shift 键并拖动可以平移。这个操作习惯跟 Figma 很像，上手很快。

嗯，到这里，你的 Dagster 环境已经搭好了，第一个 Pipeline 也跑通了，Dagit 界面也摸了个大概。接下来我们会深入 op 和 job 的细节，但今天先到这里。记住，工具是拿来用的，不是拿来学的。多动手点一点 Dagit 的各个面板，你会发现很多隐藏的小功能。