Golang简单而强大的DAG调度器和仪表板插件库goflow的使用

什么是Goflow

Goflow是一个用Go编写的简单但功能强大的DAG调度器和仪表板。它可以帮助你：

• 像Apache Airflow一样调度有向无环图(DAG)任务，但更简单
• 协调各种集群或服务的计算任务
• 提供监控仪表板
• 通过单个二进制或容器轻松部署
• 在小型虚拟机上运行，节省云成本
• 支持多种存储技术
• 使用代码而非配置文件定义DAG

快速开始

使用Docker

docker run -p 8181:8181 ghcr.io/fieldryand/goflow-example:latest

然后在浏览器中访问localhost:8181查看仪表板。

不使用Docker

1. 创建新项目并安装依赖：

go mod init # 创建新模块
go get github.com/fieldryand/goflow/v2 # 安装依赖

1. 创建main.go文件：

package main

import "github.com/fieldryand/goflow/v2"

func main() {
    options := goflow.Options{
        UIPath: "ui/",
        ShowExamples:  true,
        WithSeconds:  true,
    }
    gf := goflow.New(options)
    gf.Use(goflow.DefaultLogger())
    gf.Run(":8181")
}

1. 下载并解压仪表板：

wget https://github.com/fieldryand/goflow/releases/latest/download/goflow-ui.tar.gz
tar -xvzf goflow-ui.tar.gz
rm goflow-ui.tar.gz

1. 运行应用：

go run main.go

然后在浏览器中访问localhost:8181。

开发概述

基本概念

• Job: Goflow工作流称为Job，可以使用cron语法调度
• Task: 每个Job由一个或多个Task组成，构成依赖图
• Operator: 定义Task执行的工作，Goflow提供了一些基本Operator，也可以自定义
• Retries: 可以为Task设置重试策略
• Streaming: 使用服务器发送事件实时将状态推送到仪表板

创建Job示例

func myJob() *goflow.Job {
    j := &goflow.Job{Name: "my-job", Schedule: "* * * * *", Active: true}
    j.Add(&goflow.Task{
        Name:     "sleep-for-one-second",
        Operator: goflow.Command{Cmd: "sleep", Args: []string{"1"}},
    })
    return j
}

自定义Operator

type PositiveAddition struct{ a, b int }

func (o PositiveAddition) Run() (interface{}, error) {
    if o.a < 0 || o.b < 0 {
        return 0, errors.New("Can't add negative numbers")
    }
    result := o.a + o.b
    return result, nil
}

重试机制

func myJob() *goflow.Job {
    j := &goflow.Job{Name: "my-job", Schedule: "* * * * *"}
    j.Add(&goflow.Task{
        Name:       "sleep-for-one-second",
        Operator:   goflow.Command{Cmd: "sleep", Args: []string{"1"}},
        Retries:    5,
        RetryDelay: goflow.ConstantDelay{Period: 1},
    })
    return j
}

任务依赖

func myJob() *goflow.Job {
    j := &goflow.Job{Name: "my-job", Schedule: "* * * * *"}
    j.Add(&goflow.Task{
        Name:       "sleep-for-one-second",
        Operator:   goflow.Command{Cmd: "sleep", Args: []string{"1"}},
        Retries:    5,
        RetryDelay: goflow.ConstantDelay{Period: 1},
    })
    j.Add(&goflow.Task{
        Name:       "get-google",
        Operator:   goflow.Get{Client: &http.Client{}, URL: "https://www.google.com"},
    })
    j.Add(&goflow.Task{
        Name:       "add-two-plus-three",
        Operator:   PositiveAddition{a: 2, b: 3},
    })
    j.SetDownstream(j.Task("sleep-for-one-second"), j.Task("get-google"))
    j.SetDownstream(j.Task("sleep-for-one-second"), j.Task("add-two-plus-three"))
    return j
}

触发器规则

默认情况下，Task的触发器规则是allSuccessful，即所有上游任务成功后才执行。可以设置为allDone让任务在所有上游完成后执行：

j.Add(&goflow.Task{
    Name:        "sleep-for-one-second",
    Operator:    goflow.Command{Cmd: "sleep", Args: []string{"1"}},
    Retries:     5,
    RetryDelay:  goflow.ConstantDelay{Period: 1},
    TriggerRule: "allDone",
})

运行Goflow引擎

func main() {
    gf := goflow.New(goflow.Options{Streaming: true})
    gf.AddJob(myJob)
    gf.Use(goflow.DefaultLogger())
    gf.Run(":8181")
}

存储

Goflow支持多种键值存储，默认使用内存数据库，也可以使用Redis等：

package main

import "github.com/fieldryand/goflow/v2"
import "github.com/philippgille/gokv/redis"

func main() {
    // 创建存储客户端
    client, err := redis.NewClient(redis.DefaultOptions)
    if err != nil {
        panic(err)
    }
    defer client.Close()

    // 作为Goflow选项传入
    options := goflow.Options{
        Store: client,
        UIPath: "ui/",
        Streaming: true,
        ShowExamples:  true,
    }
    gf := goflow.New(options)
    gf.Use(goflow.DefaultLogger())
    gf.Run(":8181")
}

API和集成

Goflow提供API端点用于集成：

• GET /api/health: 检查服务健康状态
• GET /api/jobs: 列出注册的Job
• GET /api/jobs/{jobname}: 获取特定Job详情
• GET /api/executions: 查询和列出Job执行
• POST /api/jobs/{jobname}/submit: 提交Job执行
• POST /api/jobs/{jobname}/toggle: 切换Job调度开关
• /stream: 返回服务器发送事件，用于实时更新

可以使用Swagger查看完整的API文档：

docker run -p 8080:8080 -e SWAGGER_JSON=/app/swagger.json -v $(pwd):/app swaggerapi/swagger-ui

Goflow是一个功能强大但简单的任务调度系统，特别适合需要轻量级解决方案的场景。通过代码定义DAG使得复杂工作流更易于管理，而内置的仪表板提供了良好的可视化和监控能力。