앞서 소개했던 ETL은 상당한 공수를 요구한다. ETL은 물론 중요한 작업이고 설계 및 아키텍쳐도 중요하지만 특별한 알고리즘을 요구하거나 분석 모델을 필요로 하지는 않는 작업이기 때문에 최대한 공수를 줄이는 것이 좋다.

이를 위해 AWS에서는 ETL 전용 서버리스 툴 Glue를 제공한다. 데이터 파이프라인을 구축하는데 방법을 찾던 중 Glue에 대해 조사하고 공부했던 기록을 남기고자 한다.

Glue Workflow

Glue의 Workflow 자체는 생각보다 그렇게 복잡하지 않다.

전체 과정은 Tutorial을 보는 것이 훨씬 좋으므로 참고하자.

https://youtu.be/taR2hRZ2AwI?t=936

1. Crawler

크롤러라고 썼지만, 정확히는 Data Catalog를 구성하는 데 가장 중요한 것이 Crawler이고 주목적은 Data Catalog를 구성하는 것이다.

크롤러가 실행되며 데이터 추출을 해올 DB 혹은 파일의 컬럼 정보, row 수 및 기본적인 통계 정보를 담은 메타 데이터 테이블이 생성된다. 이런 메타 데이터 테이블과 커넥션 정보, 크롤러, Setting 등이 모여 Data Catalog 메뉴를 구성하고 있다.

2. Glue Job

실질적인 ETL 작업을 담당하는 Job이다. 크롤러에 의해 만들어진 메타 테이블을 바라보고 있고 설정한 정보 및 ETL 스크립트를 가지고 ETL을 수행한다.

properties

대부분 바꿀 필요가 없거나 직관적으로 이해할 수 있는 설정들이지만 몇 가지 설정들만 추려본다.

job이 도는 방법, ETL 스크립트을 선택한다.Glue에서 제공해주는 기본 스크립트를 사용할지, 기존에 있는 스크립트를 선택할지, 새로 작성할지 선택할 수 있다.

Job이 돌 때 모니터링 방법을 선택할 수 있다.

Job Metrics : CloudWatch에서 모니터링한다. 단 Job이 완전히 끝난 이후에 반영된다.
Continuous logging : Job이 돌 때를 포함해 실시간으로 모니터링이 된다.
Spark UI : Spark UI라는 웹 UI를 사용한다.

Job이 돌 때 할당될 Resource를 선택할 수 있다.

Standared : 4 vCPU, 16 GB of memory and a 50GB disk, and 2 executors per worker
G.1X : each worker maps to 1 DPU (4 vCPU, 16 GB of memory, 64 GB disk), and provides 1 executor per worker
G.2X : each worker maps to 2 DPU (8 vCPU, 32 GB of memory, 128 GB disk), and provides 1 executor per worker

ETL Script

Job을 생성한 뒤 edit Script를 이용해 스크립트를 작성할 수 있다.

기본적으로 Glue가 제공하는 ETL 스크립트이다.

우측 상단 Transform을 이용해 dataframe 형태를 원하는대로 변경할수도 있지만, 이 스크립트 자체가 직관적이지 않아보여서 pyspark function을 이용해 직접 작성하는 것을 권장한다.

Script Example

from pyspark.context import SparkContext
import pyspark.sql.functions as f

# Import glue modules
from awsglue.utils import getResolvedOptions
from awsglue.context import GlueContext
from awsglue.dynamicframe import DynamicFrame
from awsglue.job import Job

# Glue Job ETL Script

# Initialize contexts and session
spark_context = SparkContext.getOrCreate()
glue_context = GlueContext(spark_context)
session = glue_context.spark_session

# parameter
glue_db = "test-parksw2" # catalog db 이름
movie_table = "movies_csv" # table 이름
rating_table = "ratings_csv" # table
s3_write_path = "s3://aws-glue-test-ap-northeast-2-output" # output을 저장한 s3 bucket


dynamic_frame_read_movie = glue_context.create_dynamic_frame.from_catalog(database=glue_db, table_name=movie_table)
dynamic_frame_read_rating = glue_context.create_dynamic_frame.from_catalog(database=glue_db, table_name=rating_table)

movie_data_frame = dynamic_frame_read_movie.toDF()
rating_data_frame = dynamic_frame_read_rating.toDF()

merged_frame = movie_data_frame.mergeDynamicFrame(rating_data_frame, ['movieid'])

dynamic_frame_write = DynamicFrame.fromDF(merged_frame, glue_context, "dynamic_frame_write")

glue_context.write_dynamic_frame.from_options(
    frame=dynamic_frame_write,
    connection_type="s3",
    connection_options= {
        "path": s3_write_path,
    },
    format="csv"
)

기타 extension은 AWS 공식문서를 참고하자.

Job을 실행하게 되면 Script 내용에 맞게 데이터를 가져온 뒤 가공해 지정된 위치에 적재까지 완료된다.