Spark

JSON 문자열 다루기

JSON 데이터가 문자열 컬럼에 들어 있을 때, get_json_object 함수는 JSON 경로 표현식(예: "$.myJSONKey.myJSONValue[1]")으로 딕셔너리나 배열이든 원하는 값을 인라인으로 조회한다. json_tuple은 JSON 객체가 한 단계(single level)만 중첩되어 있을 때 여러 키를 한 번에 꺼내는 데 쓴다.

StructType 컬럼을 JSON 문자열로 직렬화하려면 to_json을 쓴다. 반대로 JSON 문자열을 다시 구조화된 컬럼으로 파싱하려면 from_json을 쓰는데, 이때는 반드시 대상 스키마(StructType)를 지정해야 하며 선택적으로 옵션 맵도 함께 지정할 수 있다.

UDF 등록과 실행 과정

UDF(User-Defined Function)는 Python이나 Scala로 직접 작성한 커스텀 변환 로직으로, numpy 같은 외부 라이브러리까지 활용할 수 있다. UDF는 하나 이상의 컬럼을 입력받아 하나 이상의 컬럼을 반환하는 함수이며, 특정 DSL이 아니라 그냥 일반적인 함수로 작성한다.

함수를 정의한 뒤에는 Spark에 등록해야 모든 워커 머신에서 사용할 수 있다. 등록 시 Spark는 드라이버에서 함수를 직렬화해 네트워크를 통해 모든 익스큐터 프로세스로 전달하는데, 이는 사용 언어와 무관하게 항상 일어난다.

Scala/Java로 작성한 UDF는 JVM 안에서 그대로 실행되므로 성능 손해가 크지 않다(다만 Spark의 내장 함수처럼 코드 생성 최적화의 이점은 받지 못하고, 객체를 많이 생성/사용하면 성능 이슈가 생길 수 있다). 반면 Python으로 작성한 UDF를 실행할 때는, 워커에서 별도의 Python 프로세스를 새로 띄우고, JVM에 있던 데이터를 Python이 이해할 수 있는 포맷으로 직렬화해 그 프로세스로 넘긴 뒤 행 단위로 함수를 실행하고, 결과를 다시 JVM/Spark로 돌려받는 과정을 거친다. 이 과정에서 프로세스 기동 비용도 있지만 진짜 비용은 데이터 직렬화이며, 데이터가 Python 프로세스로 넘어가는 순간부터는 Spark가 그 워커의 메모리를 관리할 수 없게 되어(JVM과 Python이 같은 머신의 메모리를 두고 경쟁) 자원 부족으로 워커가 실패할 위험도 있다. 그래서 책은 UDF를 Scala로 작성하기를 권장하며, Scala로 작성해도 Python에서 그대로 호출해 쓸 수 있다는 이점을 강조한다.

DataFrame 함수 vs SQL 함수 등록, 반환 타입 지정

udf() 함수로 등록한 UDF는 기본적으로 DataFrame 함수로만 쓸 수 있어 문자열 SQL 표현식 안에서는 쓸 수 없다. spark.udf.register("이름", 함수)로 등록하면 Spark SQL 함수로도 등록되어, SQL 문자열 표현식(selectExpr, %sql) 안에서도 쓸 수 있고 언어 간에도 공유해 쓸 수 있다 - 예를 들어 Scala에서 등록한 함수를 Python의 selectExpr에서 그대로 호출할 수 있다.

Python 함수는 Spark 고유의 타입 시스템과 정확히 일치하지 않으므로, 함수를 정의할 때 반환 타입을 명시하는 것이 모범 사례다(필수는 아니지만 권장). 만약 실제로 반환되는 값의 타입이 명시한 타입과 맞지 않으면 Spark는 에러를 내지 않고 그냥 null을 반환해 실패를 나타낸다. 예를 들어 spark.range로 만든 정수 컬럼에 DoubleType으로 반환 타입을 지정한 Python 함수를 적용하면, Python이 정수를 자동으로 float로 바꿔주지 않기 때문에 null이 나온다 - 함수가 실제로 float를 반환하도록 고쳐야 정상 동작한다.