Apache Hudi on AWS Glue에서 한글컬럼명을 조건절에 사용하는 방법

문제 현상

AWS Glue 3.0 또는 AWS Glue Interactive Session에서 아래 SparkSQL을 수행하면 에러가 발생합니다.

hudi_recordkey = "거래번호"
hudi_precombine = "판매량"
hudi_partition = "거래일자"
spark.sql(f"alter table hudi_db.sales drop partition (`{hudi_partition}` = '01-02-2020') ")

위 코드에서 {hudi_partition} 처럼 한글에 Backtick(`)을 사용할 수 있지만, SparkSQL을 사용할 때 아래와 같은 에러가 발생됩니다.

Py4JJavaError: An error occurred while calling o76.sql.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 132.0 failed 4 times, most recent failure: Lost task 0.3 in stage 132.0 (TID 4382) (172.35.253.236 executor 1): java.lang.RuntimeException: java.net.URISyntaxException: Illegal character in path at index 47: s3://aws-glue-workshop/hudi/base_table/거래일자=01-02-2020/8cbefd48-eded-42b2-8f34-7abfdb43eda8-0_352-24-765_20230704081519995.parquet
	at org.apache.hudi.common.table.timeline.dto.FilePathDTO.toPath(FilePathDTO.java:54)
	at org.apache.hudi.common.table.timeline.dto.FileStatusDTO.toFileStatus(FileStatusDTO.java:102)
	at 

위 에러 로그 "java.net.URISyntaxException: Illegal character in path"에서 확인할 수 있는 것처럼, Apache Hudi 데이터가 저장되어 있는 Amazon S3 경로에 포함된 한글 폴더명이 깨져있습니다.

원인

아래 설명에서 보시는 것처럼, 이러한 에러가 발생되는 원인은 AWS Glue 3.0에서 지원하는 Apache Hudi 버전(0.10.1)의 한계에 따른 것입니다. 현재까지 AWS Glue Interactive Session은 AWS Glue 3.0을 지원하기 때문에 동일한 에러가 발생됩니다.

Bug : Apache Hudi/[HUDI-4199](https://issues.apache.org/jira/browse/HUDI-4199)에 따르면,
url encoding이 지원되지 않습니다.
Apache Hudi Fix Version : 0.12.1

따라서, Apache Hudi Version : 0.10.1에서는 Amazon S3 경로 내에 있는 한글폴더명을 인식하지 못하게 됩니다. (예를 들면, Amazon S3 파티션 폴더명이 "거래일자=2023-07-04"인 경우)

참고로, AWS Glue 버전별로 지원하는 Apache Hudi 버전 정보는 여기를 참고하실 수 있습니다.

해결방법

AWS Glue 버전별로 지원하는 Apache Hudi 버전 정보에서 확인할 수 있는 것처럼, AWS Glue 4.0은 Apache Hudi 0.12.1을 지원합니다. 따라서, AWS Glue 4.0을 이용하면 위와 같은 오류없이 조건절에 한글컬럼명인 경우에도 정상적으로 수행할 수 있습니다.

예제 코드

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import concat, col, lit

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

DELETE_IDS_S3_LOCATION = 's3://aws-glue-workshop-icn/raw/deletes/'
BASE_TBL_S3_LOCATION = 's3://aws-glue-workshop-icn/hudi/base_table/'
delete_location = f"{DELETE_IDS_S3_LOCATION}"
hudi_tbl_location = f"{BASE_TBL_S3_LOCATION}"

hudi_database = "hudi_db"
hudi_table = "sales"
hudi_recordkey = "거래번호"
hudi_precombine = "판매량"
hudi_partition = "거래일자"

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

job.init(args['JOB_NAME'], args)

# Apache Hudi 파티션 삭제할 때(Amazon Athena에서도 해당 명령으로 파티션 삭제 가능함)
# spark.sql(f"alter table hudi_db.sales drop partition (`{hudi_partition}` = '06-11-2020') ")

# 레코드 삭제할 때
spark.sql(f"select * from hudi_db.sales where `거래번호`='15-11-2020-8532-11-2'").show()
spark.sql(f"select * from hudi_db.sales").count()

# 삭제 대상이 별도 파일로 관리되는 경우
# deleteDF=spark.read.format("csv").option("header", True).load(f"{delete_location}")
# deleteDF.show()

# Apache Hudi 테이블을 DataFrame으로 생성하는 경우 : Apache Hudi 자체 컬럼은 drop으로 삭제
dropColumnList = ['_hoodie_commit_time','_hoodie_commit_seqno','_hoodie_record_key','_hoodie_partition_path','_hoodie_file_name']

# 삭제 대상 DataFrame 생성
deleteDF = spark.sql(f"select * from hudi_db.sales where `거래번호`='15-11-2020-8532-11-2'").drop(*dropColumnList)

deleteDF.write \
.format('hudi') \
.option('hoodie.datasource.write.operation', 'delete') \
.options(**hudiOptions) \
.mode('append') \
.save()

spark.sql(f"select * from hudi_db.sales").count()
# Athena에서 정상적으로 삭제되었는지 확인할 수 있음

job.commit()