오늘은 최근 만들고 있었던 재무와 회계를 위한 매출신고용 데이터마트의 2차 수정사항을 마치고 Workflow가 설정한대로 잘 움직이고 있는지 마지막 테스트를 진행하고 있었다. 내가 체크할 건 몇 가지 없었다. Unity Catalog로 잘 저장되고 있는지 그 부분만 확인하면 충분하다고 생각했고, 마지막 테스트를 진행했다. 하지만 늘 그렇듯 자신감 뒤에는 상상도 못한 에러가 날 기다리고 있었다.
- 저기요, 선생님 이게 무슨 말이에요. 우리 Decimal 을 Loading 할 수 없다니요....
문제
Error (code 1207) while loading data into Redshift:
"Invalid digit, Value '"', Pos 0, Type: Decimal"
이번에 작업하는 데이터마트에는 사칙연산이 많다. 소계도 많다. 정말 많다. 그래서 어쩔 수 없이 부동소수점 이슈를 줄이고자 Decimal 타입을 선택했는데, 완성한 데이터를 Redshift에 넣으려니까 말썽이었다.
원인
When writing to Redshift, data is first stored in a temp folder in S3 before being loaded into Redshift. The default format used for storing temp data between Apache Spark and Redshift is Spark-Avro. However, Spark-Avro stores a decimal as a binary, which is interpreted by Redshift as empty strings or nulls.
출처 : redshift-fails-decimal-write
문제는 다음과 같았다. Spark는 자신이 다루는 데이터가 커지면 .option("tempdir", "<S3경로>")로 설정한 곳에 임시 데이터를 쌓아두고 천천히 연산하는 방식을 사용한다. 평소에 tempdir로 지정한 곳에 쌓이는 데이터 형식이 대체 뭔가 싶었는데 Avro였다는 사실을 알게 된 것도 좋은 소득이었지만, Avro는 Decimal을 다룰 때 바이너리로 변환하고 이게 null 혹은 공백이기에 발생한 에러였다.
해결 & 느낀점
.option("tempformat", "CSV")
이를 해결하기 위해 다음과 같이 수정했다. 원인에 비해 해결법은 간단했다. 위 코드만 write 하는 곳에 붙여넣어주기만 하면 해결된다.
Avro나 CSV나 둘다 행 지향인 데이터타입인지는 알았지만, 타입별로 다루는 방식이 이렇게까지 다르다는 걸 몸으로 체득했다. 여유가 되면 Spark에서 굳이 Avro를 default tempformat으로 설정해둔 이유와 CSV로 설정했을 때 또다른 문제는 없는지 확인할 예정이다.
- 콩쥐야 우리 큰일났어... metadata 가 설정이 안 돼...
위 에러까지 깔끔하게 해결한 후에 무서울 게 없었던 나는 호기롭게 Databricks의 클러스터 버전을 최신 LTS로 올려버렸다. 어차피 나중에 올려야한다면 지금 매를 먼저 맞는 게 낫다고 생각했기 때문이다. 그리고 나름대로 DLT에서도 테스트를 마쳤고, 겨우 Job Cluster인데 무슨 일이 생길까 싶었다.
하지만 겨우가 나를 아주 크게 배신했고, 오늘은 종일 이 문제만 해결하다가 이 포스팅까지 쓰게 된 계기가 되었다.
해당 프로젝트의 데이터 특성상 String의 maxlength 길이인 256을 훌쩍 넘는 상품명이 정말 많다. 그래서 무조건 메타데이터 설정을 바꿔줘야만 하는데 14.3으로 올렸더니 기존 방식이 전혀 통하지 않았다.
기존
from pyspark.sql import functions as F
# maxlength를 늘린 새로운 메타데이터 정의
new_meta = {"maxlength": 2000}
# withColumn을 사용하여 컬럼의 메타데이터 업데이트
df = df.withColumn("컬럼명", F.col("컬럼명").alias("컬럼명", metadata=new_meta))
기존에는 withColumn 를 통해서 metadata 를 수정해주고 있었다. 아뿔싸 databricks 14.3 LTS 에서는 이 방법이 전혀 통하지 않았다.
시도 1
df.schema["컬럼명"].metadata = {"maxlength": 2000}
결과는 무려 실패
spark 버전도 같이 올라가면서 억지로 metadata 를 설정하는 방식은 전혀 되지 않았다. 애초에 될 거라고 기대한 방법도 아니다.
시도 2 - 해결
set_maxlength = {"maxlength": 2000}
df_meta = df.withMetadata('컬럼명', set_maxlength)
df_meta.schema['age'].metadata
출처 : Spark withMetaData
14.3 버전에서는 Spark 버전이 3.5.0이었고, 위 기능은 3.3.0 ~ 3.4.0을 거치면서 안정화된 듯 보였다. 이렇게 메타데이터를 변경하고서는 무사히 Redshift에 write 하는 Task까지 완성시킬 수 있었다.