Flowmium

Flowmium 是一个使用 Kubernetes 的工作流编排器。您可以定义和运行一个 YAML 工作流 或运行一个 Python 工作流，其中每个函数都作为一个 Kubernetes pod 运行。

Python 工作流示例

from flowmium import Flow, FlowContext
from flowmium.serializers import plain_text, json_text, pkl


flow = Flow("testing")


@flow.task(serializer=json_text)
def foo() -> str:
    return "Hallo world"


@flow.task({"input_str": foo}, serializer=plain_text)
def replace_letter_a(input_str: str, flowctx: FlowContext) -> str:
    return input_str.replace("a", "e") + str(flowctx.task_id)


@flow.task({"input_str": foo}, serializer=pkl)
def replace_letter_t(input_str: str) -> str:
    return input_str.replace("t", "d")


@flow.task(
    {"first": replace_letter_t, "second": replace_letter_a}, serializer=plain_text
)
def concat(first: str, second: str) -> str:
    return f"{first} {second}"


if __name__ == "__main__":
    flow.run()

入门

• 在本地设置以进行测试
• 在生产环境中部署
• 示例 Python 包工作流
• 示例 Python 脚本工作流
• 示例 YAML 定义工作流
• Python 框架文档
• API 文档
• Rust 客户端文档
• 集成到现有的 Rust 项目中

flowctl 命令行界面

flowctl 命令行界面用于监视工作流当前状态、提交新工作流和下载工件。

安装

cargo install flowmium

用法

注意

密钥存储在服务器上，可以引用在 YAML 定义或 Python 工作流中设置环境变量值。这样您就不必将密钥提交到您的仓库。但是，它们不使用 Kubernetes 密钥，当工作流任务作为 Job 部署时，它们被设置为正常的环境变量。

YAML 工作流定义模式

YAML流程定义参考。请参阅示例。

根

任务

环境

输入

输出

从源运行

从源运行python流程示例

这些说明将允许您从本地源运行示例python流程（framework/tests/example_flow.py），无需从上游提取（包括执行器）。使用此功能来验证您的更改。说明假定您位于存储库的根目录。

• 安装sqlx CLI

  cargo install sqlx-cli
  

• 在本地运行测试Kubernetes集群、minio和容器注册库

  cd flowmium/
  make up
  

• 监视本地集群中运行的Pod

  cd flowmium/
  make watch
  

• 运行迁移

  cd flowmium/
  sqlx migrate run
  

• 从存储库的根目录运行flowmium服务器

  cd flowmium/
  export FLOWMIUM_POSTGRES_URL='postgres://flowmium:flowmium@localhost/flowmium'
  export FLOWMIUM_STORE_URL='https://127.0.0.1:9000'
  export FLOWMIUM_TASK_STORE_URL='http://172.16.238.4:9000'
  export FLOWMIUM_BUCKET_NAME='flowmium-test'
  export FLOWMIUM_ACCESS_KEY='minio'
  export FLOWMIUM_SECRET_KEY='password'
  export FLOWMIUM_INIT_CONTAINER_IMAGE='docker.io/shnoo28/flowmium:latest'
  export FLOWMIUM_NAMESPACE=default
  export KUBECONFIG=./kubeconfig.yaml
  cargo run --bin flowmium -- server --port 8080
  

• 使用flowctl监视流程状态

  cd flowmium/
  cargo build
  watch ./target/debug/flowctl list
  

• 构建并推送示例python流程（注意：如果您正在运行第二次测试，则可能需要使用不同的镜像名称，或者修剪您的机器上的Docker镜像）

  cd framework/
  docker build . -t py-flow-test
  docker tag py-flow-test localhost:5180/py-flow-test:latest
  docker push localhost:5180/py-flow-test:latest
  

• 将流程提交给执行器服务器

  python3 -m tests --image registry:5000/py-flow-test:latest --cmd 'python3 -m tests' --flowmium-server http://localhost:8080
  

运行端到端测试

• 对于使用上游init容器的端到端测试

  make test
  

• 对于使用源init容器的端到端测试

  FLOWMIUM_INIT_CONTAINER_IMAGE_FROM_SOURCE=true make test
  

运行python框架的单元测试

从framework/路径运行make test