AWS Glue에서 ‘Unable to infer schema’ 예외를 해결하려면 어떻게 해야 합니까?

간략한 설명

일반적으로 이 오류는 AWS Glue에서 key=val 구조를 사용하는 Apache Hive 스타일로 파티션된 경로에 저장되지 않은 Parquet 또는 Orc 파일을 읽으려는 경우에 발생합니다. AWS Glue는 Amazon Simple Storage Service(Amazon S3) 소스 파일이 키–값 페어일 것으로 예상합니다. 예를 들어, AWS Glue 작업이 **s3://s3-bucket/parquet-data/**에서 파일을 처리 중인 경우 파일의 파티션 구조는 다음과 같아야 합니다.

s3://s3-bucket/parquet-data/year=2018/month=10/day=10/file1.parquet

Parquet 또는 Orc 파일이 계층 구조로 저장된 경우 AWS Glue 작업은 "Unable to infer schema" 예외로 실패합니다. 예:

s3://s3-bucket/parquet-data/year/month/day/file1.parquet

해결 방법

이 오류를 해결하려면 다음 방법 중 하나를 사용합니다.

데이터 재구성

새 S3 버킷으로 파일을 복사하고 Hive 스타일의 파티션된 경로를 사용합니다. 작업을 다시 실행합니다.

파티션 열 이름을 별표로 대치

데이터 재구성이 가능하지 않은 경우 Amazon S3에서 직접 DynamicFrame을 생성합니다. Amazon S3 경로에서 모든 파티션 열 이름을 별표(*)로 대치합니다. 이 해결 방법을 사용하는 경우 AWS Glue는 DynamicFrame의 파티션 열을 포함하지 않으며, 데이터만 포함합니다.

예를 들어, 다음과 같은 파티션 구조의 파일이 S3 버킷에 저장된다고 가정합니다.

s3://s3-bucket/parquet-data/year/month/day/files.parquet

s3://s3-bucket/parquet-data/ 경로에 있는 모든 파일을 처리하려면 DynamicFrame을 생성합니다.

dynamic_frame0 = glueContext.create_dynamic_frame_from_options('s3',connection_options={'paths':['s3://s3-bucket/parquet-data/*/*/*'],},format="parquet",transformation_ctx = "dynamic_frame0")

맵 변환을 사용하여 파티션 열 추가

DynamicFrame에 파티션 열을 포함하려면 먼저 DataFrame을 생성하고 Amazon S3 파일 경로에 대한 열을 추가합니다. 그런 다음, 다음 예와 같이, DynamicFrame을 생성하고 맵 변환을 적용하여 파티션 열을 추가합니다. 샘플 코드를 사용하기 전에 Amazon S3 경로를 대치하고 올바른 인덱스 값을 사용하여 파티션 열 이름을 입력합니다.

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 awsglue.dynamicframe import DynamicFrame
from pyspark.sql.functions import *
args = getResolvedOptions(sys.argv, ['JOB_NAME'])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
df= spark.read.parquet("s3://s3-bucket/parquet-data/*/*/*")
modified_df = df.withColumn('partitions_column',input_file_name())
dyf_0 = DynamicFrame.fromDF(modified_df, glueContext, "dyf_0")

def modify_col(x):
     if x['partitions_column']:
        new_columns = x['partitions_column'].split('/')
        x['year'],x['month'],x['day'] = new_columns[4],new_columns[5],new_columns[6]
        del x['partitions_column']
     return x

modified_dyf = Map.apply(dyf_0,f=modify_col)
datasink2 = glueContext.write_dynamic_frame.from_options(frame =modified_dyf , connection_type = "s3", connection_options = {"path": "s3://my-output-bucket/output/","partitionKeys": ["year","month","day"]}, format = "parquet", transformation_ctx = "datasink2")

맵 변환 적용에 대한 자세한 내용은 맵 클래스를 참조하세요.

---

관련 정보

AWS Glue에서 ETL 출력값의 파티션 관리