정현석

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

4 분 소요

[Spark The Definitive Guide - BIG DATA PROCESSING MADE SIMPLE] 책을 중심으로 스파크를 개인 공부를 위해 요약 및 정리해보았습니다.1 다소 복잡한 설치 과정은 도커에 미리 이미지를 업로드해 놓았습니다. 즉, 도커 이미지를 pull하면 바로 스파크를 사용할 수 있습니다.

예제에 대한 실행 언어는 스칼라(scala)로 했습니다. 앞으로 스파크와 관련된 내용은 딥러닝 부분을 제외하고 스칼라로 진행될 예정입니다.

기본 실행 방법

  1. 예제에 사용 될 데이터들은 도커 이미지 생성 후 spark-3.3.0 안의 하위 폴더 data를 생성 후, 이 폴더에 추가합니다.
    • 1.1 데이터와 도커 설치 및 활용하는 방법은 위에 링크를 남겼습니다.
  2. 프로그램 시작은 cd spark-3.3.0 후, ./bin/spark-shell 명령어를 실행하시면 됩니다.

[Spark] 집계 연산, 함수, SQL 명령어 정리글에 이어서 진행됩니다.

이 글은 스파크에서 그룹화와 윈도우 함수, 그룹화 셋을 다루는 방법을 정리한다. groupBy와 agg로 그룹 단위 집계를 만들고, 윈도우 명세(partitionBy·orderBy·rowsBetween)로 로우 간 계산을 수행한 뒤, 롤업·큐브·grouping_id·피벗으로 여러 차원의 요약 결과를 만드는 과정을 예제와 함께 살펴본다.

초기 설정

val df = spark.read.format("csv"
    ).option("header", "true"
    ).option("inferSchema", "true"
    ).load("./data/retail-data/all/*.csv"
    ).coalesce(5)

df.cache()
df.createOrReplaceTempView("dfTable")

3. 그룹화

  • 데이터 그룹 기반의 집계는 단일 컬럼의 데이터를 그룹화하고 해당 그룹의 다른 여러 컬럼을 사용해서 계산하기 위해 카테고리형 데이터를 사용한다.
  • 데이터 그룹 기반의 집계를 설명하는 데 가장 좋은 방법은 그룹화를 해보는 것이다.
  • 그룹화 작업은 하나 이상의 컬럼을 그룹화하고 집계 연산을 수행하는 두 단계로 이뤄진다.
    • 첫 번째 단계에서는 RelationalGroupedDataset이 반환되고, 두 번째 단계에서는 DataFrame이 반환된다.
df.groupBy("InvoiceNo", "CustomerId").count().show(5)

  • 실행결과

    Screenshot from 2022-09-06 10-27-12

3.1 표현식을 이용한 그룹화

  • agg 메서드는 여러 집계 처리를 한 번에 지정할 수 있으며, 집계에 표현식을 사용할 수 있다.
  • 또한, 트랜스포메이션이 완료된 컬럼에 alias 메서드를 사용할 수 있다.
import org.apache.spark.sql.functions.{count, expr}

df.groupBy("InvoiceNo").agg(
  count("Quantity").alias("quan"),
  expr("count(Quantity)")
).show(5)

  • 실행결과

    Screenshot from 2022-09-06 10-51-43

3.2 맵을 이용한 그룹화

  • 컬럼을 키로, 수행할 집계 함수의 문자열을 값으로 하는 맵(map) 타입을 사용해 트랜스포메이션을 정의할 수 있다.
df.groupBy("InvoiceNo").agg(
  "Quantity"->"avg",
  "Quantity"->"stddev_pop"
).show(5)

  • 실행결과

    Screenshot from 2022-09-06 10-56-03

4. 윈도우 함수

  • 윈도우(window) 함수는 데이터의 특정 윈도우를 대상으로 고유의 집계 연산을 수행한다.
    • 데이터의 윈도우는 현재 데이터에 대한 참조(reference)를 사용해 정의한다.
    • 윈도우 명세(window specification)는 함수에 전달된 로우를 결정한다.
    • 윈도우(window)는 로우 간 연산을 처리하기 위한 도구이다. 2
  • 윈도우 함수는 프레임(frame)에 입력되는 모든 로우에 대해 결괏값을 계산한다.
    • 프레임 : 로우 그룹 기반의 테이블을 의미한다.
  • 스파크는 다음 3가지 종류의 윈도우 함수를 지원한다.
    1. 랭크 함수(ranking function)
    2. 분석 함수(analytic function)
    3. 집계 함수(aggregate function)
  • 다음 그림은 로우가 어떻게 여러 프레임에 할당될 수 있는지 나타낸다.

Screenshot from 2022-09-06 11-01-58

  • 예제를 위해 주문 일자(Invoice Date) 컬럼을 변환해 date 컬럼을 만든다.
    • spark.sql("set spark.sql.legacy.timeParserPolicy=LEGACY") : spark 2.4 이하에서는 java.text.SimpleDateFormat으로 날짜·시각 문자열을 해석했지만, spark 3.0부터 파싱 정책이 바뀌어 다음 예제와 같은 포맷은 그대로 지원되지 않아 에러가 발생한다. 따라서 이 코드를 삽입해줘야 한다. 34
import org.apache.spark.sql.functions.{col, to_date}

spark.sql("set spark.sql.legacy.timeParserPolicy=LEGACY")
val dfWithDate = df.withColumn("date", to_date(col("InvoiceDate"), "MM/d/yyyy H:mm"))
dfWithDate.createOrReplaceTempView("dfWithDate")
  • 윈도우 함수를 정의하기 위해 첫 번째 단계로 윈도우 명세를 만든다.
    • partitionBy 메서드는 그룹을 어떻게 나눌지 결정하는 것과 유사한 개념이다.
    • orderBy 메서드는 파티션의 정렬 방식을 정의한다.
    • 프레임 명세(rowBetween 구문)는 입력된 로우의 참조를 기반으로 프레임에 로우가 포함될 수 있는지 결정한다. 5 
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.col

val windowSpec = Window.partitionBy("CustomerId", "date"
  ).orderBy(col("Quantity").desc
  ).rowsBetween(Window.unboundedPreceding, Window.currentRow)
  • 이번에는 시간대별 최대 구매 개수를 구하는 예이다.
    • 위 예제에서 사용한 집계 함수에 컬럼명이나 표현식을 전달한다.
    • 이 함수를 적용할 데이터 프레임이 정의된 윈도우 명세도 함께 사용한다.
import org.apache.spark.sql.functions.max

val maxPurchaseQuantity = max(col("Quantity")
  ).over(windowSpec)

  • 실행결과

    Screenshot from 2022-09-06 11-44-29

  • dense_rank 함수를 사용해 모든 고객에 대해 최대 구매 수량을 가진 날짜가 언제인지 알아본다.

    • 동일한 값이 나오거나 중복 로우가 발생해 순위가 비어 있을 수 있으므로 rank 함수 대신 dense_rank 함수를 사용한다.
import org.apache.spark.sql.functions.{dense_rank, rank}

val purchaseDenseRank = dense_rank().over(windowSpec)
val purchaseRank = rank().over(windowSpec)

  • 실행결과

    Screenshot from 2022-09-06 11-52-43

  • 위 예제를 통해 select 구문에서 사용할 수 있는 컬럼을 반환했다.
  • 이제 select 메서드를 사용해 계산된 윈도우 값을 확인해본다.
import org.apache.spark.sql.functions.col

dfWithDate.where("CustomerId IS NOT NULL"
  ).orderBy("CustomerId"
  ).select(
    col("CustomerId"),
    col("date"),
    col("Quantity"),
    purchaseRank.alias("quantityRank"),
    purchaseDenseRank.alias("quantityDenseRank"),
    maxPurchaseQuantity.alias("maxPurchaseQuantity")
  ).show()

  • 실행결과

    Screenshot from 2022-09-06 12-09-13

5. 그룹화 셋

  • 그룹화 셋(grouping set)은 여러 그룹에 걸쳐 여러 집계를 결합하는 저수준 기능이다.
  • 그룹화 셋을 이용하면 group-by 구문에서 원하는 형태로 집계를 생성할 수 있다.
  • 그룹화 셋은 SQL에서만 사용할 수 있다. 7
    • 롤업과 큐브는 DataFrame API의 rollup, cube 메서드로도 사용할 수 있지만, GROUPING SETS로 임의 조합을 지정하는 것은 SQL에서만 가능하다.
val dfNoNull = dfWithDate.na.drop()

dfNoNull.createOrReplaceTempView("dfNoNull")

spark.sql("""
  SELECT CustomerID, stockCode, sum(Quantity) FROM dfNoNull
  GROUP BY customerID, stockCode
  ORDER BY CustomerId DESC, stockCode DESC
""").show(5)

  • 실행결과

    Screenshot from 2022-09-06 12-27-06

5.1 롤업

  • 다양한 컬럼을 그룹화 키로 설정하면 그룹화 키로 설정된 조합뿐만 아니라 데이터셋에서 볼 수 있는 실제 조합을 모두 살펴볼 수 있다.
  • 롤업(rollup)은 group-by 스타일의 다양한 연산을 수행할 수 있는 다차원 집계 기능이다.
    • 그룹별 결과 뿐만 아니라 총 집계도 해준다.
  • 다음 예제에서는 시간과 공간을 축으로 하는 롤업을 생성한다.
val rolledUpDF = dfNoNull.rollup("Date", "Country"
  ).agg(sum(col("Quantity"))
  ).selectExpr("date", "Country", "'sum(Quantity)' as total_quantity"
  ).orderBy("Date")
rolledUpDF.show(5)

5.2 큐브

  • 큐브(cube)는 롤업을 고차원적으로 사용할 수 있게 해준다.
  • 큐브는 요소들을 계층적으로 다루는 대신 모든 차원에 대해 동일한 작업을 수행한다.
  • 큐브를 사용하면 테이블의 모든 차원 조합을 한 번에 조회할 수 있는 요약 정보 테이블을 만들 수 있다.
dfNoNull.cube("Date", "Country"
  ).agg(sum(col("Quantity"))
  ).selectExpr("date", "Country", "'sum(Quantity)' as total_quantity"
  ).orderBy("Date"
  ).show()

5.3 그룹화 메타데이터

  • 큐브와 롤업을 사용하다 보면 집계 수준에 따라 쉽게 필터링하기 위해 집계 수준을 조회하는 경우가 발생하는 데 이때 grouping_id를 사용한다.
    • grouping_id는 결과 데이터셋의 집계 수준을 명시하는 컬럼을 제공한다.
import org.apache.spark.sql.functions.{grouping_id, sum, expr}

dfNoNull.cube("customerId", "stockCode"
    ).agg(grouping_id(), sum("Quantity")
    ).orderBy(expr("grouping_id()").desc
    ).show()

  • 실행결과

    Screenshot from 2022-09-06 12-42-55

5.4 피벗

  • 피벗(pivot)을 사용해 로우를 컬럼으로 변환할 수 있다. 8
    • USA와 관련된 컬럼을 살펴보면 USA_sum(Quantity), USA_sum(UnitPrice), USA_sum(CustomerId)가 있다.
      • pivoted.printSchema() 코드를 실행해보면 된다.
val pivoted = dfWithDate.groupBy("date").pivot("Country").sum()

pivoted.where("date > '2011-12-05'"
    ).select("date", "USA_sum(Quantity)"
    ).show()

  • 실행결과

    Screenshot from 2022-09-06 12-48-58

6. 한계와 트레이드오프

  • 그룹화·롤업·큐브는 그룹 키 기준 셔플을 동반한다 : 그룹 키로 데이터를 재분배한 뒤 집계하므로 키의 고유값이 많을수록 셔플로 주고받는 데이터가 늘어난다. 스파크는 조인·집계를 위한 셔플 파티션 수를 spark.sql.shuffle.partitions(기본값 200)로 정하므로, 데이터 크기에 맞게 조정하지 않으면 파티션당 데이터가 지나치게 커지거나 반대로 작은 태스크가 많이 생긴다. 9
  • 롤업·큐브는 조합 수만큼 결과 로우를 늘린다 : 롤업은 그룹 키를 왼쪽부터 묶어 N+1개(예제의 Date, Country는 3개), 큐브는 모든 조합 2^N개의 집계 수준을 만든다. 집계 수준이 다른 로우가 한 결과에 섞여 나오므로 5.3절의 grouping_id로 걸러내지 않으면 결과가 예상보다 커진다. 7
  • 피벗은 고유값 수만큼 컬럼을 만든다 : 피벗 대상 컬럼(예제의 Country)의 값마다 집계 컬럼이 생기므로(예제의 USA_sum(Quantity)), 고유값이 많은 컬럼을 피벗하면 컬럼 수와 결과 크기가 함께 늘어난다. 컬럼명에 sum(Quantity)처럼 함수 호출 형태가 들어가므로 spark.sql()로 이 컬럼을 참조할 때는 백틱으로 감싸야 한다. 8
  • 윈도우 함수는 파티션 단위로 계산된다 : 4절의 partitionBy 키(예제의 CustomerId, date)를 기준으로 로우가 한 태스크에 모이므로, 파티션 키의 고유값이 적으면 태스크당 로우가 많아져 병렬성이 떨어지고 고유값이 많으면 태스크 수와 셔플 비용이 늘어난다. rowsBetween으로 프레임을 넓히면 로우마다 계산에 포함되는 로우가 늘어나므로 파티션 전체를 대상으로 하는 프레임이 가장 비싸다.

References

  1. Spark: The Definitive Guide - Bill Chambers, Matei Zaharia (O’Reilly Media, 2018), Chapter 7. Aggregations ↩

  2. Spark for Data Analyst - 4-4. window ↩

  3. String to Date migration from Spark 2.0 to 3.0 gives Fail to recognize ‘EEE MMM dd HH:mm:ss zzz yyyy’ pattern in the DateTimeFormatter ↩

  4. Apache Spark Documentation - Migration Guide: SQL, Datasets and DataFrame ↩

  5. Apache Spark Documentation - Window Functions ↩

  6. org.apache.spark.sql.expressions - Class Window ↩ ↩2

  7. Apache Spark Documentation - GROUP BY Clause (GROUPING SETS, ROLLUP, CUBE) ↩ ↩2

  8. Apache Spark Documentation - PIVOT Clause ↩ ↩2

  9. Apache Spark Documentation - Performance Tuning: SQL, Shuffle Partitions ↩

댓글남기기

참고

자가 개선 에이전트 루프 설계: 5단계 아키텍처와 리워드 해킹 방어

llm

14 분 소요

자가 개선(self-improving) 루프는 사람 라벨 없이 프로덕션 트레이스를 학습 신호로 되돌리는 구조다. 롤아웃·스코어·필터·트레인·디플로이의 5단계 아키텍처, 복합 리워드 설계, PII·동의 게이트, 리워드 해킹 탐지와 방어, 조직 차원의 도입 판단 기준을 운영자 시점에서...