Bamboo-pipeline 是藍(lán)鯨智云旗下SaaS標(biāo)準(zhǔn)運(yùn)維的流程編排引擎。其具備以下特點(diǎn)：

1. 多種流程模式 ：支持串行、并行，支持子流程，可以根據(jù)全局參數(shù)自動(dòng)選擇分支執(zhí)行，節(jié)點(diǎn)失敗處理機(jī)制可配置。
2. 參數(shù)引擎 ：支持參數(shù)共享，支持參數(shù)替換。
3. 可交互的任務(wù)執(zhí)行 ：任務(wù)執(zhí)行中可以隨時(shí)暫停、繼續(xù)、撤銷，節(jié)點(diǎn)失敗后可以重試、跳過。

1.準(zhǔn)備

開始之前，你要確保Python和pip已經(jīng)成功安裝在電腦上，如果沒有，可以訪問這篇文章：超詳細(xì)Python安裝指南 進(jìn)行安裝。

(可選1) 如果你用Python的目的是數(shù)據(jù)分析，可以直接安裝Anaconda：Python數(shù)據(jù)分析與挖掘好幫手—Anaconda，它內(nèi)置了Python和pip.

(可選2) 此外，推薦大家用VSCode編輯器，它有許多的優(yōu)點(diǎn)：Python 編程的最好搭檔—VSCode 詳細(xì)指南。

請(qǐng)選擇以下任一種方式輸入命令安裝依賴 ：

1. Windows 環(huán)境 打開 Cmd (開始-運(yùn)行-CMD)。
2. MacOS 環(huán)境 打開 Terminal (command+空格輸入Terminal)。
3. 如果你用的是 VSCode編輯器 或 Pycharm，可以直接使用界面下方的Terminal.

pip install bamboo-engine
pip install bamboo-pipeline
pip install django
pip install celery

2. 項(xiàng)目初始化

（選項(xiàng)一：無Django項(xiàng)目） 如果你沒有任何的現(xiàn)成Django項(xiàng)目，請(qǐng)按下面的流程初始化

由于 ** bamboo-pipeline ** 運(yùn)行時(shí)基于 Django 實(shí)現(xiàn)，所以需要新建一個(gè) Django 項(xiàng)目：

django-admin startproject easy_pipeline
cd easy_pipeline

在 ** easy_pipeline.settings.py ** 下添加如下配置：

from pipeline.eri.celery.queues import *
from celery import Celery

app = Celery("proj")

app.config_from_object("django.conf:settings")

INSTALLED_APPS = [
    ...
    "pipeline",
    "pipeline.engine",
    "pipeline.component_framework",
    "pipeline.eri",
    ...
]

在 ** easy_pipeline **目錄下初始化數(shù)據(jù)庫：

python manage.py migrate

（選項(xiàng)二：有Django項(xiàng)目需要使用流程引擎） 如果你有現(xiàn)成的PipeLine項(xiàng)目需要使用此流程引擎，請(qǐng)?jiān)陧?xiàng)目的** settings.py **下添加如下配置：

from pipeline.eri.celery.queues import *
from celery import Celery

app = Celery("proj")

app.config_from_object("django.conf:settings")

INSTALLED_APPS = [
    ...
    "pipeline",
    "pipeline.engine",
    "pipeline.component_framework",
    "pipeline.eri",
    ...
]

然后重新執(zhí)行migrate，生成pipeline相關(guān)的流程模型：

python manage.py migrate

migrate 執(zhí)行完畢后會(huì)如下圖所示：

由于是在原有項(xiàng)目上使用流程引擎，可能會(huì)遇到一些版本不匹配的問題，如果遇到報(bào)錯(cuò)，請(qǐng)排查解決或到藍(lán)鯨官網(wǎng)上進(jìn)行詢問。

3. 簡(jiǎn)單的流程例子

首先在項(xiàng)目目錄下啟動(dòng) celery worker：

python manage.py celery worker -Q er_execute,er_schedule --pool=solo -l info

啟動(dòng)成功類似下圖所示：

（注意） 如果你是在你的原有Django項(xiàng)目上做改造，它并不一定能夠順利地啟動(dòng)成功，這是因?yàn)镻ipeline使用了 Django 2.2.24，會(huì)存在許多版本不兼容的情況。如果遇到報(bào)錯(cuò)，請(qǐng)排查解決或到藍(lán)鯨官網(wǎng)上進(jìn)行詢問。

在下面的例子中，我們將會(huì)創(chuàng)建并執(zhí)行一個(gè)簡(jiǎn)單的流程：

3.1 創(chuàng)建流程APP

在 bamboo_pipeline 中，一個(gè)流程由多個(gè)組件組成，官方推薦使用APP統(tǒng)一管控組件：

python manage.py create_plugins_app big_calculator

該命令會(huì)在 Django 工程根目錄下生成擁有以下目錄結(jié)構(gòu)的 APP：

big_calculator
├── __init__.py
├── components
│   ├── __init__.py
│   └── collections
│   ├── __init__.py
│   └── plugins.py
├── migrations
│   └── __init__.py
└── static
    └── big_calculator
        └── plugins.js

別忘了把新創(chuàng)建的這個(gè)插件添加到 Django 配置的 **INSTALLED_APPS **中：

INSTALLED_APPS = (
    ...
    'big_calculator',
    ...
)

3.2 編寫流程的Service原子

組件服務(wù) ** Service是組件的核心，Service ** 定義了組件被調(diào)用時(shí)執(zhí)行的邏輯，下面讓我們實(shí)現(xiàn)一個(gè)計(jì)算傳入的參數(shù)n的階乘，并把結(jié)果寫到輸出中的 ** Service ** ，在 **big_calculator/components/collections/plugins.py ** 中輸入以下代碼：

import math
from pipeline.core.flow.activity import Service


class FactorialCalculateService(Service):

    def execute(self, data, parent_data):
        """
        組件被調(diào)用時(shí)的執(zhí)行邏輯
        :param data: 當(dāng)前節(jié)點(diǎn)的數(shù)據(jù)對(duì)象
        :param parent_data: 該節(jié)點(diǎn)所屬流程的數(shù)據(jù)對(duì)象
        :return:
        """
        n = data.get_one_of_inputs('n')
        if not isinstance(n, int):
            data.outputs.ex_data = 'n must be a integer!'
            return False

        data.outputs.factorial_of_n = math.factorial(n)
        return True

    def inputs_format(self):
        """
        組件所需的輸入字段，每個(gè)字段都包含字段名、字段鍵、字段類型及是否必填的說明。
        :return:必須返回一個(gè) InputItem 的數(shù)組，返回的這些信息能夠用于確認(rèn)該組件需要獲取什么樣的輸入數(shù)據(jù)。
        """
        return [
            Service.InputItem(name='integer n', key='n', type='int', required=True)
        ]

    def outputs_format(self):
        """
        組件執(zhí)行成功時(shí)輸出的字段，每個(gè)字段都包含字段名、字段鍵及字段類型的說明
        :return: 必須返回一個(gè) OutputItem 的數(shù)組, 便于在流程上下文或后續(xù)節(jié)點(diǎn)中進(jìn)行引用
        """
        return [
            Service.OutputItem(name='factorial of n', key='factorial_of_n', type='int')
        ]

首先我們繼承了 ** Service基類，并實(shí)現(xiàn)了execute() ** 和 **outputs_format() ** 這兩個(gè)方法，他們的作用如下：

1. execute ：組件被調(diào)用時(shí)執(zhí)行的邏輯。接收 data 和 parent_data 兩個(gè)參數(shù)。
   其中，data 是當(dāng)前節(jié)點(diǎn)的數(shù)據(jù)對(duì)象，這個(gè)數(shù)據(jù)對(duì)象存儲(chǔ)了用戶傳遞給當(dāng)前節(jié)點(diǎn)的參數(shù)的值以及當(dāng)前節(jié)點(diǎn)輸出的值。parent_data 則是該節(jié)點(diǎn)所屬流程的數(shù)據(jù)對(duì)象，通常會(huì)將一些全局使用的常量存儲(chǔ)在該對(duì)象中，如當(dāng)前流程的執(zhí)行者、流程的開始時(shí)間等。
2. outputs_format ：組件執(zhí)行成功時(shí)輸出的字段，每個(gè)字段都包含字段名、字段鍵及字段類型的說明。這個(gè)方法必須返回一個(gè) OutputItem 的數(shù)組，返回的這些信息能夠用于確認(rèn)某個(gè)組件在執(zhí)行成功時(shí)輸出的數(shù)據(jù)，便于在流程上下文或后續(xù)節(jié)點(diǎn)中進(jìn)行引用。
3. inputs_format ：組件所需的輸入字段，每個(gè)字段都包含字段名、字段鍵、字段類型及是否必填的說明。這個(gè)方法必須返回一個(gè) InputItem 的數(shù)組，返回的這些信息能夠用于確認(rèn)某個(gè)組件需要獲取什么樣的輸入數(shù)據(jù)。

下面我們來看一下 **execute() **方法內(nèi)部執(zhí)行的邏輯，首先我們嘗試從當(dāng)前節(jié)點(diǎn)數(shù)據(jù)對(duì)象的輸出中獲取輸入?yún)?shù)n，如果獲取到的參數(shù)不是一個(gè) int 實(shí)例，那么我們會(huì)將異常信息寫入到當(dāng)前節(jié)點(diǎn)輸出的 **ex_data **字段中， 這個(gè)字段是引擎內(nèi)部的保留字段，節(jié)點(diǎn)執(zhí)行失敗時(shí)產(chǎn)生的異常信息都應(yīng)該寫入到該字段中 。隨后我們返回 ** False ** 代表組件本次執(zhí)行失敗，隨后節(jié)點(diǎn)會(huì)進(jìn)入失敗狀態(tài)：

n = data.get_one_of_inputs('n')
if not isinstance(n, int):
    data.outputs.ex_data = 'n must be a integer!'
    return False

若獲取到的 n 是一個(gè)正常的 int，我們就調(diào)用 **math.factorial() **函數(shù)來計(jì)算 n 的階乘，計(jì)算完成后，我們會(huì)將結(jié)果寫入到輸出的 factorial_of_n 字段中，以供流程中的其他節(jié)點(diǎn)使用：

data.outputs.factorial_of_n = math.factorial(n)
return True

3.3 編寫流程組件，綁定Service原子

完成 Service 的編寫后，我們需要將其與一個(gè) Component 綁定起來，才能夠注冊(cè)到組件庫中，在**big_calculatorcomponents__init__.py **文件下添加如下的代碼：

import logging
from pipeline.component_framework.component import Component
from big_calculator.components.collections.plugins import FactorialCalculateService

logger = logging.getLogger('celery')


class FactorialCalculateComponent(Component):
    name = 'FactorialCalculateComponent'
    code = 'fac_cal_comp'
    bound_service = FactorialCalculateService

我們定義了一個(gè)繼承自基類 **Component **的類 FactorialCalculateComponent ，他擁有以下屬性：

1.name：組件名。
2.code：組件代碼，這個(gè)代碼必須是全局唯一的。
3.bound_service：與該組件綁定的 Service。

這樣一來，我們就完成了一個(gè)流程原子的開發(fā)。

3.4 生成流程，測(cè)試剛編寫的組件

在 big_calculatortest.py 寫入以下內(nèi)容，生成一個(gè)流程，測(cè)試剛剛編寫的組件：

# Python 實(shí)用寶典
# 2021/06/20

import time

from bamboo_engine.builder import *
from big_calculator.components import FactorialCalculateComponent
from pipeline.eri.runtime import BambooDjangoRuntime
from bamboo_engine import api
from bamboo_engine import builder


def bamboo_playground():
    """
    測(cè)試流程引擎
    """
    # 使用 builder 構(gòu)造出流程描述結(jié)構(gòu)
    start = EmptyStartEvent()
    # 這里使用 我們剛創(chuàng)建好的n階乘組件
    act = ServiceActivity(component_code=FactorialCalculateComponent.code)
    # 傳入?yún)?shù)
    act.component.inputs.n = Var(type=Var.PLAIN, value=4)
    end = EmptyEndEvent()

    start.extend(act).extend(end)

    pipeline = builder.build_tree(start)
    api.run_pipeline(runtime=BambooDjangoRuntime(), pipeline=pipeline)

    # 等待 1s 后獲取流程執(zhí)行結(jié)果
    time.sleep(1)

    result = api.get_execution_data_outputs(BambooDjangoRuntime(), act.id).data

    print(result)

隨后，在命令行輸入：

python manage.py shell

打開 django console, 輸入以下命令，執(zhí)行此流程：

from big_calculator.test import bamboo_playground
bamboo_playground()

流程運(yùn)行完后，獲取節(jié)點(diǎn)的執(zhí)行結(jié)果，可以看到，該節(jié)點(diǎn)輸出了 factorial_of_n，并且值為 24(4 * 3 * 2 *1)，這正是我們需要的效果：

{'_loop': 0, '_result': True, 'factorial_of_n': 24}

恭喜你，你已經(jīng)成功的創(chuàng)建了一個(gè)流程并把它運(yùn)行起來了！在這期間你可能會(huì)遇到不少的坑，建議嘗試先自行解決，如果實(shí)在無法解決，可以前往 標(biāo)準(zhǔn)運(yùn)維 倉庫提 issues，或者前往藍(lán)鯨智云官網(wǎng)提問。