DFS — вторая половина обхода

Если BFS — это волна, идущая от старта в все стороны равномерно, то DFS — это пробежка: бежим вглубь, пока есть куда. Дошли до конца — возвращаемся, пробуем следующее направление.

DFS — это про две вещи:

1. «Идти до упора» — спустился в потомка, дальше спустился в потомка потомка, и так далее.
2. Возврат — когда упёрлись в тупик или уже посещённую вершину, идём назад и пробуем другую ветку.

Это в точности поведение, которое возникает само собой при рекурсивной реализации. И это же поведение даёт stack (LIFO) вместо queue. Поэтому два эквивалентных способа реализовать DFS — рекурсия и явный стек.

Идея на маленьком графе

adj = {
    "A": ["B", "C"],
    "B": ["D", "E"],
    "C": ["F"],
    "D": [],
    "E": [],
    "F": [],
}

DFS из A:

1. Зашли в A. Соседи: B, C. Идём в B первым.
2. Зашли в B. Соседи: D, E. Идём в D.
3. Зашли в D. Соседей нет — это лист. Возвращаемся в B.
4. У B остался E. Идём в E.
5. E — лист. Возвращаемся в B, потом в A.
6. У A остался C. Идём в C.
7. C -> F -> лист. Возвращаемся.
8. Готово.

Порядок посещения: A, B, D, E, C, F. Глубина была главной — мы пошли максимально далеко вниз, потом откатились.

Реализация через рекурсию

Естественная и короткая:

def dfs_recursive(adj: dict, start, visited: set = None, order: list = None) -> list:
    if visited is None:
        visited = set()
        order = []
    visited.add(start)
    order.append(start)
    for v in adj.get(start, []):
        if v not in visited:
            dfs_recursive(adj, v, visited, order)
    return order

print(dfs_recursive(adj, "A"))
# ['A', 'B', 'D', 'E', 'C', 'F']

Стек вызовов Python играет роль стека алгоритма. Каждый вызов = шаг вглубь. Возврат из функции = откат назад.

Опасность: глубина стека Python по умолчанию ~1000. На очень глубоком графе (long lineage цепочки, 10 000+ уровней) рекурсия даст RecursionError. Решение — либо sys.setrecursionlimit(100000) (рискованно — может упасть с stack overflow на нативном уровне), либо итеративная реализация.

Реализация через явный stack

Эквивалентная, без риска переполнения:

def dfs_iterative(adj: dict, start) -> list:
    visited = set()
    stack = [start]
    order = []
    while stack:
        u = stack.pop()           # LIFO — последний добавленный
        if u in visited:
            continue
        visited.add(u)
        order.append(u)
        # добавляем соседей в обратном порядке, чтобы извлечь в исходном
        for v in reversed(adj.get(u, [])):
            if v not in visited:
                stack.append(v)
    return order

print(dfs_iterative(adj, "A"))
# ['A', 'B', 'D', 'E', 'C', 'F']

Главные отличия от BFS:

Всё. Один и тот же скелет, разная очередь — два разных алгоритма.

Pre-order vs post-order

В DFS у каждой вершины есть два момента:

• pre-order — момент, когда вы заходите в вершину (только что добавили в visited),
• post-order — момент, когда выходите из неё (все соседи уже обработаны).

def dfs_pre_post(adj, start):
    visited = set()
    pre = []
    post = []
    def rec(u):
        if u in visited:
            return
        visited.add(u)
        pre.append(u)          # вход
        for v in adj.get(u, []):
            rec(v)
        post.append(u)         # выход
    rec(start)
    return pre, post

pre, post = dfs_pre_post(adj, "A")
print("pre :", pre)
# ['A', 'B', 'D', 'E', 'C', 'F']
print("post:", post)
# ['D', 'E', 'B', 'F', 'C', 'A']

Заметьте: post — это «reverse topological order» для DAG. Если у нас лист — он выходит первым. Корень — последним. Если развернуть post, получим порядок, в котором можно исполнять задачи (зависимости перед зависимыми). Это и есть topological sort.

def topological_sort(adj, vertices):
    visited = set()
    post = []
    def rec(u):
        if u in visited:
            return
        visited.add(u)
        for v in adj.get(u, []):
            rec(v)
        post.append(u)
    for v in vertices:
        if v not in visited:
            rec(v)
    return list(reversed(post))

print(topological_sort(adj, ["A", "B", "C", "D", "E", "F"]))
# ['A', 'C', 'F', 'B', 'E', 'D']
# (или другой topological-валидный порядок)

Этот код — основа Airflow scheduler, dbt compile, любого DAG-планировщика.

Сложность

Та же, что у BFS:

Память DFS — O(глубины пути) для стека плюс O(V) для visited. На сбалансированном дереве из миллиона вершин глубина = log(10^6) ≈ 20, стек ничтожный. На «верёвке» (1 -> 2 -> 3 -> … -> V) глубина = V, стек большой. Это разбираем в следующем уроке.

Применение 1: cycle detection в DAG

Главная задача DFS в DE. Делается через трёхцветную раскраску:

• white — вершина ещё не открыта,
• gray — открыта, но не закрыта (мы в её поддереве),
• black — полностью обработана (вышли).

Если во время обхода встречаем ребро в gray-вершину — это цикл.

WHITE, GRAY, BLACK = 0, 1, 2

def has_cycle(adj, vertices) -> bool:
    color = {v: WHITE for v in vertices}

    def dfs(u):
        color[u] = GRAY
        for v in adj.get(u, []):
            if color[v] == GRAY:
                return True       # back edge — цикл!
            if color[v] == WHITE and dfs(v):
                return True
        color[u] = BLACK
        return False

    for v in vertices:
        if color[v] == WHITE:
            if dfs(v):
                return True
    return False

Это единственный способ корректно обнаружить цикл в directed-графе через DFS. Просто visited (binary) не подойдёт — он не различает «уже видели в текущей ветке» и «уже видели в другой ветке». Подробнее это в уроке 12.05.

Применение 2: connected components

В undirected-графе один DFS из вершины обходит всю её связную компоненту. Запустив DFS из всех непосещённых вершин, мы найдём все компоненты:

def connected_components(adj, vertices):
    visited = set()
    components = []
    for v in vertices:
        if v not in visited:
            stack = [v]
            comp = []
            while stack:
                u = stack.pop()
                if u in visited:
                    continue
                visited.add(u)
                comp.append(u)
                for w in adj.get(u, []):
                    if w not in visited:
                        stack.append(w)
            components.append(comp)
    return components

В DE: «найти все изолированные подграфы lineage» — типичный кейс при чистке мёртвых моделей.

Применение 3: topological sort

Описано выше. Используется везде, где есть DAG задач:

• Airflow scheduler: выстраивает задачи в порядке исполнения,
• dbt compile: вычисляет порядок моделей,
• Spark physical plan,
• Make-файлы, build-системы (Bazel, Buck).

# topological sort через DFS (post-order reverse)
def topo_dfs(adj, vertices):
    visited = set()
    result = []
    def rec(u):
        visited.add(u)
        for v in adj.get(u, []):
            if v not in visited:
                rec(v)
        result.append(u)
    for v in vertices:
        if v not in visited:
            rec(v)
    return result[::-1]

DE-кейс: проверка валидности dbt-проекта

Соберём все три применения:

import json
from collections import defaultdict

def validate_dbt_project(manifest_path: str) -> dict:
    with open(manifest_path) as f:
        manifest = json.load(f)

    # построение adj
    adj: dict[str, list[str]] = defaultdict(list)
    vertices = set()
    for node_id, info in manifest["nodes"].items():
        if not node_id.startswith("model."):
            continue
        name = info["name"]
        vertices.add(name)
        for dep_id in info.get("depends_on", {}).get("nodes", []):
            if dep_id.startswith("model."):
                dep_name = manifest["nodes"][dep_id]["name"]
                adj[dep_name].append(name)   # стрелка parent -> child
                vertices.add(dep_name)

    # проверка циклов
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {v: WHITE for v in vertices}
    cycle_found = [False]
    def dfs_cycle(u):
        if cycle_found[0]:
            return
        color[u] = GRAY
        for v in adj.get(u, []):
            if color[v] == GRAY:
                cycle_found[0] = True
                return
            if color[v] == WHITE:
                dfs_cycle(v)
        color[u] = BLACK

    for v in vertices:
        if color[v] == WHITE:
            dfs_cycle(v)

    # topological sort (только если нет цикла)
    topo: list[str] = []
    if not cycle_found[0]:
        visited = set()
        post = []
        def dfs_topo(u):
            visited.add(u)
            for v in adj.get(u, []):
                if v not in visited:
                    dfs_topo(v)
            post.append(u)
        for v in vertices:
            if v not in visited:
                dfs_topo(v)
        topo = list(reversed(post))

    return {
        "has_cycle": cycle_found[0],
        "topological_order": topo,
        "model_count": len(vertices),
    }

Это рабочая утилита: 50 строк, обнаруживает циклы (главная проверка для dbt) и возвращает порядок исполнения.

Когда использовать DFS, а когда BFS

Граница не жёсткая: многие задачи решаются обоими. DFS предпочтительнее для structural-вопросов (циклы, порядок), BFS — для distance-вопросов.

Попробуй сам

Возьмём lineage:

adj = {
    "stg_users": ["int_user_orders"],
    "stg_orders": ["int_user_orders", "int_order_items"],
    "stg_products": ["int_order_items"],
    "int_user_orders": ["mart_user_revenue"],
    "int_order_items": ["mart_user_revenue", "mart_top_products"],
    "mart_user_revenue": [],
    "mart_top_products": [],
}

vertices = list(adj.keys())

Задачи:

1. Реализуйте topological_sort(adj, vertices) через post-order DFS. Корректный результат — любой, где каждая модель идёт после своих зависимостей.
2. Добавьте ребро mart_user_revenue -> stg_orders (создаёт цикл) и реализуйте has_cycle(adj, vertices) через трёхцветную раскраску.
3. Реализуйте longest_path_length(adj) — максимальную длину цепочки в DAG. Подсказка: DP по topological order, либо мемоизированная рекурсия.

В следующем уроке — компромисс памяти между BFS и DFS, разбор «когда что».