Três tipos de trabalho de ETL do AWS Glue para converter dados em Apache Parquet

Criado por Adnan Alvee (AWS), Karthikeyan Ramachandran e Nith Govindasivan (AWS)

Resumo

Na nuvem da Amazon Web Services (AWS), o AWS Glue é um serviço para extração, transformação e carregamento (ETL) totalmente gerenciado. O AWS Glue torna econômico categorizar os dados, limpá-los, aprimorá-los e movê-los de modo confiável entre vários armazenamentos e fluxos de dados.

Esse padrão fornece diferentes tipos de trabalho no AWS Glue e usa três scripts diferentes para demonstrar a criação de trabalhos de ETL.

Você pode usar o AWS Glue para escrever trabalhos de ETL em um ambiente de shell Python. Você também pode criar trabalhos ETL em lote e de streaming usando Python PySpark () ou Scala em um ambiente gerenciado do Apache Spark. Para começar a criar trabalhos de ETL, esse padrão se concentra em trabalhos ETL em lote usando Python, shell e Scala. PySpark Os trabalhos de shell do Python são destinados a workloads que exigem menor poder computacional. O ambiente gerenciado do Apache Spark é destinado a workloads que exigem alto poder computacional.

O Apache Parquet foi desenvolvido para dar suporte a esquemas eficientes de compressão e codificação. Ele pode acelerar suas workloads de análise porque armazena dados de forma colunar. A conversão de dados em Parquet pode economizar espaço de armazenamento, custo e tempo no longo prazo. Para saber mais sobre o Parquet, consulte a postagem do blog Apache Parquet: Como ser um herói com o formato de dados colunares de código aberto.

Pré-requisitos e limitações

Pré-requisitos

Arquitetura

Pilha de tecnologias de destino

Automação e escala

Ferramentas

Serviços da AWS

Outras ferramentas

Configuração

Use os dados a seguir para configurar a potência computacional do AWS Glue ETL. Para reduzir custos, use as configurações mínimas ao executar a workload fornecida nesse padrão. 

A tabela a seguir mostra os diferentes tipos de operadores do AWS Glue para o ambiente Apache Spark. Como um trabalho de Python shell não usa o ambiente Apache Spark para executar o Python, ele não está incluído na tabela.

Código

Para ver o código usado nesse padrão, incluindo o perfil do IAM e a configuração de parâmetros, consulte a seção Informações adicionais.

Épicos

Recursos relacionados

Referências

Tutoriais e vídeos

Mais informações

Perfil do IAM

Ao criar os trabalhos do AWS Glue, você pode usar um perfil existente do IAM que tenha as permissões mostradas no seguinte trecho de código ou uma nova função.

Use o seguinte código YAML para criar um novo perfil.

# (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 ] ]

Phyton shell do AWS Glue

O código Python usa os Pandas e as PyArrow bibliotecas para converter dados em Parquet. A biblioteca Pandas já está disponível. A PyArrow biblioteca é baixada quando você executa o padrão, porque é uma execução única. Você pode usar arquivos de roda PyArrow para converter em uma biblioteca e fornecer o arquivo como um pacote de biblioteca. Para obter mais informações sobre empacotamento de arquivos wheel, consulte Fornecer sua própria biblioteca Python.

Parâmetros do Python shell do AWS Glue

from awsglue.utils import getResolvedOptions

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

Código do Phyton shell do AWS Glue

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()) 

Trabalho do AWS Glue Spark com Python

Para usar um tipo de trabalho do AWS Glue Spark com Python, escolha Spark como tipo de trabalho. Escolha o Spark 3.1, Python 3 com melhor tempo de startup do trabalho (Glue versão 3.0) como a versão do AWS Glue.

Parâmetros do Python do AWS Glue

from awsglue.utils import getResolvedOptions

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

Trabalho do AWS Glue Spark com código Python

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

Para um grande número de arquivos grandes compactados (por exemplo, 1.000 arquivos com cerca de 3 MB cada), use o parâmetro compressionType com o parâmetro recurse para ler todos os arquivos que estão disponíveis dentro do prefixo, conforme mostrado no código a seguir.

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": ","}
                    )

Para um grande número de arquivos pequenos compactados (por exemplo, 1.000 arquivos cada um com cerca de 133 KB), use o parâmetro groupFiles junto com os parâmetros compressionType e os parâmetros recurse. O parâmetro groupFiles agrupa arquivos pequenos em vários arquivos grandes e o parâmetro groupSize controla o agrupamento no tamanho especificado em bytes (por exemplo, 1 MB). O trecho de código a seguir fornece um exemplo do uso desses parâmetros no código.

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": ","}
                    )

Sem nenhuma alteração nos nós de processamento, essas configurações permitem que o trabalho do AWS Glue leia vários arquivos (grandes ou pequenos, com ou sem compactação) e os grave no destino no formato Parquet.

Trabalho do AWS Glue Spark com Scala

Para usar um tipo de trabalho do AWS Glue Spark com Scala, escolha Spark como tipo de trabalho e Linguagem como Scala. Escolha o Spark 3.1, Scala 2 com melhor tempo de startup do trabalho (Glue versão 3.0) como a versão do AWS Glue. Para economizar espaço de armazenamento, o seguinte exemplo do AWS Glue com Scala também usa o atributo applyMapping para converter tipos de dados.

Parâmetros do AWS Glue Scala

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

Trabalho do AWS Glue Spark com código Scala

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)
  }
}

Anexos

Para acessar o conteúdo adicional associado a este documento, descompacte o seguinte arquivo: attachment.zip