AWS Glue 실전

AWS Glue 이론 포스팅 이후 벌써 반년이 지났네요. 글루가 나온 지 얼마 안 된 상품이어서 그런지 반년 사이에도 많은 업데이트가 있더라고요.. (파이썬 3 지원, 워크 플로우 지원, Continuous logging 지원 등) 

워낙 내용이 방대하다보니 모든 내용을 담기가 어렵더라고요. 그러다 보니 제가 겪었던 경험들을 토대로 포스팅을 진행하려고 합니다. 

다만, 업데이트가 잦다보니 실제 사용하실 때 현재 포스팅과 실제 내용이 다를 수 있다는 점 참조하시면 좋을 것 같습니다.

 

# Glue Job 동작

## 스크립트 구성

Glue는 AWS에서 제공하는 스파크 기반의 ETL 서비스입니다. 우리는 Spark 기반의 스크립트를 작성하여 Glue Job을 이용할 수 있습니다.

AWS Glue에서는 GlueContext라는 SparkContext를 래핑 한 라이브러리를 제공하고 있습니다. 여기서 Spark의 Dataframe이 GlueContext에서는 DynamicFrame으로 변환되어 사용됩니다.

AWS Glue 라이브러리는 여기에서 확인하실 수 있습니다.

 

## 네트워크 통신

Glue에서 다른 데이터 소스/타겟 혹은 다른 서비스에 접근하기 위해서는 VPC를 통한 엔드포인트 설정이 필요합니다.

저 같은 경우는 AWS EC2에 있는 MySQL에서 데이터를 가져와 S3에 Parquet 형태로 저장하는 방식을 취했습니다.
그리고 전송 실패하는 경우, Glue Script에서 바로 AWS SNS로 에러 알람을 전송하도록 구성하였습니다.

이 과정에서 EC2, S3, SNS와 통신 가능하도록 VPC 구성을 진행하였습니다.

Glue Job에서 사용할 VPC 선택은 "Data catalog"에 있는 "Connections" 서비스를 이용하면 됩니다.

Glue Data catalog 탭

여기서 생성한 Connection을 Glue Job Script에 연결하면 해당 vpc를 Glue Job이 사용하게 됩니다.

 

# 스크립트 구성

위 네트워크 섹션에서 언급했듯이, 제가 운영한 서비스의 아키텍처는 다음과 같습니다.

MySQL(EC2) --> Glue --> S3 (Parquet) --> Athena

간략히 설명하면 다음과 같습니다.

위 MySQL에는 하루에도 수많은 데이터가 쌓이고 있어서 디스크 용량 사정상 데이터를 오랫동안 보관하는 것이 힘들었습니다. 그렇기 때문에 Glue를 통해 S3에 적재 후, 일정 주기별로 MySQL 데이터를 비우는 형식으로 작업이 이루어졌습니다.

## 스크립트 구성 팁

스크립트 구성 프로세스

프로그램 언어 선택이나 IAM Role 구성 등 기본적인 내용을 제외한 몇 가지 기억에 남아있는 요소들을 설명하도록 하겠습니다.

• Maximum capacity : Glue의 서비스 단위인 DPU 사용 개수 설정입니다.
• Job timeout : default 값으로 2880분이 설정되어 있습니다. 과거 데이터 처리할 때, 이 값을 설정해주지 않아서 중간에 프로그램이 멈췄던 기억이 있네요.
• Continuous logging : 최근에 추가된 기능입니다. 로깅을 보다 편하게 볼 수 있도록 도와줍니다.
• Job bookmark : JSON 포맷으로 저장하는 경우 북마크 기능이 적용되는 걸 확인했었는데, parquet 포맷은 지원 안된다고 합니다. (2019년 2월 기준)
• python lib 경로 설정 : 단일 py 파일이라면 그냥 업로드 해도 동작하지만, 여러 파일이 있다면 _init_.py가 포함된 zip으로 압축해서 올린 후, 스크립트에서 풀어주어야 합니다.

####### Download and unzip external library files. 
import sys 
import zipfile

zip_ref = zipfile.ZipFile('site-packages.zip', 'r')
zip_ref.extractall('./tmp/packages') 
zip_ref.close() 
sys.path.insert(0, './tmp/packages'

 

## GlueLogger

놀랍게도 Glue 작업 초기에는 파이썬 로깅 기능이 지원되지 않았습니다. 파이썬 내부 라이브러리에 있는 자체 logger를 사용해보려고 했지만, 파이썬 언어 자체에 익숙하지 않아서 그저 print로 출력을 했었습니다.

다행히, 최근 aws-glue-lib에 보면 GlueLogger 기능이 추가되었더라구요. 

구성 코드를 공유 드립니다.

# Logger
logger = glueContext.get_logger()

간단하죠? 위 logger를 사용하면 CloudWatch에서 GlueLogger로 확인하실 수 있습니다.

 

## OOM

py4j.protocol.Py4JJavaError: An error occurred while calling o611.pyWriteDynamicFrame.
: org.apache.spark.SparkException: Job aborted.
Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 38.0 failed 4 times, most recent failure: Lost task 0.5 in stage 38.0 (TID 43, ip-10-10-16-157.ec2.internal, executor 4): ExecutorLostFailure (executor 4 exited caused by one of the running tasks) Reason: Executor heartbeat timed out after 130943 ms

초기에 GlueContext의 메서드를 통하여 JDBC 데이터를 가져오는 방법을 사용하려 하였습니다. 이때, 여기서 수많은 OOM(Out Of Memory) 에러가 발생하였습니다.

원인은 방대한 양의 전체 RDB 데이터를 Glue가 읽으려다 보니, 메모리가 부족해져서 발생한 에러였습니다.

이를 해결하기 위해, GlueContext에서 파티션 된 데이터를 읽어오는 방법을 찾아보기도 하고 AWS Support Case에 문의도 해보았지만, 당시(2019년 02월)에는 Glue에서 해당 기능을 지원하지 않았습니다.

데이터 소스가 RDS와 같이 AWS 내부 상품의 경우 파티션으로 데이터를 나누어 가져오는 게 가능했지만, RDB의 경우는 지원하지 않더라고요.

이 이슈를 해결하기 위하여 선택한 방법은 다음과 같습니다.

• Native Spark를 통하여 파티션 된 데이터 DataFrame에 저장
• Glue의 기능을 통하여 DataFrame을 DynamicFrame으로 전환
• GlueContext로 S3에 저장

코드로 보면 다음과 같습니다.

# Convert DataFrames to AWS Glue's DynamicFrames Object
source_df = get_source(basis_dt, db_name, table_name, jdbc_url)
dynamic_dframe = DynamicFrame.fromDF(source_df, glueContext, "dynamic_df")

# Sink
datasink = glueContext.write_dynamic_frame.from_options(frame=dynamic_dframe, connection_type="s3",
                                                                    connection_options={
                                                                        "path": s3_path},
                                                                    format="parquet",
                                                                    transformation_ctx="datasink")

이때부터 반년이 지났으니, GlueContext를 통하여 파티션된 JDBC 데이터를 가져오는 기능이 추가되었을지 모르겠네요.

 

## Connection Timeout

ConnectTimeoutError: Connect timeout on endpoint URL: "https://sns.eu-central-1.amazonaws.com/"

Glue의 실패 로그를 메일 알람으로 받고 싶어서 SNS에 Publish 하는 코드를 구성하였었습니다.

GCP 같은 경우, 프로젝트 단위로 Native VPC로 구성되어 있기 때문에 프로젝트 내, 대다수 서비스 간의 통신이 자유롭게 되었던 걸로 기억하여 AWS에서도 같은 효과를 기대했었습니다.

근데 같은 IAM Role로 구성된 AWS SNS 토픽에 Publish 과정에서 TCP Timeout 이슈가 발생하더라고요.

알고 보니 ETL에서 사용하는 VPC(Connections에서 설정하는 VPC)에 IGW 설정이 필요하더라고요..

혹시나 같은 이슈로 고생하시는 분이 있으실까 하여 공유드립니다.

 

# 크롤러 구성

개인적으로 Glue가 가지고 있는 가장 강력한 기능은 DataCatalog라고 생각합니다. Glue Crawler는 이 DataCatalog를 만들어주는 역할을 하고 있습니다.

Glue를 통해 S3에 적재한 데이터가 크롤링을 거치지 않으면 Athena에서 조회가 안되기 때문에 크롤링은 필수라 생각됩니다. (Athena에서 DataCatalog DB를 사용하는 경우) 

크롤러 생성에서는 아래와 같은 작업 단계를 거치게 됩니다.

 

## 특징

특정 Source Type 설정

이 부분에서는 크롤링의 성격을 정의하는 단계입니다. Data Stores의 데이터를 크롤링할 것인지, 다른 데이터 카탈로그의 테이블을 크롤링할 것인지 결정할 수 있습니다.

Add Source

위에서 정의한 크롤러 성격에 맞추어 데이터 소스 위치를 정의하는 단계입니다. Data Stores를 선택한다면 S3, JDBC, DynamoDB 등의 옵션이 있고 데이터 카타로그 테이블을 선택한다면 어떠한 테이블을 참조할 것인지 결정해주면 됩니다.

IAM Role 설정

DataCatalog & S3 read/write 등의 권한을 가지고 있는 IAM Role이 필요합니다.

Trigger 등록

Schedule / on Demand 중 원하는 트리거를 등록할 수 있습니다.

Target Database 등록

크롤러가 정리한 데이터를 저장할 데이터 카타로그 데이터베이스를 결정합니다.

 

## Data Catalog Versioning

스키마가 변경되면 Crawler가 이를 수집하고 DataCatalog에서는 자동으로 버전 관리를 해줍니다.

S3에서 데이터를 읽어오는 경우, 크롤러가 자동으로 경로 이름에 따라서 파티션을 구분하여 키를 생성해줍니다.

스키마 업데이트는 다음과 같이 이루어집니다.

## 단점

크롤러를 통하여 Data Catalog에 파티션 추가/스키마 업데이트를 진행하려고 하는 경우 불편한 점이 있습니다.

S3에 있는 데이터의 크롤링을 할  때마다 전체 데이터를 읽어오는 비효율적인 동작을 하는 것입니다.

증분 된 데이터만 가져오는 방법이 없을까 조사해보고 Support Case에 올려보았지만 2019년 2월 기준에서는 지원되지 않는 기능이더라고요..

이 부분이 빨리 개선되었으면 하네요.

 

# 워크 플로우

최근에 새로 추가된 기능입니다. Glue의 각 작업들(Job Script, Crawling)을 하나의 Workflow로 관리할 수 있게 도와줍니다. UI로 제공되어 간편하게 사용할 수 있습니다.

https://docs.aws.amazon.com/glue/latest/dg/workflows_overview.html

 

# 후기

Glue 포스팅 1차 마무리를 여기서 지으려고 합니다. 이전 이론 포스트 내용이 많이 부실한 거 같아서 아쉽네요.

앞으로도 Glue를 통한 서비스를 계속 운영/보수할 예정이니, 새로운 기능이 많이 추가되면 또 다른 포스팅으로 찾아오도록 하겠습니다.

AWS Athena는 이 포스팅에 별책부록 같은 느낌으로 구성할까 하다가 AWS SNS와 같이 별도로 짧게 구성하려고 합니다.