Dagster Pipes系列-1：调用外部Python脚本

news2025/5/14 4:36:00

本文是"Dagster Pipes教程"的第一部分，介绍如何通过Dagster资产调用外部Python脚本并集成到数据管道中。首先，创建Dagster资产subprocess_asset，利用PipesSubprocessClient资源执行外部脚本external_code.py，实现跨进程的数据处理。通过dagster dev启动UI，可在Dagster界面中监控子进程的执行状态和日志输出，包括标准输出(stdout)内容。本文详细讲解了资产定义、资源注入及命令执行的完整流程，为后续修改外部代码以支持Dagster Pipes通信奠定基础。此方法适用于需要将现有脚本集成到Dagster数据管道的场景，提升自动化与可观测性。完成本部分后，读者可继续学习第二部分，掌握如何增强外部脚本与Dagster的交互能力。

教程概述

本教程将指导你完成以下步骤：

创建一个调用外部Python脚本的Dagster资产
定义必要的Dagster资源(resources)
在Dagster UI中运行并查看结果

前提条件

在开始之前，请确保你已经：

安装了Dagster
创建了一个名为external_code.py的独立Python脚本，内容如下：

import pandas as pd

def main():
    orders_df = pd.DataFrame({
        "order_id": [1, 2],
        "item_id": [432, 878]
    })
    total_orders = len(orders_df)
    print(f"processing total {total_orders} orders")

第一步：定义Dagster资产

首先，在与external_code.py相同的目录下创建一个名为dagster_code.py的新文件。

1.1 创建资产定义

将以下代码复制到dagster_code.py中：

import shutil
import dagster as dg

@dg.asset
def subprocess_asset(
    context: dg.AssetExecutionContext,
    pipes_subprocess_client: dg.PipesSubprocessClient
) -> dg.MaterializeResult:
    cmd = [
        shutil.which("python"),
        dg.file_relative_path(__file__, "external_code.py")
    ]
    return pipes_subprocess_client.run(
        command=cmd,
        context=context
    ).get_materialize_result()

代码解析：

我们创建了一个名为subprocess_asset的资产
使用AssetExecutionContext作为上下文参数，它提供了系统信息如资源、配置和日志记录
指定了PipesSubprocessClient资源
构建了一个命令列表来执行外部脚本
使用pipes_subprocess_client.run()方法在管道会话中同步执行子进程

1.2 从资产调用外部代码

上述代码中的关键部分是：

pipes_subprocess_client.run(
    command=cmd,
    context=context
).get_materialize_result()

这段代码做了什么：

PipesSubprocessClient资源暴露了一个run方法
当资产执行时，这个方法会在管道会话中同步执行子进程
返回一个PipesClientCompletedInvocation对象
可以使用get_materialize_result()方法访问子进程报告的MaterializeResult事件

第二步：定义Definitions对象

为了让Dagster工具（如CLI、UI和Dagster+）能够加载和访问资产及子进程资源，我们需要创建一个Definitions对象。

在dagster_code.py文件末尾添加以下代码：

from dagster import Definitions

defs = Definitions(
    assets=[subprocess_asset],
    resources={
        "pipes_subprocess_client": dg.PipesSubprocessClient()
    }
)

此时，dagster_code.py文件应该如下所示：

import shutil
import dagster as dg

@dg.asset
def subprocess_asset(
    context: dg.AssetExecutionContext,
    pipes_subprocess_client: dg.PipesSubprocessClient
) -> dg.MaterializeResult:
    cmd = [
        shutil.which("python"),
        dg.file_relative_path(__file__, "external_code.py")
    ]
    return pipes_subprocess_client.run(
        command=cmd,
        context=context
    ).get_materialize_result()

from dagster import Definitions

defs = Definitions(
    assets=[subprocess_asset],
    resources={
        "pipes_subprocess_client": dg.PipesSubprocessClient()
    }
)