python工作任务流flow实时框架：prefect

我不是女神ヾ 2023-09-27 09:16 112阅读 0赞

## python工作任务流flow实时框架：prefect ##

prefect是一个python的工作任务流调度实时框架，prefect可以快速构建平台系统复杂模块间工作流的监测。当平台系统模块之间的调用链越来越复杂时候，任务执行起来，已经很难盘点清楚程序逻辑和工作流运转的路线图。prefect通过简单的@注解，即可清晰在线、实时的梳理清楚系统运行“思维导图”。

prefect是github的开源项目：

[GitHub - PrefectHQ/prefect: The easiest way to coordinate your dataflowThe easiest way to coordinate your dataflow. Contribute to PrefectHQ/prefect development by creating an account on GitHub.![fluidicon.png][]https://github.com/PrefectHQ/prefect][GitHub - PrefectHQ_prefect_ The easiest way to coordinate your dataflowThe easiest way to coordinate your dataflow. Contribute to PrefectHQ_prefect development by creating an account on GitHub._fluidicon.png_https_github.com_PrefectHQ_prefect]

安装：

pip install -U prefect

### 工作任务流例子 ###

@flow可以嵌套使用，@flow里面可以调用@task的任务。但@task不能反过调用@flow，原则上尽量使用@flow，@task是程序代码的最小原子单位。

import time
    
    from prefect import flow, task, get_run_logger
    
    
    @task(retries=2, retry_delay_seconds=30)
    def failure():
        print('running')
        raise Exception('异常')
    
    
    @flow(name="my_flow_b", timeout_seconds=15)
    def flowb(param: str):
        time.sleep(20)
        return 'OKb'
    
    
    @flow(name="my_flow_a")
    def flowa(param: str):
        time.sleep(15)
    
        failure.submit()
    
        return 'OKa'
    
    
    @flow(name="my_flow_2", description='第二个工作流')
    def flow2():
        logger = get_run_logger()
        logger.info('启动flow2')
        time.sleep(3)
        flowb('start b')
        return 'OK2'
    
    
    @flow(name="my_flow_1", description='第一个工作流')
    def flow1():
        logger = get_run_logger()
        logger.info('启动flow1')
        time.sleep(2)
        flowa('start a')
        return 'OK1'
    
    
    if __name__ == '__main__':
        flow1()
        flow2()

程序运行后，在控制台使用命令启动prefect服务器端：

> prefect orion start

![750b7e424e6041db85b5d63384b54812.png][]

默认prefect启动在http://127.0.0.1:4200，浏览器打开：

![b908628becce464fa8ef35a1891e48f8.png][]

查看flow1运行时日志：

![d2361db243e74b588407615e360b08a2.png][]

flowa中的实时log日志：

![0b97967f4737434fb5f89e5a96168908.png][]

flowb的启动参数

![c7a3d1f0d44b446992fe5f42e301fa8b.png][]

### prefect的配置管理和设置 ###

可以在控制台查看prefect的配置项：

> prefect config view

![4b34b74b0cfa435391b91b9d2f4fddd3.png][]

默认的prefect启动端口在4200，用户通过127.0.0.1:4200访问，可以通过配置设置新的启动端口：

> prefect config set PREFECT\_ORION\_API\_PORT=1234

指定prefect启动在1234端口。

![2cabc9a756464b17bb6ec3faed0d57e5.png][]

然后重启服务器即可：

prefect orion start

### prefect的API开发文档 ###

prefect的后端服务器也可以通过REST API访问调用。prefect的文档可以直接查看：  
http://127.0.0.1:4200/docs

![8011bc5f1c944b97b426722f30e100d0.png][]

http://127.0.0.1:4200/redoc

![7296986032c04714857ecbf337015f35.png][]

以上效果均在Windows环境下跑出来。prefect有更多的功能，值得研究和使用。

[fluidicon.png]: https://github.com/fluidicon.png
[GitHub - PrefectHQ_prefect_ The easiest way to coordinate your dataflowThe easiest way to coordinate your dataflow. Contribute to PrefectHQ_prefect development by creating an account on GitHub._fluidicon.png_https_github.com_PrefectHQ_prefect]: https://github.com/PrefectHQ/prefect
[750b7e424e6041db85b5d63384b54812.png]: https://img-blog.csdnimg.cn/750b7e424e6041db85b5d63384b54812.png
[b908628becce464fa8ef35a1891e48f8.png]: https://img-blog.csdnimg.cn/b908628becce464fa8ef35a1891e48f8.png
[d2361db243e74b588407615e360b08a2.png]: https://img-blog.csdnimg.cn/d2361db243e74b588407615e360b08a2.png
[0b97967f4737434fb5f89e5a96168908.png]: https://img-blog.csdnimg.cn/0b97967f4737434fb5f89e5a96168908.png
[c7a3d1f0d44b446992fe5f42e301fa8b.png]: https://img-blog.csdnimg.cn/c7a3d1f0d44b446992fe5f42e301fa8b.png
[4b34b74b0cfa435391b91b9d2f4fddd3.png]: https://img-blog.csdnimg.cn/4b34b74b0cfa435391b91b9d2f4fddd3.png
[2cabc9a756464b17bb6ec3faed0d57e5.png]: https://img-blog.csdnimg.cn/2cabc9a756464b17bb6ec3faed0d57e5.png
[8011bc5f1c944b97b426722f30e100d0.png]: https://img-blog.csdnimg.cn/8011bc5f1c944b97b426722f30e100d0.png
[7296986032c04714857ecbf337015f35.png]: https://img-blog.csdnimg.cn/7296986032c04714857ecbf337015f35.png