Directed Acyclic Graph (DAG) 응용 및 활용 분야

 

 

이전 플로그 포스팅(https://marcobackman.tistory.com/34)에서 모든 노드들을 시작과 끝을 순서로 선형 정렬을 하는 Topological Sort에 응용되 DAG 관련한 알고리즘을 다루었었는데요.

Topological Sort란? Kahn's Algorithm으로 DAG 정렬 쉽게 이해하기

 

 

오늘은 해당 알고리즘으로 조금더 현실적으로 적용되는 사례와 예시를 보여드리고자 합니다.

 

우선 DAG의 정의부터 다시 복기를 하자면 다음과 같습니다.

그래프에서 각 노드들을 잇는 방향성 간선이 있으며 DAG를 충족하려면 순환되는 구간이 없어야 합니다.

 

DAG에 응용되는 분야는 다음과 같습니다.

 

• 일정 스케줄, 의존성 해결(Task Scheduling & Dependency Resolution) - 만약 스케줄 관리에서 특정 작업이 다른 일정 전에 반드시 먼저 끝나야 한다면 작업을 노드로 취급하고 의존성은 방향성 간선으로 구현 합니다. Topological sort를 통해 특정 작업이 유효한지 아닌지를 확인합니다. 모듈간의 컴파일, 의존성 관리, 대학교 강의 등 많은 분야에서 응용됩니다.
• 데이터 처리 파이프라인(Data Processing Pipelines) - ETL (Extract, Transform, Load) 작업 플로우에서는 각 프로세스 순서 별로 데이터가 흐르는데, 각 순서를 stage로 취급하고 데이터 흐름을 방향성 간선으로 취급하여 DAG를 활용합니다.
• 버전 관리 도구(Version Control Systems) - Git 커밋에서는 하나의 커밋은 하나 이상의 부모를 가질 수 있으며 이 부분을 방향성 간선으로 취급 할 수 있습니다. 여기서 모든 커밋은 순환구간이 존재하면 안되며 이때도 DAG를 사용합니다.
• 스프레드 시트 연산 (Spreadsheet Calculations) - 엑셀이나 셀도구에서 각 셀을 노드로 취급했을 때, 함수 연산을 활용하여 각 셀들의 연산을 관계를 지어 의존성이 형성이 되는경우, DAG를 사용해서 순환을 피해야 무한 루프에 빠지는 문제를 피할 수 있습니다.
  
  

DAG의 기본 코드는 다음과 같습니다.

 

#Returns Fasle if there's no cycle. Otherwise return True
def has_cycle(graph: dict, node_to_start: str):
    visited_dict = set()

    if node_to_start not in graph:
        print(f"There's no node called {node_to_start} in given graph")

    neighbors = graph[node_to_start]
    visited_dict.add(node_to_start)
    node_queue = [neighbor for neighbor in neighbors]

    while node_queue:
        next_node = node_queue.pop()
        if next_node in visited_dict:
            return False
        visited_dict.add(next_node)

        for new_node in graph[next_node]:
            node_queue.append(new_node)
    return True

dag = {
    'A': ['C'],
    'B': ['C', 'D'],
    'C': ['E'],
    'D': ['F'],
    'E': [],
    'F': []
}

print(has_cycle(dag, 'A'))
print(has_cycle(dag, 'B'))
print(has_cycle(dag, 'C'))
print(has_cycle(dag, 'D'))
print(has_cycle(dag, 'E'))
print(has_cycle(dag, 'E'))

cyclic_graph = {
    'A': ['B'],
    'B': ['C'],
    'C': ['A']
}

print(has_cycle(cyclic_graph, 'A'))
print(has_cycle(cyclic_graph, 'B'))
print(has_cycle(cyclic_graph, 'C'))

 

---

 

스케줄러 구현 코드 (Topological sort)

from typing import Dict, List
from collections import deque

def topological_sort(graph : Dict[str, List[str]]):
    """
    위상 정렬을 수행하여 작업의 올바른 실행 순서를 반환합니다.

    Args:
        graph: 의존성을 나타내는 그래프 (딕셔너리 형태).
               key: 작업, value: 해당 작업이 의존하는 작업들의 리스트.
               (예: '알고리즘': ['자료구조'] -> 자료구조가 끝나야 알고리즘 시작 가능)

    Returns:
        작업의 실행 순서가 담긴 리스트. 순환이 있어 정렬이 불가능하면 빈 리스트를 반환합니다.
    """
    in_degree = {node : 0 for node in graph}
    adj = {node : [] for node in graph}

    for node, prerequisites in graph.items():
        for prereq in prerequisites:
            # prereq -> node 관계
            adj[prereq].append(node)
            in_degree[node] += 1

    #진입 차수가 0인 노드는 모두 시작지점
    queue = deque([node for node in in_degree if in_degree[node] == 0])
    result = []

    while queue:
        current_node = queue.popleft()
        result.append(current_node)

        for neighbor in adj[current_node]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)

    #결과
    if len(result) == len(graph):
        return result
    else:
        return []

course_dependencies = {
    '프로그래밍 입문': [],
    '컴퓨터 구조': [],
    '자료구조': ['프로그래밍 입문'],
    '알고리즘': ['자료구조'],
    '운영체제': ['컴퓨터 구조', '자료구조'],
    '데이터베이스': ['자료구조']
}

# 테스트 케이스 1: 순환이 있는 그래프 (A -> B -> C -> A)
# 결과: 정렬이 불가능해야 합니다.
cyclic_graph = {
    'A': ['C'],
    'B': ['A'],
    'C': ['B']
}

# 테스트 케이스 2: 여러 개의 독립적인 시작점이 있는 그래프 (Disconnected Graph)
# 'A'와 'X' 모두 선수과목이 없습니다.
disconnected_graph = {
    'A': ['B'],
    'B': ['C'],
    'C': [],
    'X': ['Y'],
    'Y': [],
    'Z': [] # Z는 완전히 독립적입니다.
}

# 테스트 케이스 3: 선형적인(일자 형태) 의존성을 가진 그래프
# 마치 체인처럼 연결됩니다.
linear_graph = {
    'Step 1': [],
    'Step 2': ['Step 1'],
    'Step 3': ['Step 2'],
    'Step 4': ['Step 3']
}

# 테스트 케이스 4: 복잡한 의존성을 가진 그래프
# 여러 과목이 하나의 과목을 선수과목으로 요구하는 경우
complex_dag = {
    '기초 수학': [],
    '자료구조': ['기초 수학'],
    '알고리즘': ['자료구조'],
    '컴파일러': ['알고리즘'],
    '운영체제': ['알고리즘', '컴퓨터 구조'],
    '컴퓨터 구조': ['기초 수학'],
    '네트워크': ['운영체제']
}

# 테스트 케이스 5: 비어 있는 그래프
empty_graph = {}


print("### 테스트 1: 순환 그래프 ###")
result1 = topological_sort(cyclic_graph)
if not result1:
    print("결과: 순환이 있어 스케줄을 만들 수 없습니다. (정상)")
else:
    print(f"결과: {result1} (오류)")

print("\n### 테스트 2: 분리된 그래프 ###")
result2 = topological_sort(disconnected_graph)
if result2:
    print(f"결과: {' -> '.join(result2)} (정상)")
else:
    print("결과: 정렬 실패 (오류)")

print("\n### 테스트 3: 선형 그래프 ###")
result3 = topological_sort(linear_graph)
if result3:
    print(f"결과: {' -> '.join(result3)} (정상)")
else:
    print("결과: 정렬 실패 (오류)")

print("\n### 테스트 4: 복잡한 그래프 ###")
result4 = topological_sort(complex_dag)
if result4:
    print(f"결과: {' -> '.join(result4)} (정상)")
else:
    print("결과: 정렬 실패 (오류)")

print("\n### 테스트 5: 빈 그래프 ###")
result5 = topological_sort(empty_graph)
if result5 == []:
    print("결과: [] (정상)")
else:
    print(f"결과: {result5} (오류)")

 

결과는 다음과 같이 나와야 정상입니다.

 

---

추가 활용 및 적용 사례

 

SCM(공급망 관리)에서 DAG(방향성 비순환 그래프)는 프로세스의 의존성을 시각화하고 병목 현상을 찾아내며 전체 흐름을 최적화하는 데 매우 유용하게 활용됩니다. 각 단계를 노드(Node)로, 단계 간의 선후 관계를 방향성이 있는 간선(Edge)으로 표현하여 복잡한 공급망 프로세스를 명확한 모델로 만들 수 있습니다.

---

SCM에서의 DAG 활용 분야

제조 및 생산 공정

제조 공정은 여러 단계가 순차적으로 또는 병렬적으로 진행됩니다. DAG를 이용하면 이 복잡한 과정을 명확하게 모델링할 수 있습니다.

• 모델링: 각 생산 단계(부품 조달, 가공, 조립, 검수, 포장)를 노드로, 작업 순서를 간선으로 표현합니다.
• 활용:
  • 병목 식별: 특정 노드로 들어오는 간선이 많거나, 해당 노드의 처리 시간이 길면 병목 지점임을 쉽게 파악할 수 있습니다.
  • 공정 스케줄링: 위상 정렬(Topological Sort)을 통해 가장 효율적인 작업 순서를 결정할 수 있습니다.
  • 최장 경로 분석 (Critical Path Analysis): 전체 생산 시간을 결정하는 가장 긴 작업 경로(Critical Path)를 찾아내 해당 경로를 집중 관리하여 리드타임을 단축할 수 있습니다