Spark

Spark Packages 생태계

Spark의 강점 중 하나는 커뮤니티가 만들어낸 패키지·도구 생태계다. 이 책이 쓰인 시점 기준 spark-packages.org에 등록된 패키지만 300개가 넘고, 계속 늘어나고 있다. 누구나 이 저장소에 패키지를 배포할 수 있고, 그 밖에도 등록되지 않은 패키지가 다수 존재한다. 일부 패키지는 성숙해지고 널리 쓰이면서 오픈소스 Spark 코어 자체에 병합되기도 한다.

이 패키지들을 하나로 묶는 공통점은, 엔지니어가 특정 애플리케이션에 최적화된 형태로 Spark를 확장할 수 있게 해준다는 것이다. GraphFrames가 대표적인 예로, 저수준 API로는 지원하지 않는 방식으로 Spark의 Structured API 위에서 훨씬 사용하기 쉽고 언어에 구애받지 않는 그래프 분석을 제공한다. 이 밖에도 머신러닝·딥러닝을 Spark 위에 얹어 확장하는 패키지들이 많다. 헬스케어·유전체 분야도 빅데이터 응용 기회가 특히 크게 늘어난 영역인데, ADAM Project는 Spark의 Catalyst 엔진에 특화된 내부 최적화를 활용해 유전체 처리용 API와 CLI를 제공하고, Hail은 VCF 등의 시퀀싱/마이크로어레이 데이터를 다루는 오픈소스 유전체 분석 프레임워크로, 노트북 수준의 기가바이트급 데이터부터 클러스터의 테라바이트급 데이터까지 정교한 도메인 특화 변환을 지원한다.

GraphFrames로 그래프 만들기

그래프 이론의 핵심은 서로 다른 노드(정점, vertex) 사이의 관계(엣지, edge)를 정의하는 것이다. 노드는 단위(unit)이고 엣지는 그 단위들 사이에 정의된 관계다. 이런 object-relationship-object 구조로 문제를 표현하는 것이 그래프 분석의 일반적인 접근이며, GraphFrames는 이를 매우 쉽게 시작할 수 있게 해준다.

GraphFrames는 외부 패키지이므로, 커맨드라인에서 실행할 때는 spark-shell 실행 시 --packages 옵션으로 Maven 좌표를 지정해 의존성을 불러와야 한다(예: graphframes:graphframes:0.5.0-spark2.1-s_2.11).

예제는 자전거 공유 데이터(정류소 정보, 이용 기록)를 사용한다. 그래프를 만들려면 정점(vertices) 테이블과 엣지(edges) 테이블을 각각 구성해야 하는데, GraphFrames가 요구하는 명명 규칙에 맞춰 컬럼 이름을 바꿔야 한다 — 정점 테이블은 식별자 컬럼을 name에서 id로, 엣지 테이블은 출발/도착 컬럼을 각각 src, dst로 바꾼다. 이렇게 준비한 두 DataFrame으로 GraphFrame(vertices, edges)을 생성하면 그래프가 만들어지고, 이후 여러 알고리즘이 반복적으로 순회할 것이므로 두 DataFrame을 cache()해 두는 것이 좋다.

val stationVertices = bikeStations
  .withColumnRenamed("name", "id")
  .distinct()
val tripEdges = bikeTrips
  .withColumnRenamed("Start Station", "src")
  .withColumnRenamed("End Station", "dst")

val stationGraph = GraphFrame(stationVertices, tripEdges)
tripEdges.cache()
stationVertices.cache()

PageRank와 station-to-station 집계

GraphFrames는 Spark의 원조 그래프 분석 패키지인 GraphX 위에 구축되어 있어서, PageRank 같은 검증된 내장 알고리즘을 바로 활용할 수 있다. PageRank는 Larry Page가 만들고 구글 검색 엔진으로 대중화된 알고리즘으로, 한 페이지(노드)가 받는 링크(엣지)의 수와 질을 근거로 그 노드의 중요도를 추정한다. 기본 가정은 중요한 노드일수록 다른 (역시 중요한) 노드로부터 더 많은 링크를 받는다는 것이다. 이 개념은 웹페이지뿐 아니라 자전거 정류소처럼 그래프 형태를 띠는 어떤 구조에도 적용할 수 있다.

자전거 공유 그래프에 resetProbability(0.15), maxIter(10) 설정으로 PageRank를 실행하고 pagerank 값 기준 내림차순으로 상위 5개 정점을 뽑아보면, Caltrain(기차역) 정류소들이 상위권에 오른다. 기차역은 자연스러운 환승 거점이라, 다른 교통수단과 연결되는 지점에서 자전거 공유 서비스 이용이 몰리는 현상을 그래프 구조만으로 포착해낸 것이다.

반면 '출발지-도착지 쌍 중 가장 흔한 이동 경로는 무엇인가'라는 질문은 그래프 알고리즘 없이도 풀 수 있다. 동일한 출발역-도착역 조합으로 발생한 트립들을 세어 합치기만 하면 되기 때문이다. edges.groupBy("src", "dst").count().orderBy(desc("count")).limit(10) 같은 단순 집계 쿼리로 상위 10개 경로를 구할 수 있다.

stationGraph
  .edges
  .groupBy("src", "dst")
  .count()
  .orderBy(desc("count"))
  .limit(10)
  .show()