정현석

개인 저장공간입니다.
Since 2021.10

14 분 소요

image

“GraphX: Graph Processing in a Distributed Dataflow Framework” 논문을 개인 공부 및 리뷰를 위해 정리한 글이다.

대규모 분산 환경에서 그래프 처리는 기존 분산 데이터플로우 프레임워크와 특화된 그래프 처리 시스템 사이에서 오랜 딜레마를 겪어왔다. 본 논문은 특화 엔진을 별도로 구축하지 않고도, 그래프 계산을 범용 데이터플로우 연산자의 특수한 패턴으로 재구성하여 Apache Spark 엔진 위에서 전문 그래프 시스템에 필적하는 성능과 유연성을 함께 확보하는 GraphX 시스템을 제안한다. 저자들이 보고한 실측 평가에서 GraphX는 기본 데이터플로우 구현 대비 한 자릿수(order of magnitude) 수준의 성능 향상을 보였고, 특화 그래프 처리 시스템과 동등한 성능을 달성했다1.

논문 출처 pdf : graphx paper 논문의 이해를 돕기 위해 GraphX에 대한 ppt 링크도 첨부하겠습니다. GraphX ppt in stanford.edu 링크 : GraphX - Graph Analytics in Spark 논문에 나온 프로그래밍 언어는 스칼라(Scala)로 기재했습니다.

1. Introduction

분산 환경의 대규모 데이터 파이프라인에서 그래프 연산은 독립적으로 존재하지 않고, 테이블 데이터나 비정형 텍스트와 빈번하게 결합한다. 그러나 기존 시스템 아키텍처는 특화 그래프 엔진과 범용 데이터플로우 프레임워크로 양분되어 데이터 이동과 복잡성 비용을 초래했다.

  • 프리겔(Pregel) 같은 시스템은 그래프별 최적화를 지원하는 특수 추상화(abstraction)를 노출함으로써 수십억 개의 정점(vertex)과 에지(edge)가 있는 그래프에서 PageRank 같은 반복 그래프 알고리즘을 효율적으로 실행할 수 있다.
  • 그 결과, 그래프 처리 시스템은 일반적으로 하둡 맵리듀스(MapReduce)와 같은 범용 분산 데이터 흐름 프레임워크보다 반복 그래프 알고리즘에서 더 나은 성능을 보인다.
  • 사실 그래프는 비정형 및 표 형식의 데이터와 결합하는 더 큰 분석 프로세스의 일부일 뿐이다.
  • 결과적으로, 분석 파이프라인(Fig 11)은 여러 시스템을 구성하도록 강요되어 복잡성을 증가시키고 불필요한 데이터 이동과 복제를 초래한다.

결국 데이터 파이프라인 분리는 불필요한 직렬화와 네트워크 전송 비용을 증폭시킨다. image

  • 맵리듀스, 스파크, 드라이어드는 풍부한 데이터 흐름 연산자(map, reduce, groupBy, join)를 노출하며, 비정형 및 표 형식의 데이터를 분석하는 데 적합하고 널리 채택된다.
  • 그러나 데이터 흐름 연산자를 사용하여 반복 그래프 알고리즘을 직접 구현하는 것은 어려울 수 있으며, 여러 단계의 복잡한 결합이 필요하다.
  • 또한 분산 데이터 흐름 프레임워크에 정의된 조인 및 집계 전략은 반복 그래프 알고리즘에서 공통 패턴과 구조를 활용하지 않으므로 중요한 최적화 기회를 놓친다.

특화 그래프 시스템과 범용 데이터플로우 시스템이 별개로 발전해 온 데에는 실행 모델과 파티셔닝 전략의 차이가 존재한다.

  • 역사적으로 그래프 처리 시스템은 몇 가지 이유로 분산 데이터 흐름 프레임워크와 별도로 진화했다.
  • 첫째로, 분산 데이터 흐름 프레임워크(예: 맵리듀스)에서 싱글 스테이지 계산 및 디스크 내 처리에 대한 초기 강조로 그래프의 서브셋(subset)에 반복적으로 랜덤하게 액세스하는 반복 그래프 알고리즘에 대한 적용성(applicability)이 제한되었다.

  • 둘째, 초기 분산 데이터 흐름 프레임워크는 데이터 파티셔닝(partitioning)에 대한 세밀한 제어를 노출하지 않아 그래프 파티셔닝 기술의 적용을 방해했다.

  • 본 논문에서는 그래프 계산에서 필수적인 데이터 흐름 패턴을 식별하고 그래프 처리 시스템의 최적화를 데이터 흐름 최적화로 재구성함으로써 범용 분산 데이터 흐름 프레임워크 내에서 전문화된 그래프 처리 시스템의 장점을 복구할 수 있다고 주장한다.
  • 이 주장을 뒷받침하기 위해 저자들은 스파크에 내장된 효율적인 그래프 처리 프레임워크인 GraphX를 소개한다.

image

  • GraphX API는 기존 그래프 처리 시스템과 달리 비정형 및 표 형식의 데이터로 그래프를 구성할 수 있으며 데이터 이동이나 중복 없이 동일한 물리적 데이터를 그래프와 컬렉션으로 모두 볼 수 있다.
    • 예를 들어, GraphX를 사용하면 사용자 코멘트와 함께 소셜 그래프를 결합하고, 그래프 알고리즘을 적용하고, 결과를 collect 또는 그래프 시각화 또는 롤업(roll-up)하는 것이 쉽다.
  • 결과적으로, GraphX를 통해 사용자는 성능이나 유연성을 희생하지 않고도 현재 작업에 가장 적합한 계산 패턴을 채택할 수 있다.
  • GraphX는 Spark를 수정할 필요가 없으므로 분산 데이터 흐름 프레임워크 내에 그래프 계산을 내장하고 그래프 계산을 특정 join-map-group-by 데이터 흐름 패턴으로 증류(distill)하는 일반적인 방법을 보여준다.
  • 그래프 계산을 특정 패턴으로 줄임으로써 시스템 최적화를 위한 중요한 경로를 식별한다.

이러한 설계를 실현하기 위해 GraphX가 도입한 핵심 메커니즘은 다음과 같다.

  • 정점 절단(Vertex-Cut) 파티셔닝: 유연한 정점 절단 파티셔닝은 그래프를 수평으로 분할된 컬렉션(collection)으로 인코딩하고 분산 그래프 파티셔닝의 최신 기술과 일치시키는 데 사용된다.
  • 데이터플로우 관점의 재구성: GraphX는 그래프 처리 시스템의 맥락에서 개발된 시스템 최적화를 조인 최적화 및 구체화된 뷰 유지 관리로 재구성하고 이러한 기술을 스파크 데이터 흐름 연산자에 적용한다.
  • 경량 장애 허용: GraphX는 논리적 파티셔닝과 리니지(lineage)를 활용하여 저렴한 장애 허용(fault tolerance)을 달성한다.
  • 불변성과 인덱스 재사용: 불변성(immutability)을 이용하여 GraphX는 그래프 및 수집 뷰와 여러 반복에 걸쳐 인덱스를 재사용하여 메모리 오버헤드를 줄이고 시스템 성능을 향상시킨다.

2. Background

그래프 처리 시스템과 분산 데이터플로우 프레임워크는 설계 철학과 최적화 지점이 근본적으로 다르다.

  • 고수준에서, 그래프 처리 시스템은 정점과 이웃의 세분성에서 계산을 정의하고 그래프에 의해 미리 정의된 희소 종속 구조(sparse dependency structure)를 활용한다.

이미지출처 2

2.1 The Property Graph Data Model

그래프 알고리즘을 체계화하기 위해 시스템은 정점과 에지에 속성을 매핑하는 모델을 채택한다.

  • 그래프 처리 시스템은 그래프 구조 데이터를 속성(property) 그래프로 표현하며, 이는 사용자 정의 속성을 각 정점 및 에지와 연관시킨다.
    • 속성에는 메타 데이터와 프로그램 상태가 포함될 수 있다.
    • 속성 그래프(property graph): 각 정점과 에지에 연결된 사용자 정의 객체를 포함하는 방향 멀티 그래프이다. 3
  • 그래프 처리 시스템의 연산은 일반적으로 미리 선언된 희소 구조를 가진 싱글 속성 그래프와 관련하여 정의된다.
    • 희소 그래프(sparse graph): 간선의 수가 적은 그래프 4
  • 이러한 제한된 초점은 다양한 최적화를 용이하게 하지만, 여러 그래프와 하위 그래프에 걸쳐 있을 수 있는 분석 작업의 표현을 복잡하게 만든다.

정적인 속성 그래프 모델 위에 구축된 연산 방식이 바로 그래프 병렬 추상화이다.

2.2 The Graph-Parallel Abstraction

그래프 병렬 처리는 정점 단위의 로컬 연산과 이웃 간 상호작용의 반복으로 정의된다.

  • 페이지랭크(PageRank) 알고리즘은 인접한 정점과 에지의 속성을 기반으로 정점 속성을 반복적으로 변환한다.
  • 이 반복적인 로컬 변환(transformation)의 공통 패턴은 그래프 병렬 추상화의 기초를 형성한다.
  • 그래프 병렬 추상화에서 사용자 정의 정점 프로그램은 각 정점에 대해 동시에 인스턴스화되고 메시지(ex. Pregel) 또는 공유 상태(ex. PowerGraph)를 통해 인접한 정점 프로그램과 상호 작용한다.
// Listing 1: PageRank in Pregel
def PageRank(v: Id, msgs: List[Double]) {
    // 메시지 합계(sum)를 계산한다.
    var msgSum = 0
    for (m <- msgs) { msgSum += m}
    // 페이지랭크를 업데이트한다.
    PR(v) = 0.15 + 0.85 * msgSum
    // 새로운 PR로 메시지를 브로드새크스한다.
    for (j <- OutNbrs(v)) {
        msg = PR(v) / NumLinks(v)
        send_msg(to=j, msg)
    }
    // 종료를 위한 체크를 한다.
    if (converged(PR(v))) voteToHalt(v)
}

더 자세히 알고 싶다면 프리겔 논문 정리 링크 : [논문 리뷰] Pregel: A System for Large-Scale Graph Processing - poeun

  • 그래프 병렬 추상화는 그래프의 정적 이웃 구조 같은 반복 그래프 알고리즘에 적합하지만, 분리된(disconnected) 정점이 상호 작용하거나 계산이 그래프 구조를 변경하는 계산을 표현하는 데는 적합하지 않다.
  • 예를 들어, raw 텍스트, 비정형 데이터에서 그래프 구성 또는 그래프 조대화(coarsening), 여러 그래프에 걸친 분석 등의 작업은 정점 중심 프로그래밍 모델에서 표현하기 어렵다.

이러한 한계에도 불구하고 기존 그래프 전용 엔진들은 독자적인 내부 최적화를 정교하게 발전시켜 왔다.

2.3 Graph System Optimizations

특화 그래프 엔진은 통신과 동기화 비용을 줄이기 위해 GAS 분해와 정점 절단 같은 구조적 최적화를 적극 활용한다.

GAS 분해(Decomposition)

  • 대부분의 정점 프로그램은 일반화된 교환(commutative) 결합(associative) 합계의 형태로 메시지를 수집한 다음 병렬 루프에서 새로운 메시지를 브로드캐스트함으로써 이웃 정점과 상호 작용한다.
  • Gonzalez et al는 정점 프로그램을 세 개의 데이터 병렬 스테이지로 나누는 GAS 분해를 제안하였다. (PowerGraph) 5
    • Gather: 이웃 정점과 에지로부터 정보를 병렬로 수집(gather)하고 교환·결합 법칙에 따라 누적 합산한다.
    • Apply: 축적된 메시지 합계를 바탕으로 정점 자체의 속성을 업데이트한다.
    • Scatter: 갱신된 정점 속성을 바탕으로 인접한 에지를 통해 새로운 메시지를 주변으로 전파한다. 
      def Gather(a: Double, b: Double) = a + b
def Apply(v, msgSum) {
  PR(v) = 0.15 + 0.85 * msgSum
  if (converged(PR(v))) voteToHalt(v)
}
def scatter(v, j) = PR(v) / NumLinks(v)
  • GAS 분해는 메시지 계산의 pull 기반 모델로 이어진다.
  • 시스템은 사용자가 정점 프로그램에서 직접 메시지를 보내는 대신 인접한 정점 사이의 메시지 값을 정점 프로그램에 묻는다.
  • 결과적으로 GAS 분해는 정점 절단 파티셔닝, 작업 균형 개선, 직렬 에지 반복 및 데이터 이동 감소를 가능하게 한다.
  • 그러나 GAS 분해는 그래프에서 인접하지 않은 정점 간의 직접적인 통신을 금지하므로, 보다 일반적인 통신 패턴의 표현을 방해한다.

Graph Partitioning

분산 그래프의 통신량은 파티셔닝 기법에 의해 결정적인 영향을 받는다.

  • 그래프 처리 시스템은 통신을 최소화하고 계산을 균형 있게 하기 위해 다양한 그래프-파티셔닝 알고리즘을 적용한다.
  • 정점 절단 파티셔닝은 각 정점이 절단되는 횟수를 최소화하는 방식으로 머신에 에지를 균등하게 할당한다.

정점 절단 환경에서는 통신량과 활성 작업량을 제어하기 위해 다음 기법들이 병행된다.

  • Mirror Vertices:
    • 종종 고차원의 정점은 동일한 원격 시스템에 여러 개의 이웃이 있을 것이다.
    • 동일한 여러 개의 메시지를 네트워크를 통해 전송하는 대신, 그래프 처리 시스템은 싱글 메시지를 미러로 보낸 다음 모든 이웃으로 전달하는 미러링 기술을 채택한다.
    • 그래프 처리 시스템은 정적(static) 그래프 구조를 활용하여 미러 데이터 구조를 재사용한다.
  • Active Vertices:
    • 그래프 알고리즘이 진행됨에 따라 그래프 내의 정점 프로그램은 서로 다른 속도로 수렴되어 활성 정점 프로그램의 집합이 급격히 축소된다.
    • 최근 시스템은 활성 정점을 추적하고 수렴된 정점에 대한 데이터 이동과 불필요한 계산을 제거한다.

이러한 그래프 전용 최적화들을 일반 데이터플로우 프레임워크의 연산 모델과 어떻게 대응시킬 것인가가 GraphX의 출발점이다.

2.4 Distributed Dataflow Frameworks

대규모 분산 데이터플로우 프레임워크는 범용성과 결함 감내 능력에서 탁월한 장점을 지닌다.

  • 분산 데이터 흐름 프레임워크의 세부 사항은 일반적으로 다음과 같은 특성을 만족한다.
    1. 유형 컬렉션 데이터 모델: 강타입(typed) 컬렉션으로 구성된 데이터 모델을 제공한다.
    2. 결정론적 연산자: 컬렉션을 변환하는 결정론적 연산자(ex. map, groupBy, join)로 구성된 데이터 정렬 프로그래밍 모델을 갖는다.
    3. DAG 스케줄러: 각 작업을 DAG(Directed Acyclic Graph)로 분할하여 최적 실행 계획을 세운다.
    4. 고신뢰 런타임: 전체 작업을 다시 시작하지 않고도 스트래글러(straggler)와 부분(partial) 클러스터 장애를 허용할 수 있다.

Spark는 이러한 특성 중에서도 반복 그래프 계산을 구현하기에 가장 적합한 기반을 제공한다.

  • 스파크는 GraphX를 위해 매력적인 특징들을 가지고 있다.
    1. RDD 기반 메모리 유지: RDD(Resilient Distributed Datasets)라고 하는 spark 스토리지 추상화는 애플리케이션이 데이터를 메모리에 유지할 수 있도록 하며, 이는 반복 그래프 알고리즘에 필수적이다.
    2. 사용자 정의 파티셔닝: RDD는 사용자 정의 데이터 파티셔닝을 허용하며 실행 엔진은 이를 활용하여 데이터 이동을 피하기 위해 RDD를 공동 파티셔닝하고 작업을 공동 스케줄링할 수 있다.
      • 이것은 파티셔닝된 그래프를 인코딩하는 데 필수적이다.
    3. 리니지 기반 경량 결함 감내: 스파크는 RDD를 구축하는 데 사용되는 작업 리니지를 기록하여 장애 발생 시 손실된 파티션을 자동으로 재구성할 수 있다.
      • 리니지 그래프는 장시간 실행되는 애플리케이션에서도 상대적으로 작기 때문에 체크포인트와 달리 런타임 오버헤드를 무시할 수 있다.
    4. 풍부한 표현력의 고수준 API: 스파크는 쉽게 확장할 수 있는 scala의 고급 API를 제공한다.

RDD의 불변성과 메모리 캐싱, 리니지는 반복 그래프 알고리즘에 적합한 기반을 제공한다.

3. The GraphX Programming Abstraction

GraphX의 핵심 추상화는 속성 그래프를 독립된 컬렉션 쌍으로 바라보고, 그래프 연산을 관계형 연산자로 통합하는 것이다.

3.1 Property Graphs as Collections

이미지출처 2

  • 속성 그래프는 한 쌍의 정점과 에지 속성 컬렉션으로 논리적으로 표현될 수 있다.
  • 정점 컬렉션은 정점 식별자에 의해 고유하게 키가 지정된 정점 속성을 포함한다.
  • GraphX 시스템에서 정점 식별자는 64비트 정수이며 외부(ex. 사용자 ID)에서 파생되거나 정점 속성(ex. 페이지 URL)에 해시 함수를 적용하여 파생될 수 있다.
  • 에지 컬렉션에는 소스 및 목적지 정점 식별자에 의해 키가 지정된 에지 속성이 포함되어 있다.
  • 속성 그래프를 컬렉션 쌍으로 줄임으로써 분산 데이터흐름 프레임워크에서 다른 컬렉션과 함께 그래프를 구성할 수 있다.

이처럼 그래프가 컬렉션으로 환원되면, 그래프 계산은 데이터플로우 프레임워크의 조인과 집계 연산으로 자연스럽게 매핑된다.

3.2 Graph Computation as Dataflow Ops

그래프의 국소적 메시지 전달은 관계형 대수의 조인과 집계 연산으로 직관적으로 설명된다.

  • 속성 그래프를 정점 및 에지 속성 컬렉션 쌍으로 정규화하면 분산 데이터 흐름 프레임워크에 그래프를 포함할 수 있다.
  • 분산 데이터 흐름 프레임워크에서 그래프 병렬 계산을 맵 연산에 의해 중단되는 join 단계와 groupBy 단계의 시퀀스로 표현할 수 있다.
  • 조인(join) 단계에서, 정점과 에지 속성은 각 에지와 그에 상응하는 소스 및 목적지 정점 속성으로 구성된 triplets 뷰를 형성하기 위해 조인된다. 
    CREATE VIEW triplets AS
SELECT s.Id, d.Id, s.P, e.P, d.P
FROM edges AS e
JOIN vertices AS s JOIN vertices AS d
ON e.srcId = s.Id AND e.dstId = d.Id

이미지출처 2

  • triplets는 소스 또는 목적지 정점별로 그룹화되어 각 정점의 이웃을 구성하고 집계를 계산한다.
  • 예를 들어, 정점의 페이지랭크를 계산하기 위해 다음과 같이 실행한다.
    • 아래 쿼리를 반복적으로 적용하여 수렴할 때까지 정점 속성을 업데이트함으로써 각 정점의 페이지랭크를 계산할 수 있다. 
      SELECT t.dstId, 0.15+0.85*sum(t.srcP*t.eP)
FROM triplets AS t GROUP BY t.dstId
  • 위 두 단계는 GAS 분해를 보여준다.
    • groupBy 단계는 동일한 정점으로 향하는 메시지를 수집하고, 개입 map 연산은 메시지 합을 적용하여 정점 속성을 업데이트하며, join 단계는 새로운 정점 속성을 인접한 모든 정점으로 분산시킨다.

조인과 그룹바이 패턴만으로 GAS 분해에 해당하는 반복 그래프 계산을 데이터플로우 연산자로 표현한다.

3.3 GraphX Operators

GraphX는 이러한 데이터플로우 기반 그래프 연산을 간결하고 표현력 높은 고수준 오퍼레이터 세트로 노출한다.

  • 그래프 생성자(constructor)는 한 쌍의 정점 및 에지 속성 컬렉션을 논리적으로 결합하여 속성 그래프를 만든다.
    • 또한 무결성(integrity) 제약 조건도 확인한다.

이미지출처 2

  • mrTriplets (Map Reduce Triplets) 연산자는 3.2절에 정의된 그래프 병렬 계산의 필수적인 2단계 프로세스를 인코딩한다.
    • mrTriplets 연산자는 트리플렛 뷰에서 맵 및 그룹별 데이터 흐름 연산자의 구성이다.
  • 사용자 정의 맵 함수는 각 트리플렛에 적용되어 값을 산출하고 사용자 정의 집계 함수를 사용하여 대상 정점에서 집계된다. 
    SELECT t.dstId, reduceF(mapF(t)) AS msgSum
FROM triplets AS t GROUP BY t.dstId
  • mrTriplets 연산자는 목적지 정점 식별자에 의해 키가 지정된 인바운드 메시지의 합계를 포함하는 컬렉션을 생성한다.

image

val graph: Graph[User, Double]
def mapUDF(t: Triplet[User, Double]) =
  if (t.src.age > t.dst.age) 1 else 0
def reduceUDF(a: Int, b: Int): Int = a + b
val seniors: Collection[(Id, Int)] = 
  graph.mrTriplets(mapUDF, reduceUDF)
  • 마지막으로, listing 4에는 정점 또는 에지 컬렉션에서 데이터 흐름 연산을 간단히 수행하는 여러 함수가 포함되어 있다.
// listing 4

// Graph Operators
class Graph[V, E] {
  // 생성자
  def Graph(v: Collection[(Id, V)],
            e: Collection[(Id, Id, E)])
  // 컬렉션 뷰로 초기화
  def vertices: Collection[(Id, V)]
  def edges: Collection[(Id, Id, E)]
  def triplets: Collection[Triplet]
  // 그래프 병렬 계산
  def mrTriplets(f: (Triplet) => M,
      sum: (M, M) => M): Collection[(Id, M)]
  // 다음은 편의 함수들이다.
  def mapV(f: (Id, V) => V): Graph[V, E]
  def mapE(f: (Id, Id, E) => E): Graph[V, E]
  def leftJoinV(v: Collection[(Id, V)],
      f: (Id, V, V) => V): Graph[V, E]
  def leftJoinE(e: Collection[(Id, Id, E)],
      f: (Id, Id, E, E) => E): Graph[V, E]
  def subgraph(vPred: (Id, V) => Boolean,
      ePred: (Triplet) => Boolean): Graph[V, E]
  def reverse: Graph[V, E]
}
  • 예를 들어 mapV는 다음과 같이 정의된다.
g.mapV(f) = Graph(g.vertices.map(f), g.edges)
  • 다음은 GraphX API를 사용하여 프리겔 추상화의 GAS 분해를 구현한다.
// GraphX Enhanced Pregel
def Pregel(g: Graph[V, E],
      vprog: (Id, V, M) => V, // vprog = vertex program
      sendMsg: (Triplet) => M,
      gather: (M, M) => M): Collection[V] = {
  // Set all vertices as active
  g = g.mapV((id, v) => (v, halt=false))
  // Loop until convergence
  while (g.vertices.exists(v => !v.halt)) {
    // Compute the messages
    val msgs: Collection[(Id, M)] =
    // Restrict to edges with active source
    g.subgraph(ePred=(s,d,sP,eP,dP)=>!sP.halt)
    // 활성 정점이 있는 동안 메시지는 mrTriplets 연산자를 사용하여 계산되고,
    // 정점 프로그램은 결과 메시지 합계에 적용된다.
      .mrTriplets(sendMsg, gather)
    // Receive messages and run vertex program
    g = g.leftJoinV(msgs).mapV(vprog)
  }
  return g.vertices
}
  • 다음은 연결 구성 요소(connected components) 알고리즘을 구현하기 위해 프리겔의 GraphX 변형을 사용한다.
    • 연결 구성 요소(connected components)는 각 정점에 대해 도달 가능한 가장 낮은 정점 ID를 계산한다. 
      def ConnectedComp(g: Graph[V, E]) = {
g = g.mapV(v => v.id) // Initialize vertices
def vProg(v: Id, m: Id): Id = {
if (v == m) voteToHalt(v)
return min(v, m)
}
def sendMsg(t: Triplet): Id = 
if (t.src.cc < t.dst.cc) t.src.cc
else None // No message required
def gatherMsg(a: Id, b: Id): Id = min(a,b)
return Pregel(g, vProg, sendMsg, gatherMsg)
}

Combining Graph and Collection Operators

GraphX의 특징은 순수 그래프 알고리즘 수행을 넘어, 일반 컬렉션 오퍼레이터와 그래프 연산자를 서로 조합할 수 있다는 점에 있다.

  • 종종 연결 정점의 그룹은 싱글 정점으로 더 잘 모델링된다.
  • 이러한 경우, 공통 특성(ex. 웹 도메인)을 공유하는 연결(connected) 정점을 집계하여 새로운 그래프(ex. 도메인 그래프)를 도출하는 것을 그래프를 coarsen하게 만드는 데 도움이 될 수 있다.
    • Coarsen: coarsening 연산자는 에지 predicate(=함수)를 만족하는 에지로 연결된 정점을 merge한다.
  • 다음 예제에서 그래프 coarsening을 구현하기 위해 GraphX 추상화를 사용한다. 
    def coarsen(g: Graph[V, E],
          pred: (Id,Id,V,E,V) => Boolean,
          reduce: (V,V) => V) = {
// Restrict graph to contractable edges
val subG = g.subgraph(v => True, pred)
// Compute connected component id for all V
val cc: Collection[(Id, ccId)] = ConnectedComp(subG).vertices
// Merge all vertices in same component
val superV: Collection[(ccId,V)] = g.vertices.leftJoin(cc).groupBy(CC_ID, reduce)
// Link remaining edges between components
val invG = g.subgraph(ePred = t => !pred(t))
val remainingE: Collection[(ccId,ccId,E)] = invG.leftJoin(cc).triplets.map {
  e => (e.src.cc, e.dst.cc, e.attr)
}
// Return the final graph
Graph(superV, remainingE)
}

4. The GraphX System

GraphX의 성능 핵심은 그래프 계산의 특수 패턴을 고전 데이터베이스 최적화 기법으로 치환하여 분산 실행 엔진에 녹여낸 구조에 있다.

  • 이 섹션에서는 indexing, incremental view maintenance, join optimization을 포함한 기존 데이터베이스 시스템의 고전적 기법의 맥락에서 이러한 최적화를 설명한다.

4.1 Distributed Graph Representation

그래프 데이터를 효율적으로 저장하고 질의하기 위해 GraphX는 RDD 위에 특화된 인덱싱과 파티셔닝 구조를 얹는다.

  • GraphX는 내부적으로 스파크 RDD 추상화에 구축된 정점 및 에지 컬렉션 쌍으로 그래프를 나타낸다.
  • 이러한 컬렉션에는 인덱싱(indexing) 및 그래프별 파티셔닝이 RDD 위에 레이어로 도입된다.
  • 다음 그림은 수평으로(horizontally) 파티셔닝된 정점 및 에지 컬렉션과 그 인덱스의 물리적 표현을 보여준다.

image

  • 정점 컬렉션(vertex collection)은 정점 ID에 의해 해시(hash) 파티셔닝된다.
    • 정점 컬렉션 간 빈번한 조인을 지원하기 위해 정점은 각 파티션 내의 로컬 해시 인덱스에 저장된다.
    • 또한 비트마스크(bitmask)는 각 정점의 가시성(visibility)을 저장하여 인덱스 재사용을 촉진하기 위해 부드러운 삭제를 가능하게 한다.
      • 비트마스크(bitmask): bit(0, 1)에 관련된, 즉 정수의 이진수 표현을 활용한 기법이다. 6
  • 에지 컬렉션(edge collection)은 사용자 정의 파티션 함수에 의해 수평으로 파티셔닝된다.
  • GraphX는 소셜 네트워크 및 웹 그래프와 같은 것에서 통신을 최소화하는 정점 절단(vertex-cut) 파티셔닝을 가능하게 한다.
  • 기본적으로 에지는 입력 컬렉션의 파티셔닝에 따라 파티션에 할당된다.
  • 그러나 GraphX는 mrTriplets와 같은 연산자의 통신 복잡성에 대한 강력한 상한(bound)을 가진 2D 해시 파티셔너를 포함하여 다양한 내장 파티셔닝 기능을 제공한다.
  • 에지 배치의 이러한 유연성은 routing table에 의해 활성화된다.
  • 소스 및 목적지 정점에 의한 에지의 효율적인 조회를 위해 파티션 내의 에지는 압축 희소 로우(CSR, compressed sparse row) 표현을 사용하여 소스 정점 ID로 클러스터링되고 목적지 ID에 의해 해시 인덱싱된다.
  • GraphX는 분산 그래프 파티셔닝에 정점 절단 접근법을 채택했다. 7

Index Reuse

반복 연산에서 메모리와 CPU 낭비를 막는 핵심 기법은 RDD의 불변성을 이용한 구조적 인덱스 재사용이다.

  • GraphX는 스파크의 불변성(immutability)을 계승하므로 모든 그래프 연산자는 기존 컬렉션을 수정하지 않고 논리적으로 새로운 컬렉션을 생성한다.
  • 결과적으로 파생된 정점과 에지 컬렉션은 종종 인덱스를 공유하여 메모리 오버헤드를 줄이고 로컬 그래프 작업을 가속화할 수 있다.
    • 예를 들어, 정점의 해시 인덱스는 빠른 집계를 가능하게 하고, 결과 집계는 원래 정점과 인덱스를 공유한다.
  • GraphX 연산자는 인덱스 재사용을 극대화하려고 한다.
  • 그래프 구조를 수정하지 않는 연산자(ex. mapV)는 자동으로 인덱스를 보존한다.
  • 그래프 구조(ex. subgraph)를 제한하는 작업에 인덱스를 재사용하기 위해 GraphX는 비트마스크를 사용하여 제한된 뷰(view)를 구성한다.
  • 인덱스 재사용이 효율성 저하로 이어질 수 있는 경우(ex. 그래프가 고도로 필터링된 경우) GraphX는 reindex 연산자를 사용하여 새 인덱스를 작성한다.

4.2 Implementing the Triplets View

트리플렛 뷰는 그래프 알고리즘 실행의 중추이며, 이를 물리적으로 얼마나 영리하게 구체화하느냐가 시스템 성능을 가른다.

  • 그래프 계산의 핵심 단계는 소스 및 목적지 정점 속성과 에지 속성 사이의 3방향 조인으로 구성된 트리플렛(triplets) 뷰를 구성하고 유지하는 것이다.

물리적인 전체 조인을 피하고 지연 평가와 미러링을 결합하는 것이 핵심이다.

GraphX는 이를 실현하기 위해 4가지 핵심 서브 메커니즘을 유기적으로 결합한다.

Vertex Mirroring

  • 정점 및 에지 속성 컬렉션은 독립적으로 분할되기 때문에 조인에는 데이터 이동이 필요하다.
  • GraphX는 네트워크를 통해 정점 속성을 에지로 전달하여 3방향 조인을 수행하며, 따라서 에지 파티션을 조인 사이트로 설정한다.
  • 이 접근법은 두 가지 이유로 통신을 상당히 감소시킨다.
    1. 에지 우세 그래프 특성: 실제 그래프는 일반적으로 정점보다 훨씬 많은 에지를 갖는다.
    2. 높은 지역성 재사용: 하나의 정점은 동일한 파티션에 많은 에지를 가질 수 있어 정점 속성의 실질적인 재사용을 가능하게 한다.

      Multicast Join

  • 모든 정점이 각 에지 파티션으로 전송되는 브로드캐스트 조인(broadcast join)은 에지 파티션에서 조인이 발생하도록 보장하지만, 대부분의 파티션에서 조인을 완료하기 위해 정점의 서브셋(subset)만 필요하므로 여전히 비효율적일 수 있다.
  • 따라서, GraphX는 각 정점 속성이 인접한 에지를 포함하는 에지 파티션에만 전송되는 멀티캐스트 조인을 도입한다.
    • 각 정점에 대해 GraphX는 인접한 에지가 있는 에지 파티션 집합을 유지한다.
    • 이 조인 사이트 정보는 정점 컬렉션을 가지고 파티셔닝된 라우팅 테이블에 저장된다.
    • 라우팅 테이블(routing table)은 에지 수집과 연관되어 있으며, 트리플렛 뷰의 첫 번째 인스턴스화 시 lazy하게 구성된다.

      Partial Materialization

  • 정점 복제(replication)는 정점 속성이 변경될 때 수행되지만, 중복(duplication)을 피하기 위해 에지 파티션의 로컬 조인은 구체화되지 않은(unmaterialized) 상태로 남아 있다.
  • 대신 미러링된 정점 속성은 각 에지 파티션의 해시 맵에 저장되고 트리플렛 구성 시 참조된다.

    Incremental View Maintenance

  • 반복 그래프 알고리즘은 종종 각 반복에서 정점 속성의 서브셋만 수정한다.
  • 따라서 변경되지 않은 데이터의 불필요한 이동을 방지하기 위해 트리플렛 뷰에 증분 보기 유지(incremental view maintenance)를 적용한다.
  • 각 그래프 작업 후, 트리플렛 뷰가 마지막으로 구성된 이후 변경된 정점 속성을 추적한다.
  • 트리플렛 뷰에 다음에 액세스하면 변경된 정점만 에지 파티션 결합 사이트로 재라우팅되고, 변경되지 않은 정점의 로컬 미러링 값이 재사용된다.
    • 이 기능은 그래프 연산자에 의해 자동으로 관리된다. image

4.3 Optimizations to mrTriplets

가장 빈번하게 호출되는 연산자인 mrTriplets에 대해 GraphX는 데이터베이스 질의 엔진 수준의 추가 최적화를 적용한다.

  • GraphX는 mrTriplets 연산자에 의해 필터링된 인덱스 스캔(filtered index scanning)과 자동 조인 제거(automatic join elimination)라는 두 가지 추가 쿼리 최적화를 통합한다.

    4.3.1 Filtered Index Scanning

  • 애플리케이션은 서브그래프(subgraph) 연산자를 사용하여 그래프를 제한하여 현재 활성 셋을 표현한다.
  • 정점 술어(predicate)는 에지 파티션으로 푸시(push)되며, 여기서 소스 정점 ID의 CSR 인덱스를 사용하여 트리플렛 필터링에 사용할 수 있다.
    • 정점 술어의 선택성(selectivity)을 측정하고 선택성이 0.8 미만일 때 시퀀셜 스캔에서 클러스터 인덱스 스캔으로 전환한다. image

4.3.2 Automatic Join Elimination

  • 일부 경우, 트리플렛 뷰의 작업은 정점 속성 중 하나만 액세스하거나 전혀 액세스하지 못할 수 있다.
  • GraphX는 JVM 바이트 코드 분석기를 사용하여 런타임에 사용자 정의 함수를 검사하고 소스 또는 목적지 정점 속성이 참조되는지 여부를 확인한다.
  • 하나의 속성만 참조되고 트리플렛 뷰가 아직 구체화되지 않은(unmaterialized) 경우, GraphX는 3방향 조인에서 2방향 조인까지 트리플렛 뷰를 생성하기 위한 쿼리 계획을 자동으로 다시 작성한다.
  • 정점 속성이 참조되지 않으면 GraphX는 조인을 완전히 제거한다.
    • 이러한 수정은 트리플렛 뷰가 스파크에서 RDD의 느긋한(lazy) 의미론을 따르기 때문에 가능하다.
  • 사용자가 트리플렛 뷰에 액세스하지 않으면 구체화되지 않는다.
  • 따라서 mrTriplets에 대한 호출은 트리플렛 뷰의 관련 부분을 생성하는 데 필요한 조인을 다시 작성할 수 있다.

바이트코드 분석으로 참조되지 않는 정점 속성에 대해서는 조인 단계를 생략한다. image

핵심 요약 및 시사점

GraphX는 전용 분산 그래프 처리 시스템의 핵심 최적화를 범용 분산 데이터플로우 프레임워크인 Apache Spark 위에서 데이터베이스적 관점으로 재구성한 시스템이다.

  • 그래프와 관계형 대수의 단일 프레임워크 통합: 비정형 데이터 정제, 그래프 알고리즘, 테이블 분석으로 이어지는 전체 파이프라인을 단일 엔진에서 처리하여 별도 그래프 엔진으로의 데이터 이동과 직렬화 병목을 없앨 수 있다.
  • 데이터플로우 연산자로의 그래프 계산 환원: 그래프의 반복적 GAS(Gather-Apply-Scatter) 연산을 조인(Join), 맵(Map), 그룹바이(GroupBy) 데이터플로우 연산자 패턴으로 모델링하여 범용 엔진과의 호환성을 확보했다.
  • 정점 절단 파티셔닝과 라우팅 테이블: 고차원 정점이 존재하는 실제 그래프의 편향을 극복하기 위해 정점 절단 기법과 2D 해시 파티셔닝, 라우팅 테이블 기반의 멀티캐스트 조인을 적용하여 네트워크 통신량을 크게 줄였다.
  • 고전 데이터베이스 기법의 분산 그래프 접목: CSR 인덱스 재사용, 증분 뷰 유지(Incremental View Maintenance), JVM 바이트코드 분석을 통한 자동 조인 제거(Join Elimination) 등 전통적 DB 최적화를 분산 그래프 처리에 접목했다.

References

  1. GONZALEZ, J. E., XIN, R. S., DAVE, A., CRANKSHAW, D., FRANKLIN, M. J., AND STOICA, I. GraphX: Graph Processing in a Distributed Dataflow Framework. OSDI’14, USENIX Association, pp. 599–613. ↩

  2. GraphX - Graph Analytics in Spark ↩ ↩2 ↩3 ↩4

  3. The Property Graph - Spark 3.3.0 ↩

  4. [자료구조] 그래프(Graph)의 개념과 이해, 용어 | 인접행렬 vs 인접리스트 그래프 구현 - 코드연구원 ↩

  5. GONZALEZ, J. E., LOW, Y., GU, H., BICKSON, D., AND GUESTRIN, C. Powergraph: Distributed graph-parallel computation on natural graphs. OSDI’12, USENIX Association, pp. 17–30. ↩

  6. Bit Mask - dob ↩

  7. GraphX Programming Guide - Spark 3.3.0 ↩

댓글남기기

참고