AWS glue에서 spark와 notebook 활용하기

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

1. 엔드포인트 추가

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

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

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

3. 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/

4. 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))

5. 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")

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

• 링크 : https://www.analyticsvidhya.com/blog/2020/08/essential-functionalities-to-guide-you-while-using-aws-glue-and-pyspark/