ジョブフロー図の書き方｜ジョブ一覧と先行関係をMermaidで描く例

ジョブフロー図とは、複数のジョブ（まとめて実行する処理の単位）を箱で、ジョブ間の「先に終わっている必要がある」という先行関係を矢印で表した図です。 先にジョブ一覧表で先行ジョブを決め、その表から図を起こす順番にすると、表と図の食い違いがなくなります。

この記事では、製品に依存しないジョブ一覧のテンプレート（タブ区切り）、その表から Mermaid のフローチャートを生成する Python スクリプト、直列・並列・合流・分岐・待ち合わせ・異常系の6パターンの Mermaid コード、循環依存の検出と最長経路（全体の所要時間）の計算をまとめました。ジョブ1本の中身の設計ではなく、ジョブ間の順序と依存関係の表し方に絞って扱います。題材は架空の「社内の備品予約システム」の夜間処理です。

確認環境：Python 3.11.15 でスクリプトを実行し、掲載した Mermaid のコードはすべて @mermaid-js/mermaid-cli 12.0.0（Mermaid 12.1.0、Chromium）で SVG に描画できることを確かめています。

ジョブフロー図の書き方の早見表

ジョブフロー図で表すものと用語

ジョブフロー図が答える問いは、「どのジョブが終われば、次のどのジョブを動かしてよいか」「途中で止まったら、どこまでが影響を受けるか」「全体は何時に終わるか」の3つです。処理の中身（どのテーブルをどう更新するか）は、ジョブごとの設計書に書きます。

ジョブ一覧のテンプレート（TSV）

ジョブ一覧では、後続ジョブではなく先行ジョブを書きます。 「このジョブを動かすには何が終わっている必要があるか」は、そのジョブの担当者が答えられるからです。後続は先行の裏返しなので、表から機械的に求められます。

先行ジョブに書くのは直前のジョブだけです。JOB010 → JOB040 → JOB050 の順なら、JOB050 の先行ジョブは JOB040 だけで足ります。JOB010 まで書くと矢印が増え、図が読みにくくなります。

空のひな形

表計算ソフトのA1セルに貼り付けると、列に分かれます。

ジョブID   ジョブ名    起動条件    先行ジョブ   所要時間(分) 異常時の扱い  再実行可否
JOB010                      
JOB020                      
JOB030                      

記入例（jobs.tsv）

ジョブID   ジョブ名    起動条件    先行ジョブ   所要時間(分) 異常時の扱い  再実行可否
JOB010  予約データ抽出 毎日 01:00        10  後続を止める  可(先頭から)
JOB020  利用者マスタ取込    毎日 01:00        5   後続を止める  可(先頭から)
JOB030  備品マスタ取込 毎日 01:00        8   後続を止める  可(先頭から)
JOB040  予約集計    先行の正常終了 JOB010,JOB020,JOB030    20  後続を止める  可(対象日を消して再集計)
JOB050  利用状況レポート作成  先行の正常終了 JOB040  15  後続を止める  可(上書き)
JOB060  返却期限の通知メール送信    先行の正常終了 JOB040  5   警告を出して続ける   不可(送信済みを除いて再送)
JOB070  棚卸ファイル到着待ち  毎日 01:00 / 02:00まで待つ        30  警告を出して続ける   可(待ち直し)
JOB080  棚卸データ取込 先行の正常終了 JOB070  12  後続を止める  可(取込済みを消して再取込)
JOB090  夜間バックアップ    先行がすべて終了    JOB050,JOB060,JOB080    30  後続を止める  可(先頭から)

9本のジョブのうち、JOB010・JOB020・JOB030・JOB070 が 01:00 に同時に起動し、JOB040 が3本の終了を待ち合わせ、JOB090 が最後に全体を受けます。

異常時の扱いの書き分け

JOB090 の起動条件を「先行がすべて終了」としているのは、JOB060 が警告で終わっても、バックアップは行うためです。「正常終了を待つ」のか「終了を待つ」のかを起動条件に書き分けると、異常時の動きが一覧から読み取れます。

ジョブ一覧からMermaidのフローチャートを生成するスクリプト

タブ区切りのジョブ一覧を読み、検査・図の生成・所要時間の計算を行うスクリプトです。標準ライブラリだけで動きます。

"""ジョブ一覧(TSV)を検査し、Mermaid のフローチャートと所要時間の表を出力する"""
import csv
import sys
from graphlib import CycleError, TopologicalSorter


def load(path):
    with open(path, encoding="utf-8", newline="") as f:
        rows = list(csv.DictReader(f, delimiter="\t"))
    jobs = {}
    for r in rows:
        jobs[r["ジョブID"]] = {
            "name": r["ジョブ名"],
            "preds": [p.strip() for p in r["先行ジョブ"].split(",") if p.strip()],
            "minutes": int(r["所要時間(分)"]),
            "on_error": r["異常時の扱い"],
        }
    return jobs


def check(jobs):
    """先行ジョブの書き間違いと循環依存を調べ、実行できる順番を返す"""
    errors = []
    for job_id, job in jobs.items():
        for p in job["preds"]:
            if p not in jobs:
                errors.append(f"{job_id}: 先行ジョブ {p} が一覧にありません")
            if p == job_id:
                errors.append(f"{job_id}: 自分自身を先行ジョブにしています")
    if errors:
        return None, errors
    try:
        order = list(TopologicalSorter({k: v["preds"] for k, v in jobs.items()}).static_order())
    except CycleError as e:
        return None, ["循環依存があります: " + " -> ".join(e.args[1])]
    return order, []


def schedule(jobs, order):
    """最早開始・最早終了・余裕と、最長経路(全体の所要時間を決める経路)を求める"""
    es, ef, via = {}, {}, {}
    for j in order:
        preds = jobs[j]["preds"]
        via[j] = max(preds, key=lambda p: ef[p]) if preds else None
        es[j] = ef[via[j]] if preds else 0
        ef[j] = es[j] + jobs[j]["minutes"]
    total = max(ef.values())
    succs = {j: [k for k in jobs if j in jobs[k]["preds"]] for j in jobs}
    lf = {}
    for j in reversed(order):
        lf[j] = min((lf[s] - jobs[s]["minutes"] for s in succs[j]), default=total)
    slack = {j: lf[j] - ef[j] for j in jobs}
    path, cur = [], max(ef, key=ef.get)
    while cur:
        path.append(cur)
        cur = via[cur]
    return es, ef, slack, total, list(reversed(path))


def to_mermaid(jobs, critical=()):
    lines = ["flowchart LR"]
    for job_id, job in jobs.items():
        lines.append(f'    {job_id}["{job_id}<br/>{job["name"]}<br/>{job["minutes"]}分"]')
    for job_id, job in jobs.items():
        for p in job["preds"]:
            arrow = "==>" if p in critical and job_id in critical else "-->"
            lines.append(f"    {p} {arrow} {job_id}")
    cont = [j for j, v in jobs.items() if "続ける" in v["on_error"]]
    lines.append("    classDef critical stroke-width:3px")
    lines.append("    classDef cont stroke-dasharray:5 5")
    if critical:
        lines.append("    class " + ",".join(critical) + " critical")
    if cont:
        lines.append("    class " + ",".join(cont) + " cont")
    return "\n".join(lines)


def hhmm(base, minutes):
    h, m = divmod(base + minutes, 60)
    return f"{h:02d}:{m:02d}"


def main(path, start="01:00"):
    jobs = load(path)
    order, errors = check(jobs)
    if errors:
        for e in errors:
            print("NG:", e)
        return 1
    es, ef, slack, total, path_ = schedule(jobs, order)
    print(to_mermaid(jobs, path_))
    print()
    base = int(start[:2]) * 60 + int(start[3:])
    print("実行順:", " ".join(order))
    print("ジョブID  開始   終了   所要  余裕")
    for j in order:
        print(f"{j}   {hhmm(base, es[j])}  {hhmm(base, ef[j])}  {jobs[j]['minutes']:>3}分 {slack[j]:>3}分")
    print(f"最長経路: {' -> '.join(path_)} = {total}分 (終了見込み {hhmm(base, total)})")
    return 0


if __name__ == "__main__":
    sys.exit(main(*sys.argv[1:]))

実行します（python3 jobflow.py jobs.tsv）。前半が Mermaid のコード、後半が所要時間の表です。

flowchart LR
    JOB010["JOB010<br/>予約データ抽出<br/>10分"]
    JOB020["JOB020<br/>利用者マスタ取込<br/>5分"]
    JOB030["JOB030<br/>備品マスタ取込<br/>8分"]
    JOB040["JOB040<br/>予約集計<br/>20分"]
    JOB050["JOB050<br/>利用状況レポート作成<br/>15分"]
    JOB060["JOB060<br/>返却期限の通知メール送信<br/>5分"]
    JOB070["JOB070<br/>棚卸ファイル到着待ち<br/>30分"]
    JOB080["JOB080<br/>棚卸データ取込<br/>12分"]
    JOB090["JOB090<br/>夜間バックアップ<br/>30分"]
    JOB010 ==> JOB040
    JOB020 --> JOB040
    JOB030 --> JOB040
    JOB040 ==> JOB050
    JOB040 --> JOB060
    JOB070 --> JOB080
    JOB050 ==> JOB090
    JOB060 --> JOB090
    JOB080 --> JOB090
    classDef critical stroke-width:3px
    classDef cont stroke-dasharray:5 5
    class JOB010,JOB040,JOB050,JOB090 critical
    class JOB060,JOB070 cont

実行順: JOB010 JOB020 JOB030 JOB070 JOB040 JOB080 JOB050 JOB060 JOB090
ジョブID  開始   終了   所要  余裕
JOB010   01:00  01:10   10分   0分
JOB020   01:00  01:05    5分   5分
JOB030   01:00  01:08    8分   2分
JOB070   01:00  01:30   30分   3分
JOB040   01:10  01:30   20分   0分
JOB080   01:30  01:42   12分   3分
JOB050   01:30  01:45   15分   0分
JOB060   01:30  01:35    5分  10分
JOB090   01:45  02:15   30分   0分
最長経路: JOB010 -> JOB040 -> JOB050 -> JOB090 = 75分 (終了見込み 02:15)

生成されたコードは、太い矢印と太い枠が最長経路、点線の枠が「警告を出して続ける」ジョブを表します。Mermaid に対応したエディタや文書ツールに貼ると、そのまま図になります。

スクリプトの要点

graphlib は Python 3.9 で追加された標準ライブラリで、依存関係から実行できる順番（トポロジカル順）を求められます。循環があると CycleError が出て、循環しているノードの並びを受け取れるので、それを検査に使っています（Python 公式ドキュメント graphlib の記載）。

先行関係の6パターン（Mermaidコード）

ジョブフロー図は、次の6つのパターンの組み合わせで描けます。

1. 直列

flowchart LR
    A["JOB010<br/>抽出"] --> B["JOB020<br/>変換"] --> C["JOB030<br/>取込"]

前のジョブの出力を次のジョブが使う場合の基本形です。途中で止まると、それより後ろはすべて起動しません。

2. 並列

flowchart LR
    S(["01:00 起動"]) --> A["JOB010<br/>予約データ抽出"]
    S --> B["JOB020<br/>利用者マスタ取込"]
    S --> C["JOB030<br/>備品マスタ取込"]

互いの結果を使わないジョブは並列にすると、全体の所要時間が「合計」から「いちばん長い1本」に縮みます。同じテーブルを更新するジョブどうしは、並列にせず直列にします。

3. 合流

flowchart LR
    A["JOB010<br/>予約データ抽出"] --> D["JOB040<br/>予約集計"]
    B["JOB020<br/>利用者マスタ取込"] --> D
    C["JOB030<br/>備品マスタ取込"] --> D

JOB040 は3本すべての正常終了を待ちます。1本でも異常終了すると起動しないので、合流点はそのまま「ここまでがそろったか」の確認点になります。

4. 分岐

flowchart LR
    A["JOB100<br/>営業日判定"] --> J{"終了コード"}
    J -- "0: 営業日" --> B["JOB110<br/>日次集計"]
    J -- "1: 月末営業日" --> C["JOB120<br/>月次集計"]
    J -- "2: 休日" --> E(["何もせず終了"])
    C --> B

判定用のジョブの終了コードで後続を切り替えます。分岐を図に描くときは、どの終了コードがどの線に対応するかを線のラベルに書きます。この例では、月末営業日は月次集計の後に日次集計も行います。

5. 待ち合わせ

flowchart LR
    S(["01:00 起動"]) --> W{{"JOB070<br/>棚卸ファイル到着待ち<br/>02:00 まで"}}
    W -- "到着" --> A["JOB080<br/>棚卸データ取込"]
    W -. "02:00 を過ぎた" .-> N["JOB910<br/>未着の通知"]
    A --> T{{"06:00 まで待つ"}}
    T --> B["JOB200<br/>オンライン開始"]

ファイルの到着待ちや時刻待ちは、六角形で描いて通常のジョブと区別します。待ち合わせにはいつまで待つかと時間切れのときの流れを必ず書き添えます。

6. 異常系

flowchart LR
    A["JOB040<br/>予約集計"] --> B["JOB050<br/>利用状況レポート作成"] --> C["JOB090<br/>夜間バックアップ"]
    A -. "異常終了" .-> X["JOB900<br/>異常通知"]
    B -. "異常終了" .-> X
    X --> H(["後続を保留して<br/>担当者の判断を待つ"])
    A --> M["JOB060<br/>通知メール送信"]
    M -. "異常終了(警告)" .-> L["JOB905<br/>警告を記録"]
    L --> C
    M --> C
    classDef err stroke-dasharray:5 5
    class X,L,H err

異常時の流れは点線で描き、正常時の流れと見分けられるようにします。上の図は、JOB040・JOB050 は「後続を止める」、JOB060 は「警告を出して続ける」の例です。異常系の線をすべて描くと図が込み入るので、全体図には正常系だけを描き、異常系は一覧の「異常時の扱い」列と、要所だけを抜き出した別の図で示すと読みやすくなります。

循環依存の検出

先行ジョブをたどって元のジョブに戻ってくる状態を循環依存といいます。循環があると、どのジョブも起動条件を満たせず、最初の1本を始められません。ジョブが増えて一覧を何人かで編集すると入り込みやすいので、図にする前に検査します。

jobs.tsv の JOB010 の先行ジョブに JOB090 を書き加えたファイル（jobs_cycle.tsv）で実行した結果です。

$ python3 jobflow.py jobs_cycle.tsv
NG: 循環依存があります: JOB010 -> JOB040 -> JOB050 -> JOB090 -> JOB010

JOB090 の先行ジョブ JOB060 を JOB066 と書き間違えたファイル（jobs_typo.tsv）では、次のようになります。

$ python3 jobflow.py jobs_typo.tsv
NG: JOB090: 先行ジョブ JOB066 が一覧にありません

どちらの場合も終了コードは1で、図は出力されません。検査を通ったものだけが図になるので、図と一覧が食い違うことがなくなります。

最長経路で全体の所要時間を求める

並列に動くジョブがあると、全体の所要時間は単純な合計にはなりません。始点から終点までのすべての経路のうち、所要時間の合計がいちばん長い経路（最長経路）が、全体の所要時間を決めます。

計算のしかた

スクリプトの出力をもう一度見ます。

実行順: JOB010 JOB020 JOB030 JOB070 JOB040 JOB080 JOB050 JOB060 JOB090
ジョブID  開始   終了   所要  余裕
JOB010   01:00  01:10   10分   0分
JOB020   01:00  01:05    5分   5分
JOB030   01:00  01:08    8分   2分
JOB070   01:00  01:30   30分   3分
JOB040   01:10  01:30   20分   0分
JOB080   01:30  01:42   12分   3分
JOB050   01:30  01:45   15分   0分
JOB060   01:30  01:35    5分  10分
JOB090   01:45  02:15   30分   0分
最長経路: JOB010 -> JOB040 -> JOB050 -> JOB090 = 75分 (終了見込み 02:15)

検算：すべての経路を書き出して比べる

9本のジョブには、始点から JOB090 までの経路が7本あります。それぞれの合計を出すと、次のとおりです。

いちばん長いのは75分で、スクリプトの結果と一致します。余裕も確かめられます。JOB070 を通る経路は72分なので、75分との差の3分が JOB070 と JOB080 の余裕です。JOB030 は73分の経路にあるので余裕は2分、JOB020 は70分で5分、JOB060 は65分で10分となり、表の値と合っています。

9本の合計は 10＋5＋8＋20＋15＋5＋30＋12＋30＝135分ですが、並列にしているので全体は75分で終わります。

所要時間が変わったときの確かめ方

余裕が小さいジョブは、少し延びるだけで最長経路が入れ替わります。JOB070 の到着待ちを30分から45分に変えたファイルで実行すると、次のようになります。

$ python3 jobflow.py jobs_late.tsv
…（省略）…
JOB060   01:30  01:35    5分  22分
JOB090   01:57  02:27   30分   0分
最長経路: JOB070 -> JOB080 -> JOB090 = 87分 (終了見込み 02:27)

45＋12＋30＝87分で、最長経路が棚卸ファイルの側に移り、終了見込みが12分延びました。ファイルの到着待ちのように自分たちで短くできない時間が最長経路に入ると、ほかのジョブを速くしても全体は縮みません。 こうしたジョブは、後続の本数を減らす、ほかの処理と切り離して別のジョブネットにする、といった構成の見直しで対応します。

読みやすいジョブフロー図にするための決まり

ジョブが多い場合は、subgraph でジョブネットごとに囲むと、まとまりが見えます。

flowchart LR
    subgraph NET1["取込ネット 01:00"]
        A["JOB010<br/>予約データ抽出"]
        B["JOB020<br/>利用者マスタ取込"]
        C["JOB030<br/>備品マスタ取込"]
    end
    subgraph NET2["集計ネット"]
        D["JOB040<br/>予約集計"] --> E["JOB050<br/>利用状況レポート作成"]
    end
    A --> D
    B --> D
    C --> D
    E --> F["JOB090<br/>夜間バックアップ"]

レビュー用チェックリスト

よくある質問

Q. ジョブフロー図とフローチャートは何が違いますか？ A. フローチャートは1つの処理の中の手順と条件分岐を表し、ジョブフロー図は複数のジョブの間の順序と依存関係を表します。箱1つがジョブ1本に当たり、箱の中の処理は描きません。この記事では、描画に Mermaid のフローチャートの記法を使っています。

Q. ジョブ管理の製品を使う場合も、この一覧は役に立ちますか？ A. 役に立ちます。ジョブID・起動条件・先行ジョブ・異常時の扱い・再実行可否は、製品が変わっても決める必要がある項目です。先に一覧で決めておくと、製品の画面や定義ファイルへの登録は書き写す作業になります。

Q. 所要時間がまだわからないジョブはどう書けばよいですか？ A. 見込みの値を入れて計算し、測れたら置き換えます。件数に比例して延びるジョブは、件数が多い日の値で計算しておくと、終了見込みに余裕を持てます。

Q. 日をまたぐ依存（前日のジョブの結果を使う）はどう表しますか？ A. 同じジョブネットの中で矢印を戻すと循環依存になります。前日分の結果は「前日のジョブネットが終わっていること」を起動条件に書くか、結果ファイルの到着待ちとして待ち合わせのジョブにします。

まとめ

• ジョブフロー図は、ジョブを箱、先行関係を矢印で表す図。 先にジョブ一覧で先行ジョブを決め、表から図を起こす
• ジョブ一覧には、ジョブID・起動条件・先行ジョブ・異常時の扱い・再実行可否をそろえる。 先行ジョブには直前のジョブだけを書く
• 先行関係は、直列・並列・合流・分岐・待ち合わせ・異常系の6パターンの組み合わせで描ける
• 図にする前に、循環依存と一覧にないIDへの参照を検査する
• 全体の所要時間は最長経路で決まる。 最長経路の上のジョブを速くすると全体が縮む
• 正常系は実線、異常系は点線。 全体図は正常系を中心にし、異常系は一覧と要所の図で示す
• 図は手で直さず、一覧を直して生成し直す

参考・出典

• Python 公式ドキュメント「graphlib — Functionality to operate with graph-like structures」 https://docs.python.org/3/library/graphlib.html （追加された版、TopologicalSorter、CycleError の内容）

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。