Apache Spark용 AWS Glue 4.0 자세히 살펴보기

2023년 7월: 이 글은 AWS Glue Studio 노트북 및 대화형 세션에서의 Glue 4.0 지원에 대해 검토 및 업데이트 되었습니다.

이 글은 AWS Big Data Blog에 게시된 ‘ Dive deep into AWS Glue 4.0 for Apache Spark by Gonzalo Herreros, Rajendra Gujja, Bo Li, Mohit Saxena, Noritaka Sekiyama, and Savio Dsouza’을 한국어 번역 및 편집하였습니다.

데이터로부터 인사이트를 도출하는 것은 어렵습니다. 조직의 여러 데이터 저장소 간의 데이터 접근을 방해하는 사일로를 다루는 경우, 이는 더욱 어려워 집니다. 현대적인 데이터 아키텍처에서 데이터 사일로를 극복하기 위해서는 원활한 데이터 통합이 중요합니다. AWS Glue는 데이터 준비를 더 간단하고, 빠르고, 저렴하게 만드는 서버리스 데이터 통합 서비스입니다. 70 개 이상의 다양한 데이터 소스를 탐색 및 연결하고, 중앙 집중식 데이터 카탈로그에서 데이터를 관리하고, 데이터 레이크 및 데이터 웨어하우스에 데이터를 로드하기 위해 데이터 통합 파이프라인을 생성하고 실행하며 모니터링 할 수 있습니다. Apache Spark용 AWS Glue는 Apache Spark의 강력한 엔진을 활용하여 대량의 데이터 통합 작업을 대규모로 처리할 수 있습니다.

AWS Glue는 AWS re:Invent 2022에서 최적화된 Apache Spark 3.3.0 런타임, Python 3.10 지원, 새롭게 향상된 Amazon Redshift 커넥터와 같은 많은 업그레이드가 포함된 4.0 버전이 출시되었습니다.

이 게시글에서는 AWS Glue 새 버전이 제공하는 주요 이점들과 더 나은 데이터 통합 파이프라인을 만드는데 어떻게 도움이 될 수 있는지에 대해 다루도록 하겠습니다.

Spark 업그레이드 주요 사항

AWS Glue 4.0의 Spark 신규 버전은 여기에서 강조하는 여러 가지 유용한 기능들을 제공합니다. 더 자세한 내용은, Spark 3.3.0 출시 정보와 Spark 3.2.0 출시 정보를 참조하시기 바랍니다.

pandas API 지원

pandas API를 지원하므로, 인기있는 Python 라이브러리인 pandas에 익숙한 분들은 Spark 프레임워크 API를 배우지 않고도, pandas API를 사용해 Spark의 분산 추출, 변환, 로드(ETL) 작업을 작성할 수 있습니다. 이에 대해서는 뒤에서 더 자세히 설명하겠습니다.

Python UDF 프로파일링

Python UDF 프로파일링 기능을 사용해 일반적인 사용자 정의 함수(UDF) 및 pandas UDF를 프로파일링 할 수 있습니다. SparkContext에서 show_profiles()를 호출하면 Python 실행에 대한 상세 정보를 얻을 수 있습니다. 이 기능은 기존 Spark 1의 RDD에만 지원이 되었으나, 이제 DataFrame Python UDF에도 작동하며, UDF 호출 수와 UDF 소요 시간에 대한 유용한 정보를 제공합니다.
다음 예시는 프로파일러가 시간을 측정하는 방법을 나타냅니다. 예시의 스크린샷에서 백만 개 이상의 행을 처리하고 (하지만 pandas UDF는 112번의 호출만 필요합니다) 1초 동안 sleep을 수행하는 pandas UDF의 프로파일 정보를 볼 수 있습니다. 스크린샷에 보이는 것처럼spark.sparkContext.show_profiles()명령을 사용할 수 있습니다.

푸쉬다운 필터

푸쉬다운 필터는 집계나 제한(limit)과 같은 많은 시나리오에서 사용됩니다. 이 업그레이드에서는 Bloom 필터와 skew 최적화도 지원합니다. 이러한 개선 사항을 통해 데이터를 더 적게 읽고 더 효율적으로 처리함으로써 더 큰 데이터셋을 처리할 수 있습니다. 자세한 내용은 Spark 3.3.0 출시 정보를 참조하세요.

ANSI SQL

이제 Spark SQL에서 ANSI SQL 표준 사양을 준수하도록 설정할 수 있습니다. 이를 통해 사용자들이 기존의 SQL 기술을 활용하여 빠르게 AWS Glue에서 가치를 창출할 수 있습니다. 자세한 내용은 ANSI Compliance를 참조하세요.

Adaptive query execution 기본 사용

Adaptive query execution(AQE)이 기본적으로 사용되어 Spark SQL 성능을 최적화하는 데 도움을 줍니다. spark.sql.adaptive.enabled 설정을 사용해서 AQE를 켜고 끌 수가 있습니다. AWS Glue 4.0부터는 기본적으로 활성화되어 있으므로 더이상 명시적으로 활성화할 필요가 없습니다.

에러 메시지 개선

개선된 오류 메시지는 더 나은 컨텍스트와 쉬운 해결 방법을 제공합니다. 예를 들어, ANSI 모드의 SQL문에서 0으로 나누기를 하는 경우에, 이전에는 단지 Java exception: java.lang.ArithmeticException: divide by zero만 나왔습니다. 복잡성과 쿼리 수에 따라 원인이 명확하지 않을 수 있으며, 식별될  때까지 시행착오를 거치는 재실행이 필요할 수도 있습니다.

새 버전에서는, 훨씬 더 명확한 오류가 표시됩니다.

Caused by: org.apache.spark.SparkArithmeticException: Division by zero. 
Use `try_divide` to tolerate divisor being 0 and return NULL instead. 
If necessary set "spark.sql.ansi.enabled" to "false" (except for ANSI interval type) to bypass this error.
== SQL(line 1, position 8) ==
select sum(cost)/(select count(1) from items where cost > 100) from items
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

문제를 일으킨 쿼리를 보여줄 뿐만 아니라, 오류가 발생한 특정 작업(이 경우에는 나누기)도 표시해 줍니다. 추가적으로 문제 해결에 대한 지침도 제공합니다.

Spark의 새로운 pandas API

Spark 업그레이드에는 새롭고 흥미로운 기능이 추가되었는데, 기존 Python pandas 프레임워크 지식을 확장 가능한 분산 런타임에서 사용할 수 있게 한 것입니다. 이를 통해 이전에 Spark 경험이 없는 팀도 진입 장벽이 낮아져 신속하게 가치를 제공하고, Apache Spark용 AWS Glue 런타임을 최대한 활용할 수 있습니다.

새로운 API는 pandas DataFrame 호환 API를 제공하므로, 100% 호환되지는 않지만 기존의 pandas 코드를 사용하고 import 문을 변경해서 Spark용 AWS Glue로 마이그레이션할 수 있습니다.

기존의 pandas코드를 Spark 에서 실행하기를 원하다면, import문을 바꿔서 테스트할 수 있습니다:

# Replace pure pandas 
import pandas as pd
# With pandas API on Spark
import pyspark.pandas as pd

어떤 경우에는 Spark용 pandas API와 Python pandas를 같이 사용하기를 원할 수 있습니다. Spark용 pandas API에서 사용할 수 없는 기능이 필요하거나, 데이터가 너무 작아 분산 시키는 것보다 로컬에서 수행하는 것이 더 효율적인 경우입니다. 그런 경우에는 혼동을 피하기 위해서 pandas와 Spark 용 pandas의 import에 서로 다른 별칭을 사용하고, 성능과 기능에 영향을 미치는 다른 타입의 DataFame에 대해 다른 명명 규칙을 따르게 하는 것이 좋습니다. 예를 들어, pandas DataFrame 변수는 pdf_, Spark용 pandas는 psdf_, 표준 Spark는 sdf_ 또는 df_를 사용합니다.

to_spark() 호출해서 표준 Spark DataFrame으로 변환할 수도 있습니다. 이렇게 하면 카탈로그 테이블에 직접 쓰거나 Spark 커넥터를 사용하는 등 pandas에서 사용할 수 없는 기능도 사용할 수 있습니다.

다음 코드는 다양한 타입의 API들을 결합한 예를 보여줍니다.

# The job has the parameter "--additional-python-modules":"openpyxl", 
#  the Excel library is not provided by default

import pandas as pd
# pdf is a pure pandas DF which resides in the driver memory only
pdf = pd.read_excel('s3://mybucket/mypath/MyExcel.xlsx', index_col=0)

import pyspark.pandas as ps
# psdf behaves like a pandas df but operations on it will be distributed among nodes
psdf = ps.from_pandas(pdf)
means_series = psdf.mean()

# Convert to a dataframe of column names and means
#  pandas on Spark series don't allow iteration so we convert to pure pandas
#  on a big dataset this could cause the driver to be overwhelmed but not here
# We reset to have the index labels as a columns to use them later
pdf_means = pd.DataFrame(means_series.to_pandas()).reset_index(level=0)
# Set meaningful column names
pdf_means.columns = ["excel_column", "excel_average"]

# We want to use this to enrich a Spark DF representing a huge table
#  convert to standard Spark DF so we can use the start API
sdf_means = ps.from_pandas(pdf_means).to_spark()

sdf_bigtable = spark.table("somecatalog.sometable")
sdf_bigtable.join(sdf_means, sdf_bigtable.category == sdf_means.excel_column)

개선된 Amazon Redshift 커넥터

새로운 버전의 Amazon Redshift 커넥터는 많은 개선 사항을 제공합니다.

• 푸쉬다운 최적화
• SUPER 타입 컬럼 읽기 지원
• 위치 대신 열 이름 기준으로 쓰기 허용
• Amazon Redshift에서 읽기 성능 향상을 위한 최적화된 직렬화
• 사전 및 사후 작업 트리밍, 숫자로 된 표준 시간대 형식 처리와 같은 기타 작은 개선사항

새로운 Amazon Redshift 커넥터는 기존 오픈소스 커넥터 프로젝트를 기반으로 만들어졌으며 성능 및 보안에 대한 추가 개선 사항을 제공하여 최대 10배 빠른 애플리케이션 성능을 얻을 수 있습니다. Amazon Redshift에서 데이터를 읽는 AWS Glue 작업을 가속화하고, 데이터 집약적인 워크로드를 보다 안정적으로 실행할 수 있습니다. 자세한 내용은 Amazon Redshift 간 데이터 이동을 참조하세요. 사용 방법에 대한 자세한 내용은 New – Amazon Redshift Integration with Apache Spark을 참조하세요.

AWS Glue DynamicFrame에서 새로운 Amazon Redshift 커넥터를  사용하는 경우, 기존 메소드인 GlueContext.create_data_frame 과 GlueContext.write_data_frame를 사용합니다.

Spark DataFrame에서 새로운 Amazon Redshift 커넥터를 사용하는 경우는, 다음 코드 조각과 같이 io.github.spark_redshift_community.spark.redshift형식을 사용합니다.

df = spark.read.format("io.github.spark_redshift_community.spark.redshift") \
    .option("url", url) \
    .option("dbtable", dbtable) \
    .option("tempdir", redshiftTmpDir) \
    .option("user", user) \
    .option("password", password) \
    .option("aws_iam_role", aws_iam_role) \
    .load()

기타 업그레이드 및 개선사항

다음은 종속 라이브러리의 업데이트 및 개선 사항입니다.

• Spark 3.3.0-amzn-1
• Hadoop 3.3.0-amzn-0
• Hive 2.3.9-amzn-2
• Parquet 1.12
• Log4j 2
• Python 3.10
• Arrow 7.0.0
• Boto3 1.26
• EMRFS 2.54.0
• AWS Glue Data Catalog client 3.7.0
• 제공되는 JDBC 드라이버의 새 버전
  • MySQL 8.0.23
  • PostgreSQL 42.3.6
  • Microsoft SQL Server 9.4.0
  • Oracle 21.7
  • MongoDB 4.7.2
  • Amazon Redshift redshift-jdbc42-2.1.0.9
• 인기있는 테이블 포맷용 플러그인 통합 및 업그레이드
  • Iceberg 1.0.0
  • Hudi 0.12.1
  • Delta Lake 2.1.0

자세한 내용은 AWS Glue 버전 4.0으로 AWS Glue 작업 마이그레이션의 부록을 참조하세요.

향상된 성능

새로운 기능들 외에도 AWS Glue 4.0은 더 낮은 비용으로 성능 향상을 제공합니다. 요약하면, Amazon Simple Storage Service(Amazon S3)를 사용하는 AWS Glue 4.0은 AWS Glue 3.0보다 2.7배 빠르고, Amazon Redshift를 사용하는 AWS Glue 4.0은 AWS Glue 3.0보다 7.1배 빠릅니다. 다음 섹션에서는 Amazon S3 및 Amazon Redshift를 사용한 AWS Glue 4.0의 성능 결과에 대해 자세히 설명합니다.

Amazon S3

다음 차트는 AWS Glue 3.0 과 AWS Glue 4.0 간 3TB 쿼리 데이터 셋에서의 모든 쿼리의 총 작업 실행시간(초)을 보여줍니다. TPC-DS 데이터셋은 Parquet 형식으로 S3 버킷에 있으며, 30개의 G.2X 워커를 사용했습니다. Amazon S3에서 TPC-DS 테스트 결과, AWS Glue 4.0의 총 작업 실행시간이 AWS Glue 3.0보다 2.7배 더 빨랐습니다. 자세한 절차들은 이 게시글 부록에 설명되어 있습니다.

Amazon Redshift

다음 차트는 AWS Glue 3.0 과 AWS Glue 4.0 간 1TB 쿼리 데이터 셋에서의 모든 쿼리의 총 작업 실행시간(초)을 보여줍니다. TPC-DS 데이터셋은 ra3.4xlarge 타입 5개 노드의  Amazon Redshift 클러스터에 있으며, 150개의 G.2X 워커를 사용했습니다. Amazon Redshift에서 TPC-DS 테스트 결과, AWS Glue 4.0의 총 작업 실행시간이 AWS Glue 3.0보다 7.1배 더 빨랐습니다.

AWS Glue 4.0 시작하기

AWS Glue 4.0은 AWS Glue Studio, AWS Glue 콘솔, 최신 AWS SDK, AWS Command Line Interface(AWS CLI)를 통해 시작할 수 있습니다.

AWS Glue Studio에서 AWS Glue 4.0을 사용하려면, AWS Glue 작업을 열고 Job details 탭에서 버전 Glue 4.0 – Supports spark 3.3, Scala 2, Python 3을 선택합니다.

AWS Glue Studio 노트북 또는 대화형 세션에서 AWS Glue 4.0을 사용하려면, %glue_version 매직 값을 4.0으로 설정해 주세요.

기존 AWS Glue 0.9, 1.0, 2.0, 3.0에서 AWS Glue 4.0으로 마이그레이션하려면, AWS Glue 버전 4.0으로 AWS Glue 작업 마이그레이션을 참조하세요.

AWS Glue 4.0 Docker 이미지를 이제 Docker Hub에서 사용 가능하므로, 새 버전을 로컬에서 개발하는 데 사용할 수 있습니다. 자세한 내용은 Develop and test AWS Glue version 3.0 and 4.0 jobs locally using a Docker container를 참조하세요.

결론

이번 게시글에서는 AWS Glue의 새로운 4.0 버전에서 제공하는 주요 업그레이드에 대해 설명했습니다. 이미 해당 버전으로 새 작업을 작성하고 모든 개선 사항을 활용할 수 있을 뿐만 아니라, 기존 작업들을 마이그레이션할 수도 있습니다.

부록 : Amazon S3 데이터셋에 대한 AWS Glue의 TPC-DS 벤치마크

S3 버킷의 데이터셋에 대한 AWS Glue의 TPC-DS 벤치마크를 수행하려면, TPC-DS 데이터셋을 여러분의 S3 버킷에 복사해야 합니다. 이 절차들은 emr-spark-benchmark를 기반으로 합니다.

1. 필요한 경우 테스트 계정에 새 S3 버킷을 생성합니다. 다음 코드에서 $YOUR_S3_BUCKET 는 여러분의 S3 버킷으로 바꾸십시오. 환경변수로 YOUR_S3_BUCKET를 지정하는 것이 좋습니다.

export YOUR_S3_BUCKET=<Your bucket name>
aws s3 mb s3://$YOUR_S3_BUCKET

2. TPC-DS 소스 데이터를 S3 버킷에 입력으로 복사합니다. 환경 변수로 내보내지 않았다면, $YOUR_S3_BUCKET을 여러분의 S3 버킷 이름으로 바꿉니다.

aws s3 sync s3://blogpost-sparkoneks-us-east-1/blog/BLOG_TPCDS-TEST-3T-partitioned/ s3://$YOUR_S3_BUCKET/blog/BLOG_TPCDS-TEST-3T-partitioned/

3. Steps to build spark-benchmark-assembly application에 있는 지침을 따라서 벤치마크 애플리케이션을 빌드합니다.
   편의를 위해서 AWS Glue 4.0용으로 빌드한 샘플 애플리케이션 JAR파일  spark-benchmark-assembly-3.3.0.jar을 제공합니다.
4. spark-benchmark-assembly JAR파일을 S3 버킷에 업로드 합니다.
5. AWS Glue Studio에서 script editor를 통해 새 AWS Glue 작업을 생성합니다.
   1. Job details의 Type에서 Spark을 선택합니다.
   2. Glue version에서 Glue 4.0 – Supports spark 3.3, Scala 2, Python 3을 선택합니다.
   3. Language에서 Scala를 선택합니다.
   4. Worker type에서 원하는 워커 타입을 선택합니다.
   5. Requested number of workers에서 원하는 숫자를 선택합니다.
   6. Advanced properties의 Dependent JARs path에 spark-benchmark-assembly JAR 파일의 S3 경로를 입력합니다.
   7. Script에 다음 코드 조각을 입력합니다.

      import com.amazonaws.eks.tpcds.BenchmarkSQL
      
      object GlueApp {
        def main(sysArgs: Array[String]): Unit = {
          BenchmarkSQL.main(
            Array(
              "s3://YOUR_S3_BUCKET/blog/BLOG_TPCDS-TEST-3T-partitioned",
              "s3://YOUR_S3_BUCKET/blog/GLUE_TPCDS-TEST-3T-RESULT/", 
              "/opt/tpcds-kit/tools", 
              "parquet", 
              "3000", 
              "3", 
              "false", 
              "q1-v2.4,q10-v2.4,q11-v2.4,q12-v2.4,q13-v2.4,q14a-v2.4,q14b-v2.4,q15-v2.4,q16-v2.4,q17-v2.4,q18-v2.4,q19-v2.4,q2-v2.4,q20-v2.4,q21-v2.4,q22-v2.4,q23a-v2.4,q23b-v2.4,q24a-v2.4,q24b-v2.4,q25-v2.4,q26-v2.4,q27-v2.4,q28-v2.4,q29-v2.4,q3-v2.4,q30-v2.4,q31-v2.4,q32-v2.4,q33-v2.4,q34-v2.4,q35-v2.4,q36-v2.4,q37-v2.4,q38-v2.4,q39a-v2.4,q39b-v2.4,q4-v2.4,q40-v2.4,q41-v2.4,q42-v2.4,q43-v2.4,q44-v2.4,q45-v2.4,q46-v2.4,q47-v2.4,q48-v2.4,q49-v2.4,q5-v2.4,q50-v2.4,q51-v2.4,q52-v2.4,q53-v2.4,q54-v2.4,q55-v2.4,q56-v2.4,q57-v2.4,q58-v2.4,q59-v2.4,q6-v2.4,q60-v2.4,q61-v2.4,q62-v2.4,q63-v2.4,q64-v2.4,q65-v2.4,q66-v2.4,q67-v2.4,q68-v2.4,q69-v2.4,q7-v2.4,q70-v2.4,q71-v2.4,q72-v2.4,q73-v2.4,q74-v2.4,q75-v2.4,q76-v2.4,q77-v2.4,q78-v2.4,q79-v2.4,q8-v2.4,q80-v2.4,q81-v2.4,q82-v2.4,q83-v2.4,q84-v2.4,q85-v2.4,q86-v2.4,q87-v2.4,q88-v2.4,q89-v2.4,q9-v2.4,q90-v2.4,q91-v2.4,q92-v2.4,q93-v2.4,q94-v2.4,q95-v2.4,q96-v2.4,q97-v2.4,q98-v2.4,q99-v2.4,ss_max-v2.4", 
              "true"
            )
          )
        }

6. 저장하고 작업을 실행합니다.

결과 파일은 s3://YOUR_S3_BUCKET/blog/GLUE_TPCDS-TEST-3T-RESULT/에 저장됩니다.