Spark

CSV 읽기와 스키마 추론

이 절에서는 미국 교통통계국이 제공하는 항공편 데이터를 사용한다. 이 CSV 파일들은 반정형(semi-structured) 데이터 형식으로, 파일 안의 한 행이 앞으로 만들 DataFrame의 한 행에 대응한다. 예를 들어 2015-summary.csv 파일은 DEST_COUNTRY_NAME, ORIGIN_COUNTRY_NAME, count 세 컬럼을 갖는다.

Spark는 다양한 데이터 소스를 읽고 쓸 수 있으며, 이를 위해 SparkSession에 연결된 DataFrameReader를 사용한다. CSV는 완전한 정형 데이터가 아니기 때문에, 아래 예제에서는 Spark가 데이터를 일부 읽어 컬럼 타입을 추측하는 스키마 추론(schema inference) 옵션을 사용하고, 첫 번째 행이 헤더임을 지정하는 옵션도 함께 준다. (스키마를 엄격하게 직접 지정하는 방법도 있다.)

val flightData2015 = spark
  .read
  .option("inferSchema", "true")
  .option("header", "true")
  .csv("/mnt/defg/flight-data/csv/2015-summary.csv")

flightData2015 = spark\
  .read\
  .option("inferSchema", "true")\
  .option("header", "true")\
  .csv("/mnt/defg/flight-data/csv/2015-summary.csv")

이렇게 만들어진 DataFrame은 컬럼 집합은 갖고 있지만 로우 수는 '미확정' 상태다. 왜냐하면 데이터를 읽는 것 자체가 트랜스포메이션이고, 트랜스포메이션은 지연 연산이기 때문이다. Spark는 이 시점에 컬럼 타입을 추측하기 위해 데이터를 살짝 들여다봤을 뿐이다. take(3) 같은 액션을 호출해야 실제로 몇 개의 로우를 눈으로 확인할 수 있다.

explain 계획으로 보는 sort의 wide dependency

이제 count 컬럼 기준으로 데이터를 정렬하는 트랜스포메이션을 추가해보자. 주의할 점은 sort가 원본 DataFrame을 수정하는 것이 아니라, 이전 DataFrame을 변환한 새로운 DataFrame을 반환한다는 것이다. sort를 호출하는 시점에는 데이터에 아무 일도 일어나지 않는다 — 트랜스포메이션 하나가 추가됐을 뿐이다. 하지만 explain()을 호출하면 Spark가 클러스터 전역에서 이를 어떻게 실행할지에 대한 계획을 쌓아가고 있음을 확인할 수 있다.

flightData2015.sort("count").explain()

explain 계획은 위에서 아래로 읽으며, 맨 위가 최종 결과이고 맨 아래가 데이터의 원천(source)이다. sort 예제의 계획을 보면 맨 앞쪽 키워드로 sort, exchange, FileScan이 등장한다. 이는 데이터 정렬이 실제로는 wide dependency이기 때문이다 — 정렬을 하려면 로우들끼리 서로 비교해야 하므로 셔플이 필요하다.

이제 액션을 호출해 이 계획을 실제로 실행시킬 수 있다.

flightData2015.sort("count").take(2)

지금까지 쌓아 올린 트랜스포메이션들의 논리적 계획은 DataFrame의 계보(lineage)를 정의한다. 덕분에 Spark는 어느 시점에서든 동일한 입력 데이터에 대해 이전에 수행했던 모든 연산을 다시 적용함으로써 임의의 파티션을 재계산하는 방법을 알고 있다. 이는 Spark 프로그래밍 모델의 핵심인 함수형 프로그래밍에 닿아 있다 — 트랜스포메이션이 동일하게 유지되는 한 동일한 입력은 항상 동일한 출력을 낳는다는 원리다.

DataFrame과 SQL은 동일한 밑단 계획으로 컴파일된다

Spark에서 DataFrame과 SQL은 사실상 동일한 것이다. 비즈니스 로직을 둘 중 어느 언어로 표현하든 Spark는 그 로직을 (explain 계획에서 보게 되는) 동일한 밑단 실행 계획으로 컴파일한 뒤 실행한다. Spark SQL을 쓰면 임의의 DataFrame을 테이블 혹은 뷰(임시 테이블)로 등록해 순수 SQL로 조회할 수 있다. SQL 쿼리를 작성하든 DataFrame 코드를 작성하든 성능 차이는 전혀 없다 — 둘 다 결국 같은 밑단 계획으로 '컴파일'된다.

flightData2015.createOrReplaceTempView("flight_data_2015")

이제 spark.sql 함수로 SQL 쿼리를 실행할 수 있는데, 이 함수는 편리하게도 새로운 DataFrame을 반환한다. SQL 쿼리 결과가 다시 DataFrame이 된다는 점이 다소 순환적으로 보일 수 있지만, 실제로는 매우 강력하다 — 사용자는 그때그때 가장 편한 방식으로 트랜스포메이션을 표현하면서도 효율성을 전혀 희생하지 않는다.

val sqlWay = spark.sql("""
SELECT DEST_COUNTRY_NAME, count(1)
FROM flight_data_2015
GROUP BY DEST_COUNTRY_NAME
""")

val dataFrameWay = flightData2015
  .groupBy('DEST_COUNTRY_NAME)
  .count()

sqlWay.explain
dataFrameWay.explain

두 explain 계획을 비교해보면 완전히 동일한 밑단 계획으로 컴파일된 것을 확인할 수 있다. max 함수를 이용해 특정 목적지의 최대 항공편 수를 구하는 예제도 SQL과 DataFrame 두 방식으로 표현 가능하다.

spark.sql("SELECT max(count) from flight_data_2015").take(1)

import org.apache.spark.sql.functions.max
flightData2015.select(max("count")).take(1)

max는 관련 컬럼의 값을 훑으며 이전에 본 값보다 큰지 확인하는 연산으로, 결국 하나의 행으로 필터링하는 셈이므로 이 역시 트랜스포메이션이다.

다단계 쿼리를 DAG로 분해하기 (top-5 목적지 국가)

이번에는 목적지 국가 상위 5개를 구하는 좀 더 복잡한 다단계 쿼리를 SQL과 DataFrame 양쪽으로 살펴본다.

SELECT DEST_COUNTRY_NAME, sum(count) as destination_total
FROM flight_data_2015
GROUP BY DEST_COUNTRY_NAME
ORDER BY sum(count) DESC
LIMIT 5

import org.apache.spark.sql.functions.desc
flightData2015
  .groupBy("DEST_COUNTRY_NAME")
  .sum("count")
  .withColumnRenamed("sum(count)", "destination_total")
  .sort(desc("destination_total"))
  .limit(5)
  .collect()

이 쿼리는 원본 데이터까지 거슬러 올라가는 7단계로 이루어져 있고, 각 단계가 새로운 불변 DataFrame을 만들어내는 방향성 비순환 그래프(DAG)를 형성한다. 실제 explain 계획은 물리 실행 최적화 때문에 아래 단계 나열과 정확히 일치하지는 않지만, 개념을 이해하는 출발점으로는 충분하다.

1. **read** — DataFrame은 이전에 정의해 두었지만, 이 DataFrame(또는 이로부터 파생된 DataFrame)에 액션이 호출되기 전까지 Spark는 실제로 데이터를 읽지 않는다. 2. **groupBy** — groupBy를 호출하면 RelationalGroupedDataset이라는 결과를 얻는다. 이는 그루핑 정보는 지정됐지만 추가로 집계 함수를 지정해야 조회 가능한 DataFrame을 가리키는 이름일 뿐이다. 이 상태에서 바로 액션을 실행하면 동작하지 않는다 — 아직 연산(관계대수적 정보 전달 외에)은 전혀 수행되지 않았다. 3. **sum (집계)** — sum 집계 메서드는 컬럼 표현식 또는 컬럼명을 입력으로 받는다. sum 호출 결과는 새로운 스키마를 가진 새 DataFrame이며, 각 컬럼의 타입은 알고 있지만 역시 아직 어떤 계산도 수행되지 않았다 — 지금까지 제공한 타입 정보를 추적하고 있을 뿐이다. 4. **withColumnRenamed (컬럼 이름 변경)** — 원래 컬럼명과 새 컬럼명 두 인자를 받는 단순 이름 변경이며, 이 역시 계산을 수행하지 않는 또 하나의 트랜스포메이션일 뿐이다. 5. **sort (정렬)** — DataFrame의 맨 위에서 값을 가져갔을 때 destination_total 컬럼의 가장 큰 값들이 오도록 정렬한다. desc 함수를 임포트해야 하는데, desc는 문자열이 아니라 Column을 반환한다. 대부분의 DataFrame 메서드는 문자열(컬럼명)이나 Column 타입·표현식을 모두 받아들이며, Column과 표현식은 사실상 같은 것이다. 6. **limit** — 값 5개만 원한다고 지정하는 단계다. 값(value) 기준이 아니라 위치(position) 기준으로 걸러낸다는 점을 빼면 필터와 비슷하며, 여전히 지연(lazy) 연산이다. 결국 특정 크기의 DataFrame을 지정하는 것과 마찬가지다. 7. **collect (액션)** — 비로소 위에서 쌓아 올린 DataFrame의 결과를 실행 언어의 리스트나 배열로 실제로 모으는 과정이 시작된다.

이 쿼리에 대해 explain()을 호출하면 다음과 유사한 물리 실행 계획을 볼 수 있다.

== Physical Plan ==
TakeOrderedAndProject(limit=5, orderBy=[destination_total#16194L DESC], output=[DEST_COUNTRY_...])
+- *HashAggregate(keys=[DEST_COUNTRY_NAME#7323], functions=[sum(count#7325L)])
   +- Exchange hashpartitioning(DEST_COUNTRY_NAME#7323, 5)
      +- *HashAggregate(keys=[DEST_COUNTRY_NAME#7323], functions=[partial_sum(count#7325L)])
         +- InMemoryTableScan [DEST_COUNTRY_NAME#7323, count#7325L]
            +- InMemoryRelation [...]
               +- *Scan csv [DEST_COUNTRY_NAME#7578, ORIGIN_COUNTRY_NAME#7579, count#7580L]

이 물리 계획이 앞서 정리한 '개념상 계획'과 한 줄씩 정확히 일치하지는 않지만 구성 요소는 모두 들어있다 — limit 문과 (첫 줄의) orderBy를 확인할 수 있고, partial_sum 호출을 통해 집계가 두 단계로 이뤄짐도 알 수 있다. 숫자 합산은 교환법칙이 성립하는 연산이므로, Spark는 먼저 파티션 단위로 부분 합(partial_sum)을 계산한 뒤(이 과정에서 Exchange로 셔플이 일어난다) 최종 합(HashAggregate)을 구하는 두 단계 집계로 최적화한다. 물론 원본 데이터를 어떻게 읽어 들였는지(Scan csv)도 계획에 그대로 남아있다.

마지막으로, 결과를 항상 collect로 모아야 하는 것은 아니다. 예컨대 JDBC를 통해 데이터베이스에 쓰거나 새로운 파일로 쓰는 것처럼, Spark가 지원하는 다른 데이터 소스로 결과를 출력할 수도 있다.