AWS glue에서 spark와 notebook 활용하기

남영민·2021년 10월 5일

aws

열명 변경 처리를 하기 위해 AWS glue 서비스를 활용하여 spark와 notebook을 사용하는 방법에 대해 정리합니다.

엔드포인트 추가

노트북과 연결하기 위한 엔드포인트를 추가합니다.
S3, sagemaker등 필요한 서비스에 접근할 수 있는 역할을 생성 후 연결합니다.
네트워킹 단계에서 rds에 접근이 필요한 경우 rds의 vpc와 서브넷 및 보안 그룹을 넣어줍니다. 여기에서 보안 그룹의 경우, rds와 연결된 보안 그룹의 inbound에 보안 그룹의 id를 ip입력란에 넣어주어 포트를 열어줘야 합니다.

위에서 생성한 엔드포인트와 연결된 노트북을 생성합니다.

AWSGlueServiceNotebookRole, AmazonS3FullAccess과 같이 필요한 권한을 가진 role을 연결합니다.

rds에 아래의 과정을 거쳐 연결합니다.

mysql connecter를 다운받은 후 jar파일을 s3에 업로드 합니다.
아래의 코드를 사용하여 연결합니다.

# report db와 연결하기 위한 연결 정보

connection_mysql8_options_source_mapping = {
    "url": "jdbc:mysql://data.ap-northeast-2.rds.amazonaws.com:3306/mapping",
    "dbtable": "tb_col",
    "user": "hello",
    "password": "1234",
    "customJdbcDriverS3Path": "s3://s3-data/service=market/mysql-connector-java-8.0.26.jar",
    "customJdbcDriverClassName": "com.mysql.cj.jdbc.Driver"}
    
glueContext=GlueContext(sc)

df_mapping = glueContext.create_dynamic_frame.from_options(connection_type="mysql", connection_options=connection_mysql8_options_source_mapping)

참고 : https://aws.amazon.com/ko/blogs/big-data/building-aws-glue-spark-etl-jobs-by-bringing-your-own-jdbc-drivers-for-amazon-rds/

s3에서 csv파일 불러오기

S3_IN = "s3://s3주소/파일명.csv"

start = datetime.datetime.now()

csv_df = (
    spark.read.format("org.apache.spark.csv")
    .option("header", True)
    .option("quote", '"')
    .option("escape", '"')
    .option("inferSchema", True)
    .option("ignoreLeadingWhiteSpace", True)
    .option("ignoreTrailingWhiteSpace", True)
    .csv(S3_IN, multiLine=False)
)

end = datetime.datetime.now()

print("##########total_time : " + str(end-start))

s3에 parquet파일로 저장

# s3에 업로드 하기 위해 spark dataframe을 glue dynamicframe으로 변환
dynamic_csv_df = DynamicFrame.fromDF(csv_df, glueContext, "dynamic_csv_df")

S3_location = "s3://s3-data-dev/service=market"

datasink = glueContext.write_dynamic_frame_from_options(
               frame=dynamic_csv_df,
               connection_type="s3",
               connection_options={
                  "path": S3_location,
                  "partitionKeys": ["Date_"]
               },
               format="glueparquet",
               format_options = {"compression": "snappy"},
               transformation_ctx ="datasink")

분석 방법관련 추가사항은 아래를 참고하실 수 있습니다.