Зачем нужен обход графа

В прошлом модуле мы научились хранить граф. Хранение — это половина дела; вторая — извлекать из графа информацию. «Кто потомки таблицы X?», «Какие модели нужно пересчитать после изменения Y?», «Есть ли цикл?», «Какой минимальный путь между задачами A и B в Airflow DAG?» — все эти вопросы — это разновидности обхода графа.

Два главных алгоритма обхода:

• BFS (Breadth-First Search) — обход в ширину. Расширяется уровнями: сначала все соседи стартовой вершины, потом все соседи соседей, и так далее.
• DFS (Depth-First Search) — обход в глубину. Идёт максимально далеко вдоль одной ветки, потом возвращается.

Этот урок — про BFS. Следующий — про DFS. Они кажутся очень похожими (структура одна, отличается одна структура данных внутри), но дают принципиально разные свойства.

Идея BFS

Представьте: вы стоите на вершине S и хотите узнать, как далеко до каждой другой вершины. Делаете так:

1. Сначала отметьте S как «открыта на расстоянии 0».
2. Посмотрите всех её соседей — это «расстояние 1».
3. Посмотрите всех соседей этих соседей — это «расстояние 2».
4. И так далее, пока не обойдёте всё достижимое.

То есть вы продвигаетесь «волной» — обходите граф по уровням близости к старту. Отсюда название: breadth (ширина) — потому что каждый уровень в один момент целиком в работе, а не одна цепочка вглубь.

Структура данных: queue

Чтобы обходить именно по уровням, нужна структура FIFO (first-in-first-out) — очередь. То, что положили раньше, обработается раньше. Тогда сначала разберёмся со всеми соседями уровня 1, потом перейдём к уровню 2.

В Python для этого есть collections.deque — двусторонняя очередь. Её append/popleft — O(1). Использовать обычный list через pop(0) — анти-паттерн: list.pop(0) сдвигает все остальные элементы, это O(n). На графе в 100k вершин это превратит BFS из O(V+E) в O(V^2).

Алгоритм BFS пошагово

from collections import deque

def bfs(adj: dict[int, list[int]], start: int) -> list[int]:
    """Возвращает вершины в порядке их посещения BFS-обходом."""
    visited = {start}             # set посещённых вершин — O(1) lookup
    queue = deque([start])        # очередь к обработке
    order = []                    # порядок посещения

    while queue:
        u = queue.popleft()       # извлекаем фронт очереди — O(1)
        order.append(u)

        for v in adj.get(u, []):  # перебираем соседей
            if v not in visited:
                visited.add(v)
                queue.append(v)   # добавляем в конец — O(1)

    return order

Запустим на маленьком графе:

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

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

Шаги:

1. Стартуем с A. visited = {A}, queue = [A].
2. Извлекли A. Добавили B, C в visited и queue. visited = {A, B, C}, queue = [B, C].
3. Извлекли B. Сосед A уже visited, D — нет, добавили. visited = {A, B, C, D}, queue = [C, D].
4. Извлекли C. Соседи A и D уже visited, E — нет. visited = {A, B, C, D, E}, queue = [D, E].
5. Извлекли D. Все соседи visited.
6. Извлекли E. Все соседи visited.
7. Queue пуст — стоп.

Порядок: A, B, C, D, E. Сначала уровень 0 (A), потом уровень 1 (B, C), потом уровень 2 (D, E).

Три обязательных компонента

В BFS три ключевых элемента:

1. Visited set — чтобы не вернуться в уже посещённую вершину. Без него BFS на графе с циклами зациклится. Set — потому что нужен O(1) lookup v in visited.
2. Queue (deque) — для FIFO-порядка.
3. Проверка visited ПЕРЕД добавлением, а не при извлечении. Если проверять при извлечении, одна вершина может попасть в очередь несколько раз — это бесполезное удваивание работы.

# хороший паттерн
for v in adj[u]:
    if v not in visited:
        visited.add(v)
        queue.append(v)

# плохой паттерн — допускает дубли в очереди
for v in adj[u]:
    queue.append(v)
# ... позже ...
u = queue.popleft()
if u in visited:
    continue

Сложность

Время — O(V + E). Каждая вершина извлекается из очереди не более одного раза (благодаря visited). Каждое ребро рассматривается не более одного раза (когда обходим соседей вершины u). Никакой работы сверх этого. Линейная сложность от размера графа — лучше быть не может.

Память — O(V). Visited хранит все вершины. Queue в худшем случае держит весь один уровень — например, на «звезде» (центральная вершина и V-1 соседей вокруг) после извлечения центра в очереди окажется V-1 элементов.

BFS на adjacency matrix

Тот же алгоритм, но соседей берём из строки матрицы:

import numpy as np

def bfs_matrix(matrix: np.ndarray, start: int) -> list[int]:
    n = matrix.shape[0]
    visited = np.zeros(n, dtype=bool)
    visited[start] = True
    queue = deque([start])
    order = []
    while queue:
        u = queue.popleft()
        order.append(u)
        # перебираем соседей через строку u
        for v in np.where(matrix[u] == 1)[0]:
            if not visited[v]:
                visited[v] = True
                queue.append(int(v))
    return order

Сложность — O(V * V) = O(V^2): для каждой вершины проходим всю строку. На sparse-графе это медленнее, чем adjacency list (O(V+E)). На dense — то же самое.

Применение 1: достижимость

«Какие вершины достижимы из S?» — это и есть результат BFS:

def reachable(adj, start):
    visited = {start}
    q = deque([start])
    while q:
        u = q.popleft()
        for v in adj.get(u, []):
            if v not in visited:
                visited.add(v)
                q.append(v)
    return visited

print(reachable(adj, "A"))
# {'A', 'B', 'C', 'D', 'E'}

В DE: «какие модели нужно пересчитать, если меняется stg_orders?» = BFS из stg_orders по children-графу. Все полученные вершины — кандидаты на пересборку.

Применение 2: shortest path в unweighted-графе

BFS даёт кратчайший путь в графе без весов. Не путь в смысле «короче по сумме рёбер», а в смысле «минимум рёбер». Доказательство простое: BFS перебирает вершины в порядке возрастания расстояния от старта, поэтому первое попадание в вершину = минимальное расстояние.

def shortest_distances(adj, start):
    dist = {start: 0}
    q = deque([start])
    while q:
        u = q.popleft()
        for v in adj.get(u, []):
            if v not in dist:
                dist[v] = dist[u] + 1
                q.append(v)
    return dist

print(shortest_distances(adj, "A"))
# {'A': 0, 'B': 1, 'C': 1, 'D': 2, 'E': 2}

dist сразу выполняет роль visited (если вершина в dict — она посещена). Никаких дополнительных структур.

Внимание: для weighted-графа BFS даёт неправильный ответ. Нужен Dijkstra (мы коснёмся в уроке 04 этого модуля). BFS работает только когда все рёбра имеют одинаковый вес.

Применение 3: level-order traversal

Если важно знать, на каком уровне находится вершина, BFS даёт это естественно. Часто это записывают «по уровням», обрабатывая весь уровень целиком в одном шаге:

def bfs_by_levels(adj, start):
    visited = {start}
    current_level = [start]
    levels = []
    while current_level:
        levels.append(current_level)
        next_level = []
        for u in current_level:
            for v in adj.get(u, []):
                if v not in visited:
                    visited.add(v)
                    next_level.append(v)
        current_level = next_level
    return levels

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

Этот вариант полезен, когда нужно работать с уровнем целиком: например, «найдите всех пользователей, отстоящих на k шагов».

DE-кейс: impact analysis dbt

Сценарий: вам нужно понять, какие модели сломаются, если изменить колонку в stg_orders. Это типичная задача impact analysis. Решается BFS-ом по downstream-направлению lineage.

def downstream_impact(children: dict[str, list[str]], target: str) -> set[str]:
    """Все модели, которые прямо или косвенно зависят от target."""
    visited = {target}
    q = deque([target])
    while q:
        u = q.popleft()
        for v in children.get(u, []):
            if v not in visited:
                visited.add(v)
                q.append(v)
    return visited - {target}  # сам target не считаем

print(downstream_impact(children, "stg_orders"))
# {'int_user_orders', 'int_order_items', 'mart_user_revenue', 'mart_top_products'}

В dbt вызывается dbt run --select stg_orders+ — это «stg_orders и всё что вниз». Внутри dbt делает ровно такой BFS.

DE-кейс: friends-of-friends

Социальный граф пользователей, нужно найти всех друзей друзей пользователя u (расстояние 2). Это два уровня BFS:

def friends_of_friends(adj, u):
    visited = {u}
    fof = set()
    current = [u]
    for depth in range(2):
        nxt = []
        for x in current:
            for y in adj.get(x, []):
                if y not in visited:
                    visited.add(y)
                    nxt.append(y)
                    if depth == 1:
                        fof.add(y)
        current = nxt
    return fof

Точнее: на depth=0 мы из u шагнули в его друзей. На depth=1 шагнули из друзей дальше. Те, кто оказался на depth=1, — friends-of-friends. visited гарантирует, что друг моего друга, который и так мой друг, не попадёт в fof дважды.

Попробуй сам

Дан граф (это foreign-key схема из урока 11.05):

adj = {
    "users": ["orders", "sessions", "events"],
    "products": ["order_items"],
    "orders": ["order_items", "payments"],
    "sessions": ["events"],
    "events": [],
    "order_items": [],
    "payments": [],
}

Задачи:

1. Реализуйте функцию count_at_distance(adj, start, k), возвращающую число вершин на расстоянии ровно k от start. Для count_at_distance(adj, "users", 2) = 3 (order_items, payments, events — на расстоянии 2 через orders/sessions).
2. Реализуйте функцию is_reachable(adj, source, target) -> bool. Проверьте is_reachable(adj, "products", "payments") — должно быть False.
3. Реализуйте функцию shortest_path(adj, source, target) -> list[str] | None, возвращающую сами вершины кратчайшего пути (не только длину). Подсказка: храните parent[v] — откуда BFS пришёл в v, и восстановите путь обратно.

В следующем уроке — DFS, ещё одна структура данных (stack или рекурсия), другие применения.