정현석

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

6 분 소요

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

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

기본 실행 방법

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

[Spark] 불리언, 수치, 문자, 정규 표현식 등의 데이터 타입 다루기에 이어서 진행됩니다.

6. 날짜와 타임스탬프 데이터 타입 다루기

  • 스파크는 달력 형태의 날짜(date)와 날짜와 시간 정보를 모두 가지는 타임스탬프(timestamp) 두 가지 종류의 시간 관련 정보만 집중적으로 관리한다.
  • 스파크는 inferSchema 옵션이 활성화된 경우 날짜와 타임스탬프를 포함해 컬럼의 데이터 타입을 최대한 정확하게 식별하려 노력한다.
  • 스파크는 특정 날짜 포맷을 명시하지 않아도 자체적으로 식별해 데이터를 읽을 수 있다.
  • 스파크는 날짜와 시간을 최대한 올바른 형태로 읽기 위해 노력한다.
    • 만약 특이한 포맷의 날짜와 시간 데이터를 어쩔 수 없이 다뤄야 한다면 각 단계별로 어떤 데이터 타입과 포맷을 유지하는지 정확히 알고 트랜스포메이션을 적용해야 한다.
  • 다음은 오늘 날짜와 현재 타임스탬프값을 구하는 예제이다.
import org.apache.spark.sql.functions.{current_date, current_timestamp}

val dateDF = spark.range(10
    ).withColumn("today", current_date()
    ).withColumn("now", current_timestamp())
dateDF.createOrReplaceTempView("dateTable")
  • 위 코드를 통해 만들어진 DataFrame을 사용해 오늘을 기준으로 5일 전후의 날짜를 구해본다.
    • date_sub(date, num) : 컬럼과 뺄 날짜 수를 인수로 전달한다.
    • date_add(date, num) : 컬럼과 더할 날짜 수를 인수로 전달한다.
    • 두 함수는 스파크 SQL 내장 함수로 제공된다. 1
import org.apache.spark.sql.functions.{date_sub, date_add}

dateDF.select(date_sub(col("today"), 5), date_add(col("today"), 5)).show(1) // 지금은 9월 4일이다.

  • 실행결과

    image

  • 두 날짜의 차이를 구하는 작업을 살펴본다.
  • datediff(date1, date2) : 두 날짜 사이의 일 수를 반환해준다.
  • months_between(date1, date2) : 두 날짜 사이의 개월 수를 반환해준다.
  • to_date() : string 포맷을 DateType으로 변환해준다. 2
import org.apache.spark.sql.functions.{datediff, months_between, to_date}

dateDF.withColumn("week_ago", date_sub(col("today"), 7)
    ).select(datediff(col("week_ago"), col("today"))).show(1)
dateDF.select(
    to_date(lit("2022-01-01")).alias("start"),
    to_date(lit("2022-09-01")).alias("end")
    ).select(months_between(col("start"), col("end"))).show(1)

  • 실행결과

    image

  • 스파크는 날짜를 파싱할 수 없다면 에러 대신 null 값을 반환한다. 이는 기본 설정인 spark.sql.ansi.enabled=false에서의 동작이며, ANSI 모드를 켜면 형식이 어긋난 입력이 런타임 예외로 바뀐다. 3

dateDF.select(to_date(lit("2022-20-12")), to_date(lit("2022-09-01"))).show(1)

  • 실행결과

    image

  • to_date 함수는 필요에 따라 날짜 포맷을 지정할 수 있지만 to_timestamp 함수는 반드시 날짜 포맷을 지정해야 한다. 포맷을 생략하면 col.cast("timestamp")와 같은 기본 캐스팅 규칙으로 해석한다. 4

import org.apache.spark.sql.functions.{to_date}

val dateFormat = "yyyy-dd-MM"
val cleanDateDF = spark.range(1).select(
    to_date(lit("2022-12-11"), dateFormat).alias("date"),
    to_date(lit("2022-20-12"), dateFormat).alias("date2")
)
cleanDateDF.createOrReplaceTempView("dateTable2")
  • 항상 날짜 포맷을 지정해야 하는 to_timestamp 함수의 예제를 살펴본다.
import org.apache.spark.sql.functions.{to_timestamp}

cleanDateDF.select(to_timestamp(col("date"), dateFormat)).show()

  • 실행결과

    image

7. null 값 다루기

  • DataFrame에서 빠져 있거나 비어 있는 데이터를 표현할 때는 항상 null 값을 사용하는 것이 좋다.
  • 스파크에서는 빈 문자열이나 대체 값 대신 null 값을 사용해야 최적화를 수행할 수 있다. 5
    • null은 값이 아니라 “알 수 없음”을 뜻하므로 =·> 같은 비교 연산의 결과도 null이 되고, 조건절에서 null 로우는 결과에 포함되지 않는다. 값의 동등 비교가 필요하면 null-safe equal 연산자 <=>를 사용한다. 5
  • DataFrame의 하위 패키지인 .na를 사용하는 것이 DataFrame에서 null 값을 다루는 기본 방식이다.

7.1 coalesce

  • coalesce 함수는 인수로 지정한 여러 컬럼 중 null이 아닌 첫 번째 값을 반환한다.
    • 모든 컬럼이 null이 아닌 값을 가지는 경우 첫 번째 컬럼의 값을 반환한다.
import org.apache.spark.sql.functions.coalesce

val df = spark.read.format("csv"
  ).option("header", "true"
  ).option("inferSchema", "true"
  ).load("./data/retail-data/by-day/2010-12-01.csv")
df.createOrReplaceTempView("dfTable")

df.select(coalesce(col("Description"), col("CustomerId"))).show(5)

  • 실행결과

    image

7.2 ifnull, nullif, nvl, nvl2

  • ifnull 함수는 첫 번째 값이 null이면 두 번째 값을 반환한다.
    • 첫 번째 값이 null이 아니면 첫 번째 값을 반환한다.
  • nullif 함수는 두 값이 같으면 null을 반환한다.
    • 두 값이 다르면 첫 번째 값을 반환한다.
  • nvl 함수는 첫 번째 값이 null이면 두 번째 값을 반환한다.
    • 첫 번째 값이 null이 아니면 첫 번째 값을 반환한다.
  • nvl2 함수는 첫 번째 값이 null이 아니면 두 번째 값을 반환한다.
    • 첫 번째 값이 null이면 세 번째 인수로 지정된 값을 반환한다.
spark.sql("""
    SELECT
        ifnull(null, 'return_value'),
        nullif('value', 'value'),
        nvl(null, 'return_value'),
        nvl2('not_null', 'return_value', 'else_value')
    FROM dfTable LIMIT 1
""")

7.3 drop

  • drop 메서드는 기본적으로 null 값을 가진 모든 로우를 제거한다. 인수를 생략한 df.na.drop()은 any를 지정한 것과 같다. 6
    • drop 메서드의 인수로 any를 지정한 경우 로우의 컬럼값 중 하나라도 null 값을 가지면 해당 로우를 제거한다.
    • all을 지정한 경우 모든 컬럼의 값이 null이거나 NaN인 경우에만 해당 로우를 제거한다.
df.na.drop()

df.na.drop("any")

df.na.drop("all")
  • drop 메서드에 배열 형태의 컬럼을 인수로 전달해 적용할 수도 있다.
df.na.drop("all", Seq("StockCode", "InvoiceNo"))

7.4 fill

  • fill 함수를 사용해 하나 이상의 컬럼을 특정 값으로 채울 수 있다.
  • 또한 스칼라 Map 타입을 사용해 다수의 컬럼에 fill 메서드를 적용할 수도 있다.
    • 키(key) : 컬럼명
    • 값(value) : null 값을 채우는 데 사용할 값
df.na.fill("All Null values become this string")

df.na.fill(5, Seq("StockCode", "InvoiceNo"))

val fillColValues = Map("StockCode" -> 5, "Description" -> "No Value")

7.5 replace

  • replace를 사용하면 조건에 따라 다른 값으로 대체할 수 있다.
df.na.replace("Description", Map("" -> "UNKNOWN"))

8. 정렬하기

  • asc_nulls_first, desc_nulls_first, asc_nulls_last, desc_nulls_last 함수를 사용해 DataFrame을 정렬할 때 null 값이 표시되는 기준을 지정할 수 있다.
  • 별도 지정이 없으면 asc는 null 값을 먼저, desc는 null 값을 나중에 배치한다. 두 방향에서 null 위치를 같게 맞추려면 위 함수를 명시해야 한다. 15

9. 복합 데이터 타입 다루기

  • 복합 데이터 타입에는 구조체(struct), 배열(array), 맵(map)이 있다.

9.1 구조체

  • 구조체는 DataFrame 내부의 DataFrame이다.
  • 쿼리문에서 다수의 컬럼을 괄호로 묶어 구조체를 만들 수 있다.
import org.apache.spark.sql.functions.struct

val complexDF = df.select(struct("Description", "InvoiceNo").alias("complex"))
complexDF.createOrReplaceTempView("complexDF")
  • 위에서 복합 데이터 타입을 가진 DataFrame을 만들어 보았다.
  • 이를 다른 DataFrame을 조회하는 것과 동일하게 사용하면 되지만, 유일한 차이점으로 .을 사용하거나 getField 메서드를 사용한다는 것이다.
complexDF.select("complex.Description")
complexDF.select(col("complex").getField("Description"))
  • * 문자를 사용해 모든 값을 조회할 수 있으며, 모든 컬럼을 DataFrame의 최상위 수준으로 끌어올릴 수 있다.
complexDF.select("complex.*")

9.2 배열

split

  • 배열로 변환하려면 split 함수를 사용한다.
  • split 함수에 구분자(delimiter)를 인수로 전달해 배열로 변환한다.
import org.apache.spark.sql.functions.split

df.select(split(col("Description"), " ")).show(2)

  • 실행결과

    image

  • split 함수는 문자열 컬럼을 배열 컬럼으로 바꾸므로, 뒤이어 size, array_contains, explode 같은 배열 함수를 그대로 이어 붙일 수 있다.
  • 파이썬과 유사한 문법을 사용해 배열값을 조회할 수 있다.
df.select(split(col("Description"), " ").alias("array_col")
    ).selectExpr("array_col[0]").show(5)

  • 실행결과

    image

배열의 길이

  • 배열의 크기(size)를 조회해 배열의 크기를 알 수 있다.
import org.apache.spark.sql.functions.size

df.select(size(split(col("Description"), " "))).show(5)

  • 실행결과

    image

array_contains

  • array_contains 함수를 사용해 배열에 특정 값이 존재하는지 확인할 수 있다.
import org.apache.spark.sql.functions.array_contains

df.select(array_contains(split(col("Description"), " "), "WHITE")).show(5)

  • 실행결과

    image

explode

  • explode 함수는 배열 타입의 컬럼을 입력받은 후 입력된 컬럼의 배열값에 포함된 모든 값을 로우로 반환한다. 나머지 컬럼값은 중복되어 표시된다.

image

import org.apache.spark.sql.functions.{split, explode}

df.withColumn("splitted", split(col("Description"), " ")
    ).withColumn("exploded", explode(col("splitted"))
    ).select("Description", "InvoiceNo", "exploded").show(5)

  • 실행결과

    image

9.3 맵

  • 맵(map)은 map 함수와 컬럼의 키-값 쌍을 이용해 생성한다. 그리고 배열과 동일한 방법으로 값을 선택할 수 있다.
import org.apache.spark.sql.functions.map

df.select(map(col("Description"), col("InvoiceNo")).alias("complex_map")).show(5)

  • 실행결과

    image

  • 적합한 키를 사용해 데이터를 조회할 수 있으며, 해당 키가 없다면 null 값을 반환한다.

df.select(map(col("Description"), col("InvoiceNo")).alias("complex_map")
    ).selectExpr("complex_map['WHITE METAL LANTERN']").show(5)

  • 실행결과

    image

10. JSON 다루기

  • 스파크에서는 문자열 형태의 JSON을 직접 조작할 수 있으며, JSON을 파싱하거나 JSON 객체로 만들 수 있다.
    • JSON : 속성-값 쌍, 배열 자료형 또는 기타 모든 시리얼화 가능한 값 또는 “키-값 쌍”으로 이루어진 데이터 오브젝트를 전달하기 위해 인간이 읽을 수 있는 텍스트를 사용하는 개방형 표준 포맷이다. 7
  • 다음은 JSON 컬럼을 생성하는 예제이다.
val jsonDF = spark.range(1).selectExpr("""
    '{"myJSONKey" : {"myJSONValue" : [1, 2, 3]}}' as jsonString
""")
  • get_json_object 함수로 JSON 객체를 인라인(inline) 쿼리로 조회할 수 있다.
  • 중첩이 없는 단일 수준의 JSON 객체라면 json_tuple을 사용할 수도 있다.
import org.apache.spark.sql.functions.{get_json_object, json_tuple}

jsonDF.select(
    get_json_object(col("jsonString"), "$.myJSONKey.myJSONValue[1]") as "column",
    json_tuple(col("jsonString"), "myJSONKey")
).show(2, false)

  • 실행결과

    image

  • to_json 함수를 사용해 StructType을 JSON 문자열로 변경할 수 있다.

    • to_json 함수에 JSON 데이터소스와 동일한 형태의 딕셔너리(맵)를 파라미터로 사용할 수 있다.
import org.apache.spark.sql.functions.to_json

df.selectExpr("(InvoiceNo, Description) as myStruct"
    ).select(to_json(col("myStruct")))
  • from_json 함수를 사용해 JSON 문자열을 다시 객체로 변환할 수 있다.
    • 파라미터로 반드시 스키마를 지정해야 한다. 8
import org.apache.spark.sql.functions.from_json
import org.apache.spark.sql.types._

val parseSchema = new StructType(Array(
    new StructField("InvoiceNo", StringType, true),
    new StructField("Description", StringType, true)
))
df.selectExpr("(InvoiceNo, Description) as myStruct"
    ).select(to_json(col("myStruct")).alias("newJSON")
    ).select(from_json(col("newJSON"), parseSchema), col("newJSON")).show(2, false)

  • 실행결과

    image

11. 사용자 정의 함수

  • 사용자 정의 함수(UDF, user defined function)는 파이썬이나 스칼라 그리고 외부 라이브러리를 사용해 사용자가 원하는 형태로 트랜스포메이션을 만들 수 있게 한다.

  • UDF는 기본적으로 특정 SparkSession이나 Context에서 사용할 수 있도록 임시 함수 형태로 등록된다.

  • UDF를 생성해 스파크에 등록하고, 등록한 함수로 코드를 실행하는 과정에서 어떤 일이 발생하는지 알아본다. 9
  • 먼저 함수를 정의해 입력값을 원하는 결과로 만들어 낸다.
val udfExampleDF = spark.range(5).toDF("num")
def power3(number:Double):Double = number * number * number
power3(2.0)

  • 실행결과

    Screenshot from 2022-09-05 01-44-14

  • 이제 함수를 만들고 테스트를 완료했으므로 모든 워커 노드에서 생성된 함수를 사용할 수 있도록 스파크에 등록할 차례이다.

  • 스파크는 드라이버에서 함수를 직렬화하고 네트워크를 통해 모든 익스큐터 프로세스로 전달한다.

  • UDF를 실행해본다. 먼저 DataFrame에서 사용할 수 있도록 함수를 등록한다.
  • 이제 power3 함수를 DataFrame의 다른 함수와 동일한 방법으로 사용할 수 있다.
import org.apache.spark.sql.functions.udf

val power3udf = udf(power3(_:Double):Double)
  • 스칼라를 사용해 사용자 정의 함수를 등록해본다.
spark.udf.register("power3", power3(_:Double):Double)
udfExampleDF.selectExpr("power3(num)").show(5)

  • 실행결과

    Screenshot from 2022-09-05 01-51-59

12. 한계와 트레이드오프

  • 날짜 파싱 실패는 조용한 null로 끝난다 : to_date(lit("2022-20-12"))처럼 형식이 어긋난 문자열은 예외 없이 null이 된다. 이 글의 실행 결과도 기본 설정(spark.sql.ansi.enabled=false) 기준이며, ANSI 모드를 켜면 같은 입력이 런타임 예외로 바뀐다. 날짜 컬럼을 적재한 뒤에 null 비율을 따로 확인하는 편이 안전하다. 3
  • to_date와 to_timestamp의 인자 요구가 다르다 : to_date는 포맷을 생략할 수 있지만 to_timestamp는 포맷 없이 호출하면 col.cast("timestamp")와 같은 캐스팅 규칙을 따른다. 파일마다 날짜 표기가 다르면 포맷 문자열을 명시해야 한다. 4
  • null의 기본 정렬 위치는 방향마다 다르다 : 별도 지정이 없으면 asc는 null을 먼저, desc는 null을 나중에 배치하므로 두 방향에서 같은 순서를 기대하면 안 된다. 순서를 고정하려면 asc_nulls_first·desc_nulls_last 같은 함수를 명시해야 한다. 15
  • explode는 로우 수를 늘린다 : 배열 원소가 n개인 로우 하나가 n개 로우로 복제되므로, explode 뒤에 집계를 붙이면 원본 로우 수를 기준으로 센 값과 달라진다. 1
  • UDF는 모든 익스큐터로 직렬화되어 전달된다 : 함수와 그 함수가 캡처한 객체가 드라이버에서 직렬화돼 익스큐터로 전송되므로, 무거운 외부 라이브러리를 참조하거나 큰 객체를 캡처할수록 전송 부담이 커진다. 반복 호출되는 변환이라면 내장 함수로 표현할 수 있는지 먼저 검토하는 편이 좋다. 9
  • 실행 조건을 밝혀 둔다. 이 글의 실행 결과는 스파크 3.3.0과 책의 retail 데이터(2010-12-01.csv)라는 조건에서 얻은 것이며, 데이터 크기·파티션 수·스파크 버전이 달라지면 결과와 성능도 달라질 수 있다.

References

  1. Apache Spark - Built-in Functions (Spark SQL API Reference) ↩ ↩2 ↩3 ↩4

  2. Spark to_date() – Convert String to Date format ↩

  3. Apache Spark - ANSI Compliance (Spark SQL Guide) ↩ ↩2

  4. Apache Spark - Datetime Patterns for Formatting and Parsing ↩ ↩2

  5. Apache Spark - NULL Semantics (Spark SQL Guide) ↩ ↩2 ↩3 ↩4

  6. Apache Spark - DataFrameNaFunctions (Scala API) ↩

  7. 위키백과 - JSON ↩

  8. Apache Spark - JSON Files (Spark SQL Data Sources) ↩

  9. Apache Spark - Scalar User Defined Functions (UDFs) ↩ ↩2

댓글남기기

참고

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

llm

14 분 소요

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