기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다. # 데이터를 Apache Parquet으로 변환하기 위한 세 가지 AWS Glue ETL 작업 유형 *Adnan Alvee, Nith Govindasivan, Karthikeyan Ramachandran, Amazon Web Services* ## 요약 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: 오픈 소스 컬럼형 데이터 형식으로 영웅이 되는 방법](https://blog.openbridge.com/how-to-be-a-hero-with-powerful-parquet-google-and-amazon-f2ae0f35ee04)을 참조하십시오. ## 사전 조건 및 제한 사항 **사전 조건 ** + AWS Identity and Access Management(IAM) 역할(역할이 없는 경우 [추가 정보](#three-aws-glue-etl-job-types-for-converting-data-to-apache-parquet-additional) 섹션 참조) ## 아키텍처 **대상 기술 스택  ** + Glue + Amazon Simple Storage Service(Amazon S3) + Apache Parquet **자동화 및 규모 조정** + [AWS Glue 워크플로](https://docs.aws.amazon.com/glue/latest/dg/workflows_overview.html)는 ETL 파이프라인의 완전 자동화를 지원합니다. + 데이터 처리 단위(DPU)의 수 또는 작업자 유형을 변경하여 수평 및 수직으로 규모를 조정할 수 있습니다. ## 도구 **AWS 서비스** + [Amazon Simple Storage Service(S3)](https://docs.aws.amazon.com/AmazonS3/latest/userguide/Welcome.html)는 원하는 양의 데이터를 저장, 보호 및 검색하는 데 도움이 되는 클라우드 기반 객체 스토리지 서비스입니다. + [AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/what-is-glue.html)는 다양한 데이터 스토어와 데이터 스트림 간에 데이터를 분류, 정리, 보강하고 이동할 수 있는 완전관리형 ETL 서비스입니다. **기타 도구** + [Apache Parquet](https://parquet.apache.org/)는 스토리지 및 검색을 위해 설계된 오픈 소스 열 지향 데이터 파일 형식입니다. **구성** 다음 설정을 사용하여 AWS Glue ETL의 컴퓨팅 파워를 구성합니다. 비용을 줄이려면 이 패턴으로 제공되는 워크로드를 실행할 때 최소 설정을 사용합니다.  + **Python 쉘** – 1DPU를 사용하여 16GB의 메모리를 활용하거나 0.0625DPU를 사용하여 1GB의 메모리를 활용할 수 있습니다. 이 패턴은 AWS Glue 콘솔의 기본값인 0.0625 DPU를 사용합니다. + **Python 또는 Scala for Spark** – 콘솔에서 Spark 관련 작업 유형을 선택하면 AWS Glue는 기본적으로 10개의 작업자와 G.1X 작업자 유형을 사용합니다. 이 패턴은 허용된 최소 수인 두 개의 작업자를 사용하며, 표준 작업자 유형은 충분하고 비용 효율적입니다. 다음 표에는 Apache Spark 환경의 다양한 AWS Glue 작업자 유형이 나와 있습니다. Python 쉘 작업은 Apache Spark 환경을 사용하여 Python을 실행하지 않으므로 테이블에 포함되지 않습니다. | | | | 표준 | G.1X | G.2X | | --- |--- |--- |--- | | vCPU | 4 | 4 | 8 | | Memory | 16 GB | 16 GB | 32GB | | 디스크 공간 | 50GB | 64GB | 128GB | | 작업자당 실행기 | 2 | 1  | 1 | **코드** IAM 역할 및 파라미터 구성을 포함하여 이 패턴에 사용되는 코드는 [추가 정보](#three-aws-glue-etl-job-types-for-converting-data-to-apache-parquet-additional) 섹션을 참조하십시오. ## 에픽 ### 데이터 업로드 | 작업 | 설명 | 필요한 기술 | | --- | --- | --- | | 새 S3 버킷이나 기존 S3 버킷에 데이터를 업로드합니다. | 계정에서 기존 S3 버킷을 생성하거나 사용합니다. [첨부 파일](#attachments-8c926709-8fa4-417f-9aaf-bcc8113d018f) 섹션에서 sample\_data.csv 파일을 업로드하고 S3 버킷과 접두사 위치를 기록해 둡니다. | 일반 AWS | ### AWS Glue 작업 생성 및 실행 | 작업 | 설명 | 필요한 기술 | | --- | --- | --- | | AWS Glue 작업을 생성합니다. | AWS Glue 콘솔의 ETL 섹션에서 AWS Glue 작업을 추가합니다. 적절한 작업 유형, AWS Glue 버전, 해당 DPU/작업자 유형 및 작업자 수를 선택합니다. 자세한 내용은 *구성* 섹션을 참조하십시오. | 개발자, 클라우드 또는 데이터 | | 입력 및 출력 위치를 변경합니다. | AWS Glue 작업에 해당하는 코드를 복사하고 **데이터 업로드** 에픽에서 기록해 둔 입력 및 출력 위치를 변경합니다. | 개발자, 클라우드 또는 데이터 | | 파라미터를 구성합니다. | [추가 정보](#three-aws-glue-etl-job-types-for-converting-data-to-apache-parquet-additional) 섹션에 제공된 스니펫을 사용하여 ETL 작업의 파라미터를 설정할 수 있습니다. AWS Glue는 내부적으로 4개의 인수 이름을 사용합니다.[See the AWS documentation website for more details](http://docs.aws.amazon.com/ko_kr/prescriptive-guidance/latest/patterns/three-aws-glue-etl-job-types-for-converting-data-to-apache-parquet.html)
`--JOB_NAME` 파라미터는 AWS Glue 콘솔에 명시적으로 입력해야 합니다. **작업**, **작업 편집**, **보안 구성, 스크립트 라이브러리 및 작업 파라미터(선택 사항)**를 선택합니다. `--JOB_NAME`을 키로 입력하고 값을 제공합니다. AWS Command Line Interface(AWS CLI) 또는 AWS Glue API를 사용하여 이 파라미터를 설정할 수도 있습니다. 이 `--JOB_NAME` 파라미터는 Spark에서 사용되며 Python 쉘 환경 작업에는 필요하지 않습니다.
모든 파라미터 이름 앞에 `--`을 추가해야 합니다. 그렇지 않으면 코드가 작동하지 않습니다. 예를 들어 코드 스니펫의 경우 `--input_loc` 및 `--output_loc`를 사용하여 위치 파라미터를 간접적으로 호출해야 합니다. | 개발자, 클라우드 또는 데이터 | | ETL 작업을 실행합니다. | 작업을 실행하고 출력을 확인합니다. 소스 파일보다 공간이 얼마나 줄어들었는지 확인합니다. | 개발자, 클라우드 또는 데이터 | ## 관련 리소스 **참조** + [Apache Spark](https://spark.apache.org/) + [AWS Glue: 작동 방식](https://docs.aws.amazon.com/glue/latest/dg/how-it-works.html) + [AWS Glue 요금](https://aws.amazon.com/glue/pricing/) **튜토리얼 및 동영상 ** + [AWS Glue란 무엇입니까?](https://www.youtube.com/watch?v=qgWMfNSN9f4) ## 추가 정보 **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 라이브러리 제공](https://docs.aws.amazon.com/glue/latest/dg/add-job-python.html)을 참조하십시오. *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 = "s3://bucket-name/prefix/sample_data.csv" output_loc = "s3://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 = "s3://bucket-name/prefix/sample_data.csv" output_loc = "s3://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](samples/p-attach/8c926709-8fa4-417f-9aaf-bcc8113d018f/attachments/attachment.zip) 파일의 압축을 풉니다.