GraphX 프로그래밍 가이드
GraphX 프로그래밍 가이드
GraphX는 Spark에서 그래프와 그래프 병렬 계산을 위한 새 구성 요소예요. 높은 수준에서 GraphX는 새로운 Graph 추상화를 도입해 Spark RDD를 확장해요. 이 그래프는 각 정점(vertex)과 간선(edge)에 속성이 붙은 방향성 다중 그래프(directed multigraph)예요. 그래프 계산을 지원하기 위해 GraphX는 subgraph, joinVertices, aggregateMessages 같은 기본 연산자 집합과 최적화된 Pregel API 변형을 노출해요. 또한 GraphX는 그래프 분석 작업을 단순화하는 그래프 알고리즘과 빌더 컬렉션을 계속 추가하고 있어요.
본문
시작하기(Getting Started)
시작하려면 먼저 Spark와 GraphX를 프로젝트에 import해야 해요. 다음과 같이요.
import org.apache.spark._
import org.apache.spark.graphx._
// To make some of the examples work we will also need RDD
import org.apache.spark.rdd.RDD
Spark 셸을 사용하지 않는다면 SparkContext도 필요해요. Spark 시작 방법에 대해 더 자세히 알고 싶으면 Spark Quick Start Guide를 참고하세요.
프로퍼티 그래프(The Property Graph)
프로퍼티 그래프는 각 정점과 간선에 사용자 정의 객체가 붙는 방향성 다중 그래프예요. 방향성 다중 그래프는 같은 출발·도착 정점을 공유하는 평행 간선(parallel edges)을 여러 개 가질 수 있는 방향성 그래프예요. 평행 간선을 지원하는 능력은 같은 정점들 사이에 여러 관계(예: 동료이자 친구)가 있을 수 있는 모델링 시나리오를 단순화해요. 각 정점은 고유한 64비트 long 식별자(VertexId)로 키가 지정돼요. GraphX는 정점 식별자에 어떤 정렬 제약도 부과하지 않아요. 마찬가지로 간선도 해당하는 출발·도착 정점 식별자를 가져요.
프로퍼티 그래프는 정점(VD)과 간선(ED) 타입에 대해 파라미터화돼요. 이 타입들은 각각 각 정점과 간선에 연결된 객체의 타입이에요.
GraphX는 정점·간선 타입이 원시 데이터 타입(예: int, double 등)일 때 그 표현을 최적화해요. 전용 배열에 저장해 메모리 사용량을 줄이죠.
어떤 경우에는 같은 그래프에 서로 다른 프로퍼티 타입을 가진 정점을 두는 것이 바람직할 수 있어요. 이는 상속으로 해결할 수 있어요. 예를 들어 사용자와 상품을 이분 그래프(bipartite graph)로 모델링하려면 다음과 같이 할 수 있어요.
class VertexProperty()
case class UserProperty(val name: String) extends VertexProperty
case class ProductProperty(val name: String, val price: Double) extends VertexProperty
// The graph might then have the type:
var graph: Graph[VertexProperty, String] = null
RDD처럼 프로퍼티 그래프는 불변(immutable)이고, 분산되어 있으며, 내결함성이 있어요. 그래프의 값이나 구조에 대한 변경은 원하는 변경이 적용된 새 그래프를 생성해 이루어져요. 원래 그래프의 상당 부분(즉, 영향받지 않은 구조, 속성, 인덱스)은 새 그래프에서 재사용되어 이 본질적으로 함수형인 데이터 구조의 비용을 줄여요. 그래프는 일련의 정점 파티셔닝 휴리스틱을 사용해 실행자들 사이에 분할돼요. RDD처럼 그래프의 각 파티션은 실패 시 다른 머신에서 재생성될 수 있어요.
논리적으로 프로퍼티 그래프는 각 정점과 간선의 프로퍼티를 인코딩하는 타입 있는 컬렉션(RDD) 쌍에 대응해요. 결과적으로 그래프 클래스는 그래프의 정점과 간선에 접근하는 멤버를 포함해요.
class Graph[VD, ED] {
val vertices: VertexRDD[VD]
val edges: EdgeRDD[ED]
}
VertexRDD[VD]와 EdgeRDD[ED] 클래스는 각각 RDD[(VertexId, VD)]와 RDD[Edge[ED]]를 확장하고 최적화한 버전이에요. VertexRDD[VD]와 EdgeRDD[ED] 둘 다 그래프 계산 주변에 구축된 추가 기능을 제공하고 내부 최적화를 활용해요. VertexRDDVertexRDD와 EdgeRDDEdgeRDD API는 정점·간선 RDD 섹션에서 더 자세히 다룰게요. 지금은 단순히 RDD[(VertexId, VD)]와 RDD[Edge[ED]] 형태의 RDD라고 생각하면 돼요.
예제 프로퍼티 그래프
GraphX 프로젝트의 다양한 협업자로 구성된 프로퍼티 그래프를 만들고 싶다고 가정해봐요. 정점 프로퍼티는 사용자 이름과 직업을 담을 수 있어요. 간선에는 협업자들 사이의 관계를 설명하는 문자열을 표시할 수 있죠.
결과 그래프는 다음과 같은 타입 시그니처를 가져요.
val userGraph: Graph[(String, String), String]
원시 파일, RDD, 합성 생성기에서 프로퍼티 그래프를 구성하는 방법은 여러 가지가 있고, 이는 그래프 빌더 섹션에서 더 자세히 다뤄요. 아마 가장 일반적인 방법은 Graph 객체를 사용하는 거예요. 예를 들어 다음 코드는 RDD 컬렉션에서 그래프를 구성해요.
// Assume the SparkContext has already been constructed
val sc: SparkContext
// Create an RDD for the vertices
val users: RDD[(VertexId, (String, String))] =
sc.parallelize(Seq((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
(5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))
// Create an RDD for edges
val relationships: RDD[Edge[String]] =
sc.parallelize(Seq(Edge(3L, 7L, "collab"), Edge(5L, 3L, "advisor"),
Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi")))
// Define a default user in case there are relationship with missing user
val defaultUser = ("John Doe", "Missing")
// Build the initial Graph
val graph = Graph(users, relationships, defaultUser)
위 예제에서 Edge 케이스 클래스를 사용해요. 간선은 출발·도착 정점 식별자에 해당하는 srcId와 dstId를 가져요. 게다가 Edge 클래스는 간선 프로퍼티를 저장하는 attr 멤버를 가져요.
graph.vertices와 graph.edges 멤버를 각각 사용해 그래프를 정점·간선 뷰로 분해할 수 있어요.
val graph: Graph[(String, String), String] // Constructed from above
// Count all users which are postdocs
graph.vertices.filter { case (id, (name, pos)) => pos == "postdoc" }.count
// Count all the edges where src > dst
graph.edges.filter(e => e.srcId > e.dstId).count
참고로 graph.vertices는 RDD[(VertexId, (String, String))]를 확장하는 VertexRDD[(String, String)]를 반환하므로 scala case 표현식을 사용해 튜플을 분해해요. 반면 graph.edges는 Edge[String] 객체를 포함하는 EdgeRDD를 반환해요. 다음과 같이 케이스 클래스 타입 생성자도 사용할 수 있었어요.
graph.edges.filter { case Edge(src, dst, prop) => src > dst }.count
프로퍼티 그래프의 정점·간선 뷰 외에도 GraphX는 트리플렛(triplet) 뷰를 노출해요. 트리플렛 뷰는 정점과 간선 프로퍼티를 논리적으로 조인해 EdgeTriplet 클래스 인스턴스를 포함하는 RDD[EdgeTriplet[VD, ED]]를 만들어요. 이 조인은 다음 SQL 표현식으로 나타낼 수 있어요.
SELECT src.id, dst.id, src.attr, e.attr, dst.attr
FROM edges AS e LEFT JOIN vertices AS src, vertices AS dst
ON e.srcId = src.Id AND e.dstId = dst.Id
혹은 그래프로 나타낼 수도 있어요. EdgeTriplet 클래스는 출발·도착 프로퍼티를 각각 담는 srcAttr와 dstAttr 멤버를 추가해 Edge 클래스를 확장해요. 그래프의 트리플렛 뷰를 사용해 사용자들 사이의 관계를 설명하는 문자열 컬렉션을 렌더링할 수 있어요.
val graph: Graph[(String, String), String] // Constructed from above
// Use the triplets view to create an RDD of facts.
val facts: RDD[String] =
graph.triplets.map(triplet =>
triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1)
facts.collect.foreach(println(_))
그래프 연산자(Graph Operators)
RDD에 map, filter, reduceByKey 같은 기본 연산이 있듯이, 프로퍼티 그래프도 사용자 정의 함수를 받아 변환된 프로퍼티와 구조를 가진 새 그래프를 만드는 기본 연산자 컬렉션을 가져요. 최적화된 구현이 있는 핵심 연산자는 Graph에 정의돼 있고, 핵심 연산자의 구성으로 표현되는 편리한 연산자는 GraphOps에 정의돼 있어요. Scala 암시(implicits) 덕분에 GraphOps의 연산자는 Graph의 멤버로 자동 제공돼요. 예를 들어 각 정점의 in-degree(GraphOps에 정의)를 다음과 같이 계산할 수 있어요.
val graph: Graph[(String, String), String]
// Use the implicit GraphOps.inDegrees operator
val inDegrees: VertexRDD[Int] = graph.inDegrees
핵심 그래프 연산과 GraphOps를 구분하는 이유는 미래에 서로 다른 그래프 표현을 지원하기 위해서예요. 각 그래프 표현은 핵심 연산의 구현을 제공해야 하고 GraphOps에 정의된 많은 유용한 연산을 재사용해요.
연산자 요약 목록
다음은 Graph와 GraphOps 둘 다에 정의된 기능을 간단히 요약한 목록이에요. 단순화를 위해 Graph의 멤버로 제시해요. 일부 함수 시그니처는 단순화됐고(예: 기본 인자와 타입 제약 제거) 일부 고급 기능은 제거됐으니, 공식 연산 목록은 API 문서를 참고하세요.
/** Summary of the functionality in the property graph */
class Graph[VD, ED] {
// Information about the Graph ===================================================================
val numEdges: Long
val numVertices: Long
val inDegrees: VertexRDD[Int]
val outDegrees: VertexRDD[Int]
val degrees: VertexRDD[Int]
// Views of the graph as collections =============================================================
val vertices: VertexRDD[VD]
val edges: EdgeRDD[ED]
val triplets: RDD[EdgeTriplet[VD, ED]]
// Functions for caching graphs ==================================================================
def persist(newLevel: StorageLevel = StorageLevel.MEMORY_ONLY): Graph[VD, ED]
def cache(): Graph[VD, ED]
def unpersistVertices(blocking: Boolean = false): Graph[VD, ED]
// Change the partitioning heuristic ============================================================
def partitionBy(partitionStrategy: PartitionStrategy): Graph[VD, ED]
// Transform vertex and edge attributes ==========================================================
def mapVertices[VD2](map: (VertexId, VD) => VD2): Graph[VD2, ED]
def mapEdges[ED2](map: Edge[ED] => ED2): Graph[VD, ED2]
def mapEdges[ED2](map: (PartitionID, Iterator[Edge[ED]]) => Iterator[ED2]): Graph[VD, ED2]
def mapTriplets[ED2](map: EdgeTriplet[VD, ED] => ED2): Graph[VD, ED2]
def mapTriplets[ED2](map: (PartitionID, Iterator[EdgeTriplet[VD, ED]]) => Iterator[ED2])
: Graph[VD, ED2]
// Modify the graph structure ====================================================================
def reverse: Graph[VD, ED]
def subgraph(
epred: EdgeTriplet[VD,ED] => Boolean = (x => true),
vpred: (VertexId, VD) => Boolean = ((v, d) => true))
: Graph[VD, ED]
def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]
def groupEdges(merge: (ED, ED) => ED): Graph[VD, ED]
// Join RDDs with the graph ======================================================================
def joinVertices[U](table: RDD[(VertexId, U)])(mapFunc: (VertexId, VD, U) => VD): Graph[VD, ED]
def outerJoinVertices[U, VD2](other: RDD[(VertexId, U)])
(mapFunc: (VertexId, VD, Option[U]) => VD2)
: Graph[VD2, ED]
// Aggregate information about adjacent triplets =================================================
def collectNeighborIds(edgeDirection: EdgeDirection): VertexRDD[Array[VertexId]]
def collectNeighbors(edgeDirection: EdgeDirection): VertexRDD[Array[(VertexId, VD)]]
def aggregateMessages[Msg: ClassTag](
sendMsg: EdgeContext[VD, ED, Msg] => Unit,
mergeMsg: (Msg, Msg) => Msg,
tripletFields: TripletFields = TripletFields.All)
: VertexRDD[A]
// Iterative graph-parallel computation ==========================================================
def pregel[A](initialMsg: A, maxIterations: Int, activeDirection: EdgeDirection)(
vprog: (VertexId, VD, A) => VD,
sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)],
mergeMsg: (A, A) => A)
: Graph[VD, ED]
// Basic graph algorithms ========================================================================
def pageRank(tol: Double, resetProb: Double = 0.15): Graph[Double, Double]
def connectedComponents(): Graph[VertexId, ED]
def triangleCount(): Graph[Int, ED]
def stronglyConnectedComponents(numIter: Int): Graph[VertexId, ED]
}
프로퍼티 연산자(Property Operators)
RDD map 연산자처럼, 프로퍼티 그래프는 다음을 포함해요.
class Graph[VD, ED] {
def mapVertices[VD2](map: (VertexId, VD) => VD2): Graph[VD2, ED]
def mapEdges[ED2](map: Edge[ED] => ED2): Graph[VD, ED2]
def mapTriplets[ED2](map: EdgeTriplet[VD, ED] => ED2): Graph[VD, ED2]
}
이 각 연산자는 사용자 정의 map 함수로 정점 또는 간선 프로퍼티가 수정된 새 그래프를 만들어요.
각 경우에 그래프 구조는 영향받지 않는다는 점에 유의하세요. 이것은 결과 그래프가 원래 그래프의 구조적 인덱스를 재사용할 수 있게 해주는 이 연산자의 핵심 기능이에요. 다음 스니펫은 논리적으로 동등하지만, 첫 번째는 구조적 인덱스를 보존하지 않아 GraphX 시스템 최적화의 이점을 얻지 못해요.
val newVertices = graph.vertices.map { case (id, attr) => (id, mapUdf(id, attr)) }
val newGraph = Graph(newVertices, graph.edges)
대신 mapVertices를 사용해 인덱스를 보존하세요.
val newGraph = graph.mapVertices((id, attr) => mapUdf(id, attr))
이 연산자들은 종종 특정 계산을 위해 그래프를 초기화하거나 불필요한 프로퍼티를 걸러내는 데 사용돼요. 예를 들어 정점 프로퍼티로 out degree를 가진 그래프(나중에 그런 그래프를 만드는 방법을 설명할게요)가 주어졌을 때, PageRank를 위해 초기화해요.
// Given a graph where the vertex property is the out degree
val inputGraph: Graph[Int, String] =
graph.outerJoinVertices(graph.outDegrees)((vid, _, degOpt) => degOpt.getOrElse(0))
// Construct a graph where each edge contains the weight
// and each vertex is the initial PageRank
val outputGraph: Graph[Double, Double] =
inputGraph.mapTriplets(triplet => 1.0 / triplet.srcAttr).mapVertices((id, _) => 1.0)
구조 연산자(Structural Operators)
현재 GraphX는 흔히 쓰이는 간단한 구조 연산자 집합만 지원하고, 미래에 더 추가할 예정이에요. 다음은 기본 구조 연산자 목록이에요.
class Graph[VD, ED] {
def reverse: Graph[VD, ED]
def subgraph(epred: EdgeTriplet[VD,ED] => Boolean,
vpred: (VertexId, VD) => Boolean): Graph[VD, ED]
def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]
def groupEdges(merge: (ED, ED) => ED): Graph[VD,ED]
}
reverse 연산자는 모든 간선 방향이 뒤집힌 새 그래프를 반환해요. 예를 들어 역 PageRank(inverse PageRank)를 계산할 때 유용할 수 있어요. reverse 연산은 정점·간선 프로퍼티를 수정하지 않고 간선 수를 바꾸지도 않으므로, 데이터 이동이나 복제 없이 효율적으로 구현할 수 있어요.
subgraph 연산자는 정점·간선 술어를 받아, 정점 술어를 만족(참으로 평가)하는 정점과, 간선 술어를 만족하면서 정점 술어를 만족하는 정점을 연결하는 간선만 포함하는 그래프를 반환해요. subgraph 연산자는 그래프를 관심 있는 정점·간선으로 제한하거나 깨진 링크를 제거하는 여러 상황에 쓸 수 있어요. 예를 들어 다음 코드에서 깨진 링크를 제거해요.
// Create an RDD for the vertices
val users: RDD[(VertexId, (String, String))] =
sc.parallelize(Seq((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),
(5L, ("franklin", "prof")), (2L, ("istoica", "prof")),
(4L, ("peter", "student"))))
// Create an RDD for edges
val relationships: RDD[Edge[String]] =
sc.parallelize(Seq(Edge(3L, 7L, "collab"), Edge(5L, 3L, "advisor"),
Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi"),
Edge(4L, 0L, "student"), Edge(5L, 0L, "colleague")))
// Define a default user in case there are relationship with missing user
val defaultUser = ("John Doe", "Missing")
// Build the initial Graph
val graph = Graph(users, relationships, defaultUser)
// Notice that there is a user 0 (for which we have no information) connected to users
// 4 (peter) and 5 (franklin).
graph.triplets.map(
triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
).collect.foreach(println(_))
// Remove missing vertices as well as the edges to connected to them
val validGraph = graph.subgraph(vpred = (id, attr) => attr._2 != "Missing")
// The valid subgraph will disconnect users 4 and 5 by removing user 0
validGraph.vertices.collect.foreach(println(_))
validGraph.triplets.map(
triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
).collect.foreach(println(_))
위 예제에서는 정점 술어만 제공됐어요. subgraph 연산자는 정점·간선 술어가 제공되지 않으면 기본값으로 true를 사용해요.
mask 연산자는 입력 그래프에도 있는 정점과 간선을 포함하는 그래프를 반환해 부분 그래프를 구성해요. 이는 subgraph 연산자와 함께 사용해 다른 관련 그래프의 프로퍼티를 기준으로 그래프를 제한할 수 있어요. 예를 들어 누락된 정점이 있는 그래프로 connected components를 실행한 뒤 답을 유효한 부분 그래프로 제한할 수 있어요.
// Run Connected Components
val ccGraph = graph.connectedComponents() // No longer contains missing field
// Remove missing vertices as well as the edges to connected to them
val validGraph = graph.subgraph(vpred = (id, attr) => attr._2 != "Missing")
// Restrict the answer to the valid subgraph
val validCCGraph = ccGraph.mask(validGraph)
groupEdges 연산자는 다중 그래프의 평행 간선(즉, 정점 쌍 사이의 중복 간선)을 병합해요. 많은 수치 애플리케이션에서 평행 간선은 더해져(가중치가 결합되어) 단일 간선이 되고, 그래프 크기가 줄어들 수 있어요.
조인 연산자(Join Operators)
많은 경우 외부 컬렉션(RDD)의 데이터를 그래프와 조인해야 해요. 예를 들어 기존 그래프에 병합하고 싶은 추가 사용자 프로퍼티가 있거나, 한 그래프에서 다른 그래프로 정점 프로퍼티를 가져오고 싶을 수 있어요. 이런 작업은 조인 연산자로 수행할 수 있어요. 아래에 핵심 조인 연산자를 나열해요.
class Graph[VD, ED] {
def joinVertices[U](table: RDD[(VertexId, U)])(map: (VertexId, VD, U) => VD)
: Graph[VD, ED]
def outerJoinVertices[U, VD2](table: RDD[(VertexId, U)])(map: (VertexId, VD, Option[U]) => VD2)
: Graph[VD2, ED]
}
joinVertices 연산자는 정점을 입력 RDD와 조인하고, 조인된 정점의 결과에 사용자 정의 map 함수를 적용해 얻은 정점 프로퍼티를 가진 새 그래프를 반환해요. RDD에 일치하는 값이 없는 정점은 원래 값을 유지해요.
RDD가 주어진 정점에 대해 값이 두 개 이상 포함되면 하나만 사용된다는 점에 유의하세요. 따라서 입력 RDD는 다음을 사용해 고유하게 만드는 것이 좋아요. 이렇게 하면 결과 값이 사전 인덱싱되어 이후의 조인이 크게 빨라져요.
val nonUniqueCosts: RDD[(VertexId, Double)]
val uniqueCosts: VertexRDD[Double] =
graph.vertices.aggregateUsingIndex(nonUnique, (a,b) => a + b)
val joinedGraph = graph.joinVertices(uniqueCosts)(
(id, oldCost, extraCost) => oldCost + extraCost)
더 일반적인 outerJoinVertices는 사용자 정의 map 함수가 모든 정점에 적용되고 정점 프로퍼티 타입을 바꿀 수 있다는 점만 제외하면 joinVertices와 비슷하게 동작해요. 모든 정점이 입력 RDD에 일치하는 값을 가지는 것은 아니므로 map 함수는 Option 타입을 받아요. 예를 들어 정점 프로퍼티를 outDegree로 초기화해 PageRank용 그래프를 설정할 수 있어요.
val outDegrees: VertexRDD[Int] = graph.outDegrees
val degreeGraph = graph.outerJoinVertices(outDegrees) { (id, oldAttr, outDegOpt) =>
outDegOpt match {
case Some(outDeg) => outDeg
case None => 0 // No outDegree means zero outDegree
}
}
위 예제에서 사용된 커리(curried) 함수 패턴의 여러 파라미터 목록(예: f(a)(b))을 알아차렸을 거예요. f(a)(b)를 f(a,b)로 쓸 수도 있지만, 그러면 b에 대한 타입 추론이 a에 의존하지 않게 돼요. 결과적으로 사용자는 사용자 정의 함수에 타입 어노테이션을 제공해야 해요.
val joinedGraph = graph.joinVertices(uniqueCosts,
(id: VertexId, oldCost: Double, extraCost: Double) => oldCost + extraCost)
이웃 집계(Neighborhood Aggregation)
많은 그래프 분석 작업의 핵심 단계는 각 정점의 이웃에 대한 정보를 집계하는 것이에요. 예를 들어 각 사용자가 가진 팔로워 수나 각 사용자의 팔로워 평균 나이를 알고 싶을 수 있어요. PageRank, 최단 경로, connected components 같은 많은 반복 그래프 알고리즘은 이웃 정점의 프로퍼티(예: 현재 PageRank 값, 소스까지의 최단 경로, 도달 가능한 가장 작은 정점 id)를 반복적으로 집계해요.
성능을 개선하기 위해 기본 집계 연산자가 graph.mapReduceTriplets에서 새 graph.AggregateMessages로 바뀌었어요. API 변경은 비교적 작지만, 아래에 전환 가이드를 제공할게요.
Aggregate Messages(aggregateMessages)
GraphX의 핵심 집계 연산은 aggregateMessages예요. 이 연산자는 사용자 정의 sendMsg 함수를 그래프의 각 EdgeTriplet에 적용한 뒤, mergeMsg 함수로 그 메시지들을 목적지 정점에서 집계해요.
class Graph[VD, ED] {
def aggregateMessages[Msg: ClassTag](
sendMsg: EdgeContext[VD, ED, Msg] => Unit,
mergeMsg: (Msg, Msg) => Msg,
tripletFields: TripletFields = TripletFields.All)
: VertexRDD[Msg]
}
사용자 정의 sendMsg 함수는 EdgeContext를 받아요. 이 컨텍스트는 출발·도착 속성과 간선 속성을 노출하고, 출발·도착 속성으로 메시지를 보내는 함수(sendToSrc, sendToDst)를 제공해요. sendMsg를 map-reduce의 map 함수로 생각하면 돼요. 사용자 정의 mergeMsg 함수는 같은 정점으로 향하는 두 메시지를 받아 단일 메시지를 만들어요. mergeMsg를 map-reduce의 reduce 함수로 생각하면 돼요. aggregateMessages 연산자는 각 정점으로 향하는 집계 메시지(타입 Msg)를 포함하는 VertexRDD[Msg]를 반환해요. 메시지를 받지 못한 정점은 반환된 VertexRDDVertexRDD에 포함되지 않아요.
[EdgeContext]는 추가적인 ([sendToSrc], [sendToDst])를 노출하기 위해 [EdgeTriplet] 대신 제공되는데, GraphX는 이를 이용해 메시지 라우팅을 최적화해요.
게다가 aggregateMessages는 선택적으로 tripletsFields를 받는데, 이것은 EdgeContext에서 어떤 데이터에 접근하는지(예: 도착 정점 속성이 아닌 출발 정점 속성)를 나타내요. tripletsFields의 가능한 옵션은 TripletFields에 정의돼 있고, 기본값은 사용자 정의 sendMsg 함수가 EdgeContext의 어떤 필드에도 접근할 수 있다는 뜻인 TripletFields.All이에요. tripletFields 인자는 GraphX에 EdgeContext의 일부만 필요하다고 알려 최적화된 조인 전략을 선택하게 할 수 있어요. 예를 들어 각 사용자의 팔로워 평균 나이를 계산한다면 출발 필드만 필요하므로 TripletFields.Src를 사용해 출발 필드만 필요하다고 나타낼 수 있어요.
이전 GraphX 버전에서는 TripletFields를 추론하기 위해 바이트코드 검사를 사용했지만, 바이트코드 검사는 약간 신뢰할 수 없다는 것을 발견해 대신 더 명시적인 사용자 제어를 선택했어요.
다음 예제에서는 aggregateMessages 연산자를 사용해 각 사용자의 나이가 더 많은 팔로워의 평균 나이를 계산해요.
import org.apache.spark.graphx.{Graph, VertexRDD}
import org.apache.spark.graphx.util.GraphGenerators
// Create a graph with "age" as the vertex property.
// Here we use a random graph for simplicity.
val graph: Graph[Double, Int] =
GraphGenerators.logNormalGraph(sc, numVertices = 100).mapVertices( (id, _) => id.toDouble )
// Compute the number of older followers and their total age
val olderFollowers: VertexRDD[(Int, Double)] = graph.aggregateMessages[(Int, Double)](
triplet => { // Map Function
if (triplet.srcAttr > triplet.dstAttr) {
// Send message to destination vertex containing counter and age
triplet.sendToDst((1, triplet.srcAttr))
}
},
// Add counter and age
(a, b) => (a._1 + b._1, a._2 + b._2) // Reduce Function
)
// Divide total age by number of older followers to get average age of older followers
val avgAgeOfOlderFollowers: VertexRDD[Double] =
olderFollowers.mapValues( (id, value) =>
value match { case (count, totalAge) => totalAge / count } )
// Display the results
avgAgeOfOlderFollowers.collect().foreach(println(_))
전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/graphx/AggregateMessagesExample.scala"에서 찾을 수 있어요.
aggregateMessages 연산은 메시지(그리고 메시지의 합) 크기가 일정할 때(예: 리스트와 연결 대신 float와 덧셈) 최적으로 수행돼요.
Map Reduce Triplets 전환 가이드(레거시)
GraphX의 이전 버전에서 이웃 집계는 mapReduceTriplets 연산자로 수행됐어요.
class Graph[VD, ED] {
def mapReduceTriplets[Msg](
map: EdgeTriplet[VD, ED] => Iterator[(VertexId, Msg)],
reduce: (Msg, Msg) => Msg)
: VertexRDD[Msg]
}
mapReduceTriplets 연산자는 각 트리플렛에 적용되는 사용자 정의 map 함수를 받고, 사용자 정의 reduce 함수로 집계되는 메시지를 만들 수 있어요. 하지만 반환된 iterator의 사용이 비싸고 추가 최적화(예: 로컬 정점 재번호)를 적용하는 능력을 억제한다는 것을 발견했어요. aggregateMessages에서 트리플렛 필드를 노출하고 출발·도착 정점으로 메시지를 명시적으로 보내는 함수도 제공하는 EdgeContext를 도입했어요. 게다가 바이트코드 검사를 제거하고 대신 사용자가 트리플렛의 어떤 필드가 실제로 필요한지 나타내도록 했어요.
mapReduceTriplets를 사용하는 다음 코드 블록은,
val graph: Graph[Int, Float] = ...
def msgFun(triplet: Triplet[Int, Float]): Iterator[(Int, String)] = {
Iterator((triplet.dstId, "Hi"))
}
def reduceFun(a: String, b: String): String = a + " " + b
val result = graph.mapReduceTriplets[String](msgFun, reduceFun)
aggregateMessages로 다음과 같이 다시 쓸 수 있어요.
val graph: Graph[Int, Float] = ...
def msgFun(triplet: EdgeContext[Int, Float, String]) {
triplet.sendToDst("Hi")
}
def reduceFun(a: String, b: String): String = a + " " + b
val result = graph.aggregateMessages[String](msgFun, reduceFun)
차수 정보 계산(Computing Degree Information)
흔한 집계 작업은 각 정점의 차수(degree), 즉 각 정점에 인접한 간선 수를 계산하는 것이에요. 방향성 그래프의 맥락에서는 각 정점의 in-degree, out-degree, total degree를 아는 것이 자주 필요해요. GraphOps 클래스는 각 정점의 차수를 계산하는 연산자 컬렉션을 포함해요. 예를 들어 다음에서 max in, out, total degree를 계산해요.
// Define a reduce operation to compute the highest degree vertex
def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId, Int) = {
if (a._2 > b._2) a else b
}
// Compute the max degrees
val maxInDegree: (VertexId, Int) = graph.inDegrees.reduce(max)
val maxOutDegree: (VertexId, Int) = graph.outDegrees.reduce(max)
val maxDegrees: (VertexId, Int) = graph.degrees.reduce(max)
이웃 수집(Collecting Neighbors)
어떤 경우에는 각 정점에서 이웃 정점과 그 속성을 수집해 계산을 표현하는 것이 더 쉬울 수 있어요. 이는 collectNeighborIds와 collectNeighbors 연산자로 쉽게 해결돼요.
class GraphOps[VD, ED] {
def collectNeighborIds(edgeDirection: EdgeDirection): VertexRDD[Array[VertexId]]
def collectNeighbors(edgeDirection: EdgeDirection): VertexRDD[ Array[(VertexId, VD)] ]
}
이 연산자들은 정보를 복제하고 상당한 통신을 필요로 하므로 비용이 꽤 들 수 있어요. 가능하면 aggregateMessages 연산자로 같은 계산을 직접 표현해 보세요.
캐싱과 언캐싱(Caching and Uncaching)
Spark에서 RDD는 기본적으로 메모리에 영속화되지 않아요. 재계산을 피하려면 여러 번 사용할 때 명시적으로 캐시해야 해요(Spark Programming Guide 참고). GraphX의 그래프도 똑같이 동작해요. 그래프를 여러 번 사용할 때는 반드시 먼저 Graph.cache()를 호출하세요.
반복 계산에서 최고 성능을 위해 언캐싱도 필요할 수 있어요. 기본적으로 캐시된 RDD와 그래프는 메모리 압박이 LRU 순서로 퇴거하도록 강제할 때까지 메모리에 남아요. 반복 계산에서 이전 반복의 중간 결과가 캐시를 채워요. 결국 퇴거되긴 하지만, 메모리에 저장된 불필요한 데이터는 가비지 컬렉션을 느리게 해요. 중간 결과가 더 이상 필요 없어지면 즉시 언캐싱하는 것이 더 효율적이에요. 이는 매 반복마다 그래프나 RDD를 구체화(materialize, 캐시하고 강제로)하고, 다른 모든 데이터셋을 언캐싱하고, 미래 반복에서 구체화된 데이터셋만 사용하는 것을 포함해요. 하지만 그래프는 여러 RDD로 구성되므로 올바르게 unpersist하기 어려울 수 있어요. 반복 계산에는 중간 결과를 올바르게 unpersist하는 Pregel API 사용을 권장해요.
Pregel API
그래프는 본질적으로 재귀적인 데이터 구조예요. 정점의 프로퍼티가 이웃의 프로퍼티에 의존하고, 이웃의 프로퍼티는 그들의 이웃의 프로퍼티에 의존하니까요. 결과적으로 많은 중요한 그래프 알고리즘은 고정점(fixed-point) 조건에 도달할 때까지 각 정점의 프로퍼티를 반복적으로 다시 계산해요. 이런 반복 알고리즘을 표현하기 위해 다양한 그래프 병렬 추상화가 제안됐어요. GraphX는 Pregel API의 변형을 노출해요.
높은 수준에서 GraphX의 Pregel 연산자는 그래프의 토폴로지에 제약된 벌크 동기 병렬(bulk-synchronous parallel) 메시징 추상화예요. Pregel 연산자는 일련의 슈퍼 스텝(super steps)으로 실행되는데, 각 슈퍼 스텝에서 정점은 이전 슈퍼 스텝의 인바운드 메시지의 합을 받고, 정점 프로퍼티의 새 값을 계산한 뒤, 다음 슈퍼 스텝에서 이웃 정점으로 메시지를 보내요. Pregel과 달리 메시지는 간선 트리플렛의 함수로 병렬로 계산되며, 메시지 계산은 출발·도착 정점 속성 모두에 접근할 수 있어요. 메시지를 받지 못한 정점은 슈퍼 스텝 내에서 건너뛰어져요. Pregel 연산자는 남은 메시지가 없을 때 반복을 종료하고 최종 그래프를 반환해요.
참고로, 더 표준적인 Pregel 구현과 달리 GraphX의 정점은 이웃 정점으로만 메시지를 보낼 수 있고, 메시지 구성은 사용자 정의 메시징 함수로 병렬로 수행돼요. 이 제약들은 GraphX 내에서 추가 최적화를 가능하게 해요.
다음은 Pregel 연산자의 타입 시그니처와 구현 스케치예요 (참고: 긴 계보 체인으로 인한 stackOverflowError를 피하기 위해, "spark.graphx.pregel.checkpointInterval"을 예를 들어 10 같은 양수로 설정해 pregel이 주기적으로 그래프와 메시지를 체크포인트하도록 하고, SparkContext.setCheckpointDir(directory: String)로 체크포인트 디렉터리도 설정해야 해요):
class GraphOps[VD, ED] {
def pregel[A]
(initialMsg: A,
maxIter: Int = Int.MaxValue,
activeDir: EdgeDirection = EdgeDirection.Out)
(vprog: (VertexId, VD, A) => VD,
sendMsg: EdgeTriplet[VD, ED] => Iterator[(VertexId, A)],
mergeMsg: (A, A) => A)
: Graph[VD, ED] = {
// Receive the initial message at each vertex
var g = mapVertices( (vid, vdata) => vprog(vid, vdata, initialMsg) ).cache()
// compute the messages
var messages = GraphXUtils.mapReduceTriplets(g, sendMsg, mergeMsg)
var activeMessages = messages.count()
// Loop until no messages remain or maxIterations is achieved
var i = 0
while (activeMessages > 0 && i < maxIterations) {
// Receive the messages and update the vertices.
g = g.joinVertices(messages)(vprog).cache()
val oldMessages = messages
// Send new messages, skipping edges where neither side received a message. We must cache
// messages so it can be materialized on the next line, allowing us to uncache the previous
// iteration.
messages = GraphXUtils.mapReduceTriplets(
g, sendMsg, mergeMsg, Some((oldMessages, activeDirection))).cache()
activeMessages = messages.count()
i += 1
}
g
}
}
Pregel은 두 개의 인자 목록을 받는 것에 주목하세요(즉, graph.pregel(list1)(list2)). 첫 번째 인자 목록은 초기 메시지, 최대 반복 횟수, 메시지를 보낼 간선 방향(기본적으로 out 간선을 따라)을 포함한 구성 파라미터를 담아요. 두 번째 인자 목록은 메시지 수신(정점 프로그램 vprog), 메시지 계산(sendMsg), 메시지 결합(mergeMsg)을 위한 사용자 정의 함수를 담아요.
Pregel 연산자를 사용해 단일 소스 최단 경로 같은 계산을 다음 예제처럼 표현할 수 있어요.
import org.apache.spark.graphx.{Graph, VertexId}
import org.apache.spark.graphx.util.GraphGenerators
// A graph with edge attributes containing distances
val graph: Graph[Long, Double] =
GraphGenerators.logNormalGraph(sc, numVertices = 100).mapEdges(e => e.attr.toDouble)
val sourceId: VertexId = 42 // The ultimate source
// Initialize the graph such that all vertices except the root have distance infinity.
val initialGraph = graph.mapVertices((id, _) =>
if (id == sourceId) 0.0 else Double.PositiveInfinity)
val sssp = initialGraph.pregel(Double.PositiveInfinity)(
(id, dist, newDist) => math.min(dist, newDist), // Vertex Program
triplet => { // Send Message
if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
} else {
Iterator.empty
}
},
(a, b) => math.min(a, b) // Merge Message
)
println(sssp.vertices.collect().mkString("\n"))
전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/graphx/SSSPExample.scala"에서 찾을 수 있어요.
그래프 빌더(Graph Builders)
GraphX는 RDD 또는 디스크에서 정점·간선 컬렉션으로 그래프를 만드는 여러 방법을 제공해요. 어떤 그래프 빌더도 기본적으로 그래프의 간선을 다시 파티셔닝하지 않아요. 대신 간선은 기본 파티션(예: HDFS의 원래 블록)에 남아요. Graph.groupEdges는 동일한 간선이 같은 파티션에 함께 위치한다고 가정하므로 그래프를 다시 파티셔닝해야 해요. 그래서 groupEdges를 호출하기 전에 반드시 Graph.partitionBy를 호출해야 해요.
object GraphLoader {
def edgeListFile(
sc: SparkContext,
path: String,
canonicalOrientation: Boolean = false,
minEdgePartitions: Int = 1)
: Graph[Int, Int]
}
GraphLoader.edgeListFile는 디스크의 간선 목록에서 그래프를 로드하는 방법을 제공해요. #로 시작하는 주석 줄을 건너뛰면서 다음과 같은 형태의 (출발 정점 ID, 도착 정점 ID) 쌍 인접 목록을 파싱해요.
# This is a comment
2 1
4 1
1 2
지정된 간선에서 Graph를 만들고, 간선이 언급한 모든 정점을 자동 생성해요. 모든 정점·간선 속성은 기본적으로 1이에요. canonicalOrientation 인자는 간선을 양의 방향(srcId < dstId)으로 재정렬할 수 있게 하는데, 이는 connected components 알고리즘에 필요해요. minEdgePartitions 인자는 생성할 최소 간선 파티션 수를 지정해요. 예를 들어 HDFS 파일이 더 많은 블록을 가지면 지정한 수보다 더 많은 간선 파티션이 있을 수 있어요.
object Graph {
def apply[VD, ED](
vertices: RDD[(VertexId, VD)],
edges: RDD[Edge[ED]],
defaultVertexAttr: VD = null)
: Graph[VD, ED]
def fromEdges[VD, ED](
edges: RDD[Edge[ED]],
defaultValue: VD): Graph[VD, ED]
def fromEdgeTuples[VD](
rawEdges: RDD[(VertexId, VertexId)],
defaultValue: VD,
uniqueEdges: Option[PartitionStrategy] = None): Graph[VD, Int]
}
Graph.apply는 정점·간선 RDD에서 그래프를 만들 수 있게 해요. 중복 정점은 임의로 선택되고, 간선 RDD에는 있지만 정점 RDD에는 없는 정점은 기본 속성이 할당돼요.
Graph.fromEdges는 간선 RDD만으로 그래프를 만들 수 있게 해요. 간선이 언급한 모든 정점을 자동 생성하고 기본값을 할당해요.
Graph.fromEdgeTuples는 간선 튜플 RDD만으로 그래프를 만들 수 있게 해요. 간선에 값 1을 할당하고, 간선이 언급한 모든 정점을 자동 생성하며 기본값을 할당해요. 또한 간선 중복 제거를 지원해요. 중복을 제거하려면 PartitionStrategy의 Some을 uniqueEdges 파라미터로 전달해요(예: uniqueEdges = Some(PartitionStrategy.RandomVertexCut)). 동일한 간선을 같은 파티션에 함께 위치시켜 중복을 제거할 수 있으려면 파티션 전략이 필요해요.
정점·간선 RDD(Vertex and Edge RDDs)
GraphX는 그래프 안에 저장된 정점·간선의 RDD 뷰를 노출해요. 하지만 GraphX는 정점·간선을 최적화된 데이터 구조로 유지하고, 이 데이터 구조가 추가 기능을 제공하므로, 정점·간선은 각각 VertexRDDVertexRDD와 EdgeRDDEdgeRDD로 반환돼요. 이 섹션에서는 이 타입들의 추가 유용한 기능 중 일부를 살펴볼게요. 이건 불완전한 목록이므로 공식 연산 목록은 API 문서를 참고하세요.
VertexRDD
VertexRDD[A]는 RDD[(VertexId, A)]를 확장하고 각 VertexId가 한 번만 나타난다는 추가 제약을 가져요. 게다가 VertexRDD[A]는 각각 타입 A의 속성을 가진 정점 집합을 나타내요. 내부적으로 이는 정점 속성을 재사용 가능한 hash-map 데이터 구조에 저장해 달성돼요. 결과적으로 두 VertexRDD가 같은 기본 VertexRDDVertexRDD에서 파생됐다면(예: filter나 mapValues로), 해시 평가 없이 상수 시간에 조인될 수 있어요. 이 인덱싱된 데이터 구조를 활용하기 위해 VertexRDDVertexRDD는 다음 추가 기능을 노출해요.
class VertexRDD[VD] extends RDD[(VertexId, VD)] {
// Filter the vertex set but preserves the internal index
def filter(pred: Tuple2[VertexId, VD] => Boolean): VertexRDD[VD]
// Transform the values without changing the ids (preserves the internal index)
def mapValues[VD2](map: VD => VD2): VertexRDD[VD2]
def mapValues[VD2](map: (VertexId, VD) => VD2): VertexRDD[VD2]
// Show only vertices unique to this set based on their VertexId's
def minus(other: RDD[(VertexId, VD)])
// Remove vertices from this set that appear in the other set
def diff(other: VertexRDD[VD]): VertexRDD[VD]
// Join operators that take advantage of the internal indexing to accelerate joins (substantially)
def leftJoin[VD2, VD3](other: RDD[(VertexId, VD2)])(f: (VertexId, VD, Option[VD2]) => VD3): VertexRDD[VD3]
def innerJoin[U, VD2](other: RDD[(VertexId, U)])(f: (VertexId, VD, U) => VD2): VertexRDD[VD2]
// Use the index on this RDD to accelerate a `reduceByKey` operation on the input RDD.
def aggregateUsingIndex[VD2](other: RDD[(VertexId, VD2)], reduceFunc: (VD2, VD2) => VD2): VertexRDD[VD2]
}
예를 들어 filter 연산자가 VertexRDDVertexRDD를 반환하는 방식을 알아보세요. filter는 실제로 BitSet을 사용해 구현되므로 인덱스를 재사용하고 다른 VertexRDD와의 빠른 조인 능력을 보존해요. 마찬가지로 mapValues 연산자는 map 함수가 VertexId를 변경하지 못하게 해 동일한 HashMap 데이터 구조가 재사용되게 해요. leftJoin과 innerJoin 둘 다 같은 HashMap에서 파생된 두 VertexRDD를 조인할 때 이를 식별하고, 비싼 점 조회 대신 선형 스캔으로 조인을 구현할 수 있어요.
aggregateUsingIndex 연산자는 RDD[(VertexId, A)]에서 새 VertexRDDVertexRDD를 효율적으로 구성하는 데 유용해요. 개념적으로 정점 집합 위에 VertexRDD[B]를 구성했다면, 그 집합이 어떤 RDD[(VertexId, A)]의 정점의 초집합(super-set)이라면 인덱스를 재사용해 RDD[(VertexId, A)]를 집계하고 그 후 인덱싱할 수 있어요. 예를 들어:
val setA: VertexRDD[Int] = VertexRDD(sc.parallelize(0L until 100L).map(id => (id, 1)))
val rddB: RDD[(VertexId, Double)] = sc.parallelize(0L until 100L).flatMap(id => List((id, 1.0), (id, 2.0)))
// There should be 200 entries in rddB
rddB.count
val setB: VertexRDD[Double] = setA.aggregateUsingIndex(rddB, _ + _)
// There should be 100 entries in setB
setB.count
// Joining A and B should now be fast!
val setC: VertexRDD[Double] = setA.innerJoin(setB)((id, a, b) => a + b)
EdgeRDD
RDD[Edge[ED]]를 확장하는 EdgeRDD[ED]는 PartitionStrategy에 정의된 다양한 파티셔닝 전략 중 하나로 분할된 블록으로 간선을 구성해요. 각 파티션 내에서 간선 속성과 인접 구조는 분리되어 저장돼, 속성 값을 변경할 때 최대 재사용을 가능하게 해요.
EdgeRDDEdgeRDD가 노출하는 추가 함수 세 개는 다음과 같아요.
// Transform the edge attributes while preserving the structure
def mapValues[ED2](f: Edge[ED] => ED2): EdgeRDD[ED2]
// Reverse the edges reusing both attributes and structure
def reverse: EdgeRDD[ED]
// Join two `EdgeRDD`s partitioned using the same partitioning strategy.
def innerJoin[ED2, ED3](other: EdgeRDD[ED2])(f: (VertexId, VertexId, ED, ED2) => ED3): EdgeRDD[ED3]
대부분의 애플리케이션에서 EdgeRDDEdgeRDD에 대한 연산은 그래프 연산자를 통해 수행되거나 기본 RDD 클래스에 정의된 연산에 의존한다는 것을 발견했어요.
최적화된 표현(Optimized Representation)
GraphX가 분산 그래프를 표현하는 데 사용하는 최적화의 상세 설명은 이 가이드의 범위를 벗어나지만, 몇 가지 고수준 이해는 확장 가능한 알고리즘 설계와 API의 최적 사용에 도움이 될 수 있어요. GraphX는 분산 그래프 파티셔닝에 vertex-cut 접근법을 채택해요.
간선을 따라 그래프를 분할하는 대신, GraphX는 정점을 따라 그래프를 분할해 통신과 저장 오버헤드를 모두 줄일 수 있어요. 논리적으로 이는 간선을 머신에 할당하고 정점이 여러 머신에 걸쳐 있게 하는 것에 해당해요. 간선을 할당하는 정확한 방법은 PartitionStrategy에 의존하며, 다양한 휴리스틱에는 여러 트레이드오프가 있어요. 사용자는 Graph.partitionBy 연산자로 그래프를 다시 파티셔닝해 다른 전략을 선택할 수 있어요. 기본 파티셔닝 전략은 그래프 구성 시 제공된 간선의 초기 파티셔닝을 사용하는 거예요. 하지만 사용자는 쉽게 2D 파티셔닝이나 GraphX에 포함된 다른 휴리스틱으로 전환할 수 있어요.
간선이 파티셔닝되면 효율적인 그래프 병렬 계산의 핵심 과제는 정점 속성을 간선과 효율적으로 조인하는 것이에요. 실제 그래프는 보통 정점보다 간선이 많으므로, 정점 속성을 간선쪽으로 옮겨요. 모든 파티션이 모든 정점에 인접한 간선을 포함하지는 않으므로 내부적으로 라우팅 테이블을 유지하는데, 이 테이블은 triplets와 aggregateMessages 같은 연산에 필요한 조인을 구현할 때 정점을 어디로 브로드캐스트할지 식별해요.
그래프 알고리즘(Graph Algorithms)
GraphX는 분석 작업을 단순화하는 그래프 알고리즘 집합을 포함해요. 알고리즘은 org.apache.spark.graphx.lib 패키지에 있고, GraphOps를 통해 Graph의 메서드로 직접 접근할 수 있어요. 이 섹션은 알고리즘과 사용 방법을 설명해요.
PageRank
PageRank는 u에서 v로의 간선이 v의 중요성을 u가 보증하는 것이라고 가정하고, 그래프에서 각 정점의 중요성을 측정해요. 예를 들어 Twitter 사용자가 다른 많은 사람에게 팔로우된다면 그 사용자는 높게 랭크돼요.
GraphX는 PageRank 객체의 메서드로 PageRank의 정적·동적 구현을 제공해요. 정적 PageRank는 고정된 반복 횟수 동안 실행되고, 동적 PageRank는 랭크가 수렴(즉, 지정된 허용 오차보다 더 이상 변하지 않을 때)할 때까지 실행돼요. GraphOps는 이 알고리즘을 Graph의 메서드로 직접 호출할 수 있게 해요.
GraphX는 PageRank를 실행할 수 있는 예제 소셜 네트워크 데이터셋도 포함해요. 사용자 집합은 data/graphx/users.txt에, 사용자 간 관계 집합은 data/graphx/followers.txt에 주어져요. 각 사용자의 PageRank를 다음과 같이 계산해요.
import org.apache.spark.graphx.GraphLoader
// Load the edges as a graph
val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt")
// Run PageRank
val ranks = graph.pageRank(0.0001).vertices
// Join the ranks with the usernames
val users = sc.textFile("data/graphx/users.txt").map { line =>
val fields = line.split(",")
(fields(0).toLong, fields(1))
}
val ranksByUsername = users.join(ranks).map {
case (id, (username, rank)) => (username, rank)
}
// Print the result
println(ranksByUsername.collect().mkString("\n"))
전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/graphx/PageRankExample.scala"에서 찾을 수 있어요.
Connected Components
connected components 알고리즘은 그래프의 각 연결 요소를 가장 작은 번호의 정점 ID로 레이블링해요. 예를 들어 소셜 네트워크에서 connected components는 클러스터를 근사할 수 있어요. GraphX는 ConnectedComponents 객체에 알고리즘 구현을 포함하고, PageRank 섹션의 예제 소셜 네트워크 데이터셋의 connected components를 다음과 같이 계산해요.
import org.apache.spark.graphx.GraphLoader
// Load the graph as in the PageRank example
val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt")
// Find the connected components
val cc = graph.connectedComponents().vertices
// Join the connected components with the usernames
val users = sc.textFile("data/graphx/users.txt").map { line =>
val fields = line.split(",")
(fields(0).toLong, fields(1))
}
val ccByUsername = users.join(cc).map {
case (id, (username, cc)) => (username, cc)
}
// Print the result
println(ccByUsername.collect().mkString("\n"))
전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/graphx/ConnectedComponentsExample.scala"에서 찾을 수 있어요.
Triangle Counting
정점이 서로 간에 간선이 있는 두 인접 정점을 가지면, 그 정점은 삼각형의 일부예요. GraphX는 TriangleCount 객체에 각 정점을 통과하는 삼각형 수를 결정하는 triangle counting 알고리즘을 구현하며, 클러스터링의 척도를 제공해요. PageRank 섹션의 소셜 네트워크 데이터셋의 triangle count를 계산해요. TriangleCount는 간선이 정규 방향(srcId < dstId)이고 그래프가 Graph.partitionBy로 파티셔닝돼야 함을 요구한다는 점에 유의하세요.
import org.apache.spark.graphx.{GraphLoader, PartitionStrategy}
// Load the edges in canonical order and partition the graph for triangle count
val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt", true)
.partitionBy(PartitionStrategy.RandomVertexCut)
// Find the triangle count for each vertex
val triCounts = graph.triangleCount().vertices
// Join the triangle counts with the usernames
val users = sc.textFile("data/graphx/users.txt").map { line =>
val fields = line.split(",")
(fields(0).toLong, fields(1))
}
val triCountByUsername = users.join(triCounts).map { case (id, (username, tc)) =>
(username, tc)
}
// Print the result
println(triCountByUsername.collect().mkString("\n"))
전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/graphx/TriangleCountingExample.scala"에서 찾을 수 있어요.
예제(Examples)
텍스트 파일에서 그래프를 만들고, 그래프를 중요한 관계와 사용자로 제한하고, 부분 그래프에서 page-rank를 실행한 뒤, 마지막으로 상위 사용자와 연관된 속성을 반환하고 싶다고 가정해봐요. 이 모든 것을 GraphX로 단 몇 줄에 할 수 있어요.
import org.apache.spark.graphx.GraphLoader
// Load my user data and parse into tuples of user id and attribute list
val users = (sc.textFile("data/graphx/users.txt")
.map(line => line.split(",")).map( parts => (parts.head.toLong, parts.tail) ))
// Parse the edge data which is already in userId -> userId format
val followerGraph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt")
// Attach the user attributes
val graph = followerGraph.outerJoinVertices(users) {
case (uid, deg, Some(attrList)) => attrList
// Some users may not have attributes so we set them as empty
case (uid, deg, None) => Array.empty[String]
}
// Restrict the graph to users with usernames and names
val subgraph = graph.subgraph(vpred = (vid, attr) => attr.size == 2)
// Compute the PageRank
val pagerankGraph = subgraph.pageRank(0.001)
// Get the attributes of the top pagerank users
val userInfoWithPageRank = subgraph.outerJoinVertices(pagerankGraph.vertices) {
case (uid, attrList, Some(pr)) => (pr, attrList.toList)
case (uid, attrList, None) => (0.0, attrList.toList)
}
println(userInfoWithPageRank.vertices.top(5)(Ordering.by(_._2._1)).mkString("\n"))
전체 예제 코드는 Spark 저장소의 "examples/src/main/scala/org/apache/spark/examples/graphx/ComprehensiveExample.scala"에서 찾을 수 있어요.
더 알아보기 (Learn more)
- Spark RDD 프로그래밍 가이드 — RDD 개념과 영속화.
- GraphX API 문서 — 모든 연산자의 공식 목록.