DolphinScheduler 能用 Python 脚本编排工作流了！PyDolphinScheduler 简介与使用演示

海豚调度 2022-11-08

3404

点亮 ⭐️ Star · 照亮开源之路

重要功能介绍

2.0.5 版本更新之后，Apache DolphinScheduler 新增了 Python API 功能，用户可以通过 Python 脚本编排工作流，最后实现工作流的创建、更新、调度等操作，这给 Python 用户带来了很多便利。

Apache DolphinScheduler 是一个分布式、高扩展性的可视化开源工作流任务调度框架，适用于企业级应用场景，提供了可视化任务操作、工作流调度和整个数据处理流程的可视化解决方案。

Apache DolphinScheduler 提供了丰富的组件，让用户可以进行 MR、SageMaker、DMS、DataSync 等 AWS 服务的调用，同时也能调用 Spark、Flink、Hive 等私有化部署的服务；它还提供了针对调度核心场景的工具，如失败重试、失败告警、任务 SLA、复杂依赖处理等；提供调度相关的丰富功能，如外部资源文件管理、集群物理机资源监控；以及多种 Metric 用于外部监控。

DolphinScheduler Python

API

有了 Python API，用户可以通过 Python 脚本编排 workflow，最后实现工作流的创建、更新、调度等操作。

安装

你可以通过 pip 简单方便地安装 PyDolphinScheduler：

python -m pip install apache-dolphinscheduler

运行

我们提供了开箱即用的例子，用户可以通过以下命令运行 Python API：

wget https://raw.githubusercontent.com/apache/dolphinscheduler/dev/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/examples/tutorial.py
python tutorial.py

Python API 功能

Python API 提供了我们能直接在 DolphinScheduler Web UI 中可以操作的全部功能，例如工作流的创建、任务的创建、工作流的调度、资源文件的实用、参数传递等。目前已经支持 DolphinScheduler 中绝大部分的任务类型，用户可以通过 Python API 完成 DolphinScheduler 绝大部分核心功能的体验。

如果想要通过 DolphinScheduler Python API 创建并运行 DolphinScheduler 工作流，我们需要先启动 DolphinScheduler 服务，具体可以在 https://dolphinscheduler.apache.org/python/dev/start.html#start-python-gateway-service 中查看安装和运行的方式。本文主要带大家领略 Python API 的魅力，所以会着重讲解 Python API 任务。

示例

这是一个关于如何使用 Python API 运行 DolphinScheduler 工作流的示例，我们需要从官网下载资源，并进行一些计算。

概览

这是我们的代码概览，在这个脚本中用户需要下载 DolphinScheduler 源代码，然后找到最大的文件和最常用的文件扩展名。完成这项工作需要 3 步。所有的步骤在 DolphinScheduler 中被称为“task”，而我们想要完成的整个工作称为“ProcessDefinition”。除了这 3 个主要步骤之外，我们还有一些额外的步骤来准备我们的环境并确保主要步骤正常工作。例如，我们需要创建新目录来保存我们从官网上获得的资源，并确保传入的压缩包与已有环境不存在冲突。下载资源后，我们需要对其进行压缩并从压缩包中提取内容。

from pydolphinscheduler.core.process_definition import ProcessDefinition
from pydolphinscheduler.tasks.python import Python
from pydolphinscheduler.tasks.shell import Shell
download_dir = "/tmp/demo"
store_dir = "dolphinscheduler"
download_link = "https://github.com/apache/dolphinscheduler/archive/refs/heads/dev.zip"
file_name = download_link.split("/")[-1]
def largest_size():
    from pathlib import Path
    download_dir = "/tmp/demo"
    store_dir = "dolphinscheduler"
    result = (None, 0)
    paths = Path(download_dir).joinpath(store_dir).glob("**/*")
    for path in paths:
        # skip is path is directory
        if path.is_dir():
            continue
        file_size = path.stat().st_size
        if result[0] is None or file_size > result[1]:
            result = (path.name, file_size)
    print(result)
def most_frequently():
    from pathlib import Path
    download_dir = "/tmp/demo"
    store_dir = "dolphinscheduler"
    ext_cnt = {}
    paths = Path(download_dir).joinpath(store_dir).glob("**/*")
    for path in paths:
        # skip is path is directory
        if path.is_dir():
            continue
        ext = path.suffix
        ext_cnt[ext] = ext_cnt[ext] + 1 if ext in ext_cnt else 1
    print(max(ext_cnt.items(), key=lambda p: p[1]))
with ProcessDefinition(
        name="top_ten_size_files",
        tenant="zhongjiajie",
) as pd:
    prepare = Shell(
        name="prepare_dir",
        command=f"mkdir -p {download_dir}; rm -rf {download_dir}/*"
    )
    download = Shell(
        name="download_resources",
        command=f"wget -P {download_dir} {download_link}"
    )
    compress = Shell(
        name="compress_tar",
        command=f"cd {download_dir}; unzip {file_name} -d {store_dir}"
    )
    largest_file = Python(
        name="largest_file",
        definition=largest_size,
    )
    most_type = Python(
        name="most_type",
        definition=most_frequently,
    )
    prepare >> download >> compress >> [
        largest_file,
        most_type,
    ]
    pd.run()

下载资源

在这个任务中，我们使用 shell 命令 wget 从 GitHub 下载资源，可以简单地使用 PyDolphinScheduler 的内置 Shell 将 wget 命令传递给参数 command。

你可以看到我们在这里使用名为download_dir和download_link的 Python 变量来让我们的代码可维护，实际运行命令是wget -P tmp/demo https://github.com/apache/dolphinscheduler/archive/ refs/heads/dev.zip。完成后，DolphinScheduler 源代码将下载到目录/tmp/demo。

download = Shell(
    name="download_resources",
    command=f"wget -P {download_dir} {download_link}"
)

下载资源前后需要做的事

正如我们在 overview 中提到的，我们在从官网下载资源之前或之后都需要进行一些额外步骤。对于环境准备，我们使用单个Shell任务来确保下载目录存在并且目录中没有其他资源文件。传递命令 mkdir -p tmp/demo; rm -rf tmp/demo/* 到参数 command（注意我们在这里也使用 Python 变量）。我们必须为.zip文件名压缩源代码包，然后将它们解压缩到特定目录，以便我们可以在此路径上运行一些 Python 代码。

prepare = Shell(
    name="prepare_dir",
    command=f"mkdir -p {download_dir}; rm -rf {download_dir}/*"
)
compress = Shell(
    name="compress_tar",
    command=f"cd {download_dir}; unzip {file_name} -d {store_dir}"
)

计算体积最大的文件

我们使用 Python 任务来计算 DolphinScheduler 源代码中的最大文件和使用最频繁的文件类型。可以看到，我们还需要命名 Python 任务，这里我将其命名为 largest_file，再将函数传递给 Python 任务的参数 definition。这一步的核心逻辑来自 largest_size 函数。在这个函数中，我们通过pathlib库遍历源代码目录，找到最大的文件。使用名为 result 的元组来存储命名文件及其大小，最后打印 result ，Console 上就显示出来了。

def largest_size():
    from pathlib import Path
    download_dir = "/tmp/demo"
    store_dir = "dolphinscheduler"
    result = (None, 0)
    paths = Path(download_dir).joinpath(store_dir).glob("**/*")
    for path in paths:
        # skip is path is directory
        if path.is_dir():
            continue
        file_size = path.stat().st_size
        if result[0] is None or file_size > result[1]:
            result = (path.name, file_size)
    print(result)
largest_file = Python(
    name="largest_file",
    definition=largest_size,
)

计算出现频率最高的文件类型

这里的做法与 最大文件 相同，我们使用 Python 任务，将 most_type 作为 name， most_frequently 作为definition。为了找到最常用的文件类型，我们创建了一个字典来保存所有存在的文件扩展名。再用pathlib库来遍历源代码，这个过程中要注意跳过没有扩展名的目录和文件，从字典中找到最常用的值。

def most_frequently():
    from pathlib import Path
    download_dir = "/tmp/demo"
    store_dir = "dolphinscheduler"
    ext_cnt = {}
    paths = Path(download_dir).joinpath(store_dir).glob("**/*")
    for path in paths:
        # skip is path is directory
        if path.is_dir():
            continue
        ext = path.suffix
        # skip file without suffix
        if ext == "":
            continue
        ext_cnt[ext] = ext_cnt[ext] + 1 if ext in ext_cnt else 1
    print(max(ext_cnt.items(), key=lambda p: p[1]))
most_type = Python(
    name="most_type",
    definition=most_frequently,
)

设置任务依赖

设置好之后，我们需要声明任务的依赖，不用说，在此之前我们应该先运行prepare任务，因为没有压缩包就不能compress，所以我们需要将任务download设置为compress的上游。另一个计算可以并行运行，因此我们将它们都设置在compress任务的下游，我们在这里用语法糖compress >> [largest_file，most_type]，可以在一个 statement 中设置两个或多个依赖项。

运行！

添加以下代码到脚本的尾，就可以把工作流和任务提交给 DolphinScheduler 并默认触发。

pd.run()

所有这些完成后，就可以在 Python Console 中运行代码，并观察运行结果。之后就可以看到 DolphinScheduler UI 中运行的工作流和任务，默认是

http://127.0.0.1:12345/dolphinscheduler。
python3 demo.py

总结

在本文中，我们介绍了什么是 DolphinScheduler 及其最近新增的 Python API 功能，并演示了如何使用 Python API 构建具有多个任务的工作流，如何设置任务，以及如何将其提交给 DolphinScheduler。

你可以在 task 中找到 Python API 支持的所有任务。更多有关 DolphinScheduler Python API 的信息，请访问文档(https://dolphinscheduler.apache.org/python/dev/index.html）。

参与贡献

随着国内开源的迅猛崛起，Apache DolphinScheduler 社区迎来蓬勃发展，为了做更好用、易用的调度，真诚欢迎热爱开源的伙伴加入到开源社区中来，为中国开源崛起献上一份自己的力量，让本土开源走向全球。

参与 DolphinScheduler 社区有非常多的参与贡献的方式，包括：

贡献第一个PR(文档、代码) 我们也希望是简单的，第一个PR用于熟悉提交的流程和社区协作以及感受社区的友好度。

社区汇总了以下适合新手的问题列表：https://github.com/apache/dolphinscheduler/issues/5689

非新手问题列表：https://github.com/apache/dolphinscheduler/issues?q=is%3Aopen+is%3Aissue+label%3A%22volunteer+wanted%22

如何参与贡献链接：https://dolphinscheduler.apache.org/zh-cn/community/development/contribute.html

来吧，DolphinScheduler开源社区需要您的参与，为中国开源崛起添砖加瓦吧，哪怕只是小小的一块瓦，汇聚起来的力量也是巨大的。

参与开源可以近距离与各路高手切磋，迅速提升自己的技能，如果您想参与贡献，我们有个贡献者种子孵化群，可以添加社区小助手微信(Leonard-ds) ，手把手教会您( 贡献者不分水平高低，有问必答，关键是有一颗愿意贡献的心 )。

添加小助手微信时请说明想参与贡献。

来吧，开源社区非常期待您的参与。

< 🐬🐬 >

☞挑战海量数据：基于Apache DolphinScheduler对千亿级数据应用实践

☞金融科技数据中台基于 DolphinScheduler 的应用改造

☞突破单点瓶颈、挑战海量离线任务，Apache Dolphinscheduler在生鲜电商领域的落地实践

☞名额已排到10月 | Apache DolphinScheduler Meetup分享嘉宾继续火热招募中

☞【Meetup讲师】您有一张社区认证讲师证书未领取，点击领取！

☞Workflow as code+SageMaker, DolphinScheduler的机器学习选股系统新玩法

我知道你在看哟

python python函数工作流 apache

文章转载自海豚调度，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

DolphinScheduler 能用 Python 脚本编排工作流了！PyDolphinScheduler 简介与使用演示

评论