데이터를 Apache Parquet으로 변환하기 위한 세 가지 AWS Glue ETL 작업 유형

작성자: Adnan Alvee(AWS), Karthikeyan Ramachandran 및 Nith Govindasivan(AWS)

요약

Amazon Web Services(AWS) 클라우드에서 AWS Glue는 완전관리형 추출, 전환, 적재(ETL) 서비스입니다. AWS Glue를 사용하면 경제적으로 데이터를 분류, 정리, 보강하고, 다양한 데이터 스토어와 데이터 스트림 간에 안정적으로 이동할 수 있습니다.

이 패턴은 AWS Glue에서 다양한 작업 유형을 제공하며 세 가지 스크립트를 사용하여 ETL 작업 작성을 보여줍니다.

AWS Glue를 사용하여 Python 쉘 환경에서 ETL 작업을 작성할 수 있습니다. 관리형 Apache Spark 환경에서 Python (PySpark) 또는 Scala를 사용하여 일괄 처리 및 스트리밍 ETL 작업을 모두 생성할 수도 있습니다. ETL 작업 작성을 시작하기 위해 이 패턴은 Python, 셸 PySpark, Scala를 사용하는 일괄 ETL 작업에 중점을 둡니다. Python 쉘 작업은 컴퓨팅 파워가 덜 필요한 워크로드를 위한 것입니다. 관리형 Apache Spark 환경은 높은 컴퓨팅 파워가 필요한 워크로드에 적합합니다.

Apache Parquet은 효율적인 압축 및 인코딩 체계를 지원하도록 구축되었습니다. 데이터를 컬럼 방식으로 저장하므로 분석 워크로드의 속도를 높일 수 있습니다. 데이터를 Parquet으로 변환하면 장기적으로 스토리지 공간, 비용 및 시간을 절감할 수 있습니다. Parquet에 대해 자세히 알아보려면 블로그 게시물 Apache Parquet: 오픈 소스 컬럼형 데이터 형식으로 영웅이 되는 방법을 참조하십시오.

사전 조건 및 제한 사항

사전 조건 

아키텍처

대상 기술 스택  

자동화 및 규모 조정

도구

AWS 서비스

기타 도구

구성

다음 설정을 사용하여 AWS Glue ETL의 컴퓨팅 파워를 구성합니다. 비용을 줄이려면 이 패턴으로 제공되는 워크로드를 실행할 때 최소 설정을 사용합니다. 

다음 표에는 Apache Spark 환경의 다양한 AWS Glue 작업자 유형이 나와 있습니다. Python 쉘 작업은 Apache Spark 환경을 사용하여 Python을 실행하지 않으므로 테이블에 포함되지 않습니다.

코드

IAM 역할 및 파라미터 구성을 포함하여 이 패턴에 사용되는 코드는 추가 정보 섹션을 참조하십시오.

에픽

관련 리소스

참조

튜토리얼 및 동영상 

추가 정보

IAM 역할

AWS Glue 작업을 생성할 때 다음 코드 스니펫에 표시된 권한이 있는 기존 IAM 역할을 사용하거나 새 역할을 사용할 수 있습니다.

새 역할을 생성하려면 다음 YAML 코드를 사용합니다.

# (c) 2022 Amazon Web Services, Inc. or its affiliates. All Rights Reserved. This AWS Content is provided subject to the terms of the AWS Customer
# Agreement available at https://aws.amazon.com/agreement/ or other written agreement between Customer and Amazon Web Services, Inc.

AWSTemplateFormatVersion: "2010-09-09"

Description: This template will setup IAM role for AWS Glue service.

Resources:
  rGlueRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service:
                - "glue.amazonaws.com"
            Action:
              - "sts:AssumeRole"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
      Policies:
        - PolicyName: !Sub "${AWS::StackName}-s3-limited-read-write-inline-policy"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - "s3:PutObject"
                  - "s3:GetObject"
                Resource: "arn:aws:s3:::*/*"
      Tags:
        - Key  : "Name"
          Value : !Sub "${AWS::StackName}"

Outputs:
  oGlueRoleName:
    Description: AWS Glue IAM role
    Value:
      Ref: rGlueRole
    Export:
      Name: !Join [ ":", [ !Ref "AWS::StackName", rGlueRole ] ]

AWS Glue(Python 쉘)

Python 코드는 Pandas와 PyArrow 라이브러리를 사용하여 데이터를 Parquet으로 변환합니다. Pandas 라이브러리는 이미 사용할 수 있습니다. PyArrow 라이브러리는 한 번만 실행되므로 패턴을 실행하면 다운로드됩니다. 휠 파일을 사용하여 라이브러리로 PyArrow 변환하고 파일을 라이브러리 패키지로 제공할 수 있습니다. 휠 파일 패키징에 대한 자세한 내용은 자체 Python 라이브러리 제공을 참조하십시오.

AWS Glue Python 쉘 파라미터

from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv, ["input_loc", "output_loc"])

AWS Glue Python 쉘 코드

from io import BytesIO
import pandas as pd
import boto3
import os
import io
import site
from importlib import reload
from setuptools.command import easy_install
install_path = os.environ['GLUE_INSTALLATION']
easy_install.main( ["--install-dir", install_path, "pyarrow"] )
reload(site)
import pyarrow


input_loc = "bucket-name/prefix/sample_data.csv"
output_loc = "bucket-name/prefix/"


input_bucket = input_loc.split('/', 1)[0]
object_key = input_loc.split('/', 1)[1]


output_loc_bucket = output_loc.split('/', 1)[0]
output_loc_prefix = output_loc.split('/', 1)[1] 


s3 = boto3.client('s3')
obj = s3.get_object(Bucket=input_bucket, Key=object_key)
df = pd.read_csv(io.BytesIO(obj['Body'].read()))


parquet_buffer = BytesIO()
s3_resource = boto3.resource('s3')
df.to_parquet(parquet_buffer, index=False) 
s3_resource.Object(output_loc_bucket, output_loc_prefix +  'data' + '.parquet').put(Body=parquet_buffer.getvalue()) 

Python을 사용한 AWS Glue Spark 작업

Python에서 AWS Glue Spark 작업 유형을 사용하려면 작업 유형으로 Spark를 선택합니다. 작업 스타트업 시간이 개선된 Spark 3.1, Python 3(Glue 버전 3.0)을 AWS Glue 버전으로 선택합니다.

AWS Glue Python 파라미터

from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv, ["JOB_NAME", "input_loc", "output_loc"])

Python 코드를 사용한 AWS Glue Spark 작업

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.transforms import *
from awsglue.dynamicframe import DynamicFrame
from awsglue.utils import getResolvedOptions
from awsglue.job import Job


sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

input_loc = "bucket-name/prefix/sample_data.csv"
output_loc = "bucket-name/prefix/"

inputDyf = glueContext.create_dynamic_frame_from_options(\
    connection_type = "s3", \
    connection_options = { 
        "paths": [input_loc]}, \
    format = "csv",
    format_options={
        "withHeader": True,
        "separator": ","
    })


outputDF = glueContext.write_dynamic_frame.from_options(\
    frame = inputDyf, \
    connection_type = "s3", \
    connection_options = {"path": output_loc \
        }, format = "parquet")    

압축된 대용량 파일이 많은 경우(예: 각각 약 3MB인 파일 1,000개), 다음 코드와 같이 recurse 파라미터와 함께 compressionType 파라미터를 사용하여 접두사 내에서 사용 가능한 모든 파일을 읽습니다.

input_loc = "bucket-name/prefix/"
output_loc = "bucket-name/prefix/"

inputDyf = glueContext.create_dynamic_frame_from_options(
                    connection_type = "s3", 
                    connection_options = {"paths": [input_loc], 
                                            "compressionType":"gzip","recurse" :"True",
                                            },
                    format = "csv",
                    format_options={"withHeader": True,"separator": ","}
                    )

압축된 작은 파일(예: 각각 약 133KB인 파일 1,000개) 이 많은 경우 groupFiles 파라미터를 compressionType 및 recurse 파라미터와 함께 사용하십시오. groupFiles 파라미터는 작은 파일을 여러 개의 큰 파일로 그룹화하고 groupSize 파라미터는 지정된 크기(예: 1MB)로의 그룹화를 제어합니다. 다음 코드 스니펫은 코드 내에서 이러한 파라미터를 사용하는 예를 제공합니다.

input_loc = "bucket-name/prefix/"
output_loc = "bucket-name/prefix/"

inputDyf = glueContext.create_dynamic_frame_from_options(
                    connection_type = "s3", 
                    connection_options = {"paths": [input_loc], 
                                            "compressionType":"gzip","recurse" :"True",
                                             "groupFiles" :"inPartition",  "groupSize" :"1048576",
                                            },
                    format = "csv",
                    format_options={"withHeader": True,"separator": ","}
                    )

워커 노드를 변경하지 않고도 AWS Glue 작업에서 여러 파일(크든 작든, 압축이 있든 없든)을 읽고 Parquet 형식으로 대상에 쓸 수 있습니다.

Scala를 사용한 AWS Glue Spark 작업

Scala에서 AWS Glue Spark 작업 유형을 사용하려면 작업 유형으로 Spark를 선택하고 Scala로 언어를 선택합니다. 작업 스타트업 시간이 개선된 Spark 3.1, Scala 2(Glue 버전 3.0)를 AWS Glue 버전으로 선택합니다. 스토리지 스페이스를 절약하기 위해 다음 AWS Glue with Scala 샘플도 이 applyMapping 기능을 사용하여 데이터 유형을 변환합니다.

AWS Glue Scala 파라미터

import com.amazonaws.services.glue.util.GlueArgParser val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME", "inputLoc", "outputLoc").toArray)

Scala 코드를 사용한 AWS Glue Spark 작업

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.MappingSpec
import com.amazonaws.services.glue.DynamicFrame
import com.amazonaws.services.glue.errors.CallSite
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._


object GlueScalaApp {
  def main(sysArgs: Array[String]) {
    
    @transient val spark: SparkContext = SparkContext.getOrCreate()
    val glueContext: GlueContext = new GlueContext(spark)

    val inputLoc = "s3://bucket-name/prefix/sample_data.csv"
    val outputLoc = "s3://bucket-name/prefix/"

    val readCSV = glueContext.getSource("csv", JsonOptions(Map("paths" -> Set(inputLoc)))).getDynamicFrame()

    val applyMapping = readCSV.applyMapping(mappings = Seq(("_c0", "string", "date", "string"), ("_c1", "string", "sales", "long"),
    ("_c2", "string", "profit", "double")), caseSensitive = false)

    val formatPartition = applyMapping.toDF().coalesce(1)

    val dynamicFrame = DynamicFrame(formatPartition, glueContext)

    val dataSink = glueContext.getSinkWithFormat(
        connectionType = "s3", 
        options = JsonOptions(Map("path" -> outputLoc )),
        transformationContext = "dataSink", format = "parquet").writeDynamicFrame(dynamicFrame)
  }
}

첨부

이 문서와 관련된 추가 콘텐츠에 액세스하려면 attachment.zip 파일의 압축을 풉니다.