Reduzir a quantidade de dados verificados

Tamanho do arquivo: recomendamos manter o tamanho do arquivo de entradas e saídas dentro de uma faixa moderada (por exemplo, 128 MB). Arquivos muito pequenos e muito grandes podem causar problemas.

Um grande número de arquivos pequenos causa os seguintes problemas:

Por outro lado, se o tipo de arquivo não for divisível (como gzip) e os arquivos forem muito grandes, a aplicação do Spark deverá esperar até que uma única tarefa termine de ler o arquivo todo.

Para reduzir o paralelismo excessivo que ocorre quando uma tarefa do Apache Spark é criada para cada arquivo pequeno, use o agrupamento de arquivos para. DynamicFrames Essa abordagem reduz a probabilidade de uma exceção de OOM do driver do Spark. Para configurar o agrupamento de arquivos, defina os parâmetros groupFiles e groupSize. O exemplo de código a seguir usa a AWS Glue DynamicFrame API em um script ETL com esses parâmetros.

dyf = glueContext.create_dynamic_frame_from_options("s3",
    {'paths': ["s3://input-s3-path/"],
    'recurse':True,
    'groupFiles': 'inPartition',
    'groupSize': '1048576'},
    format="json")

Compactação: se seus objetos do S3 estiverem na casa das centenas de megabytes, considere compactá-los. Existem vários formatos de compressão, que podem ser amplamente classificados em dois tipos:

Para o Spark (e outros mecanismos comuns de processamento distribuído), você dividirá seu arquivo de dados de origem em fragmentos que seu mecanismo possa processar paralelamente. Essas unidades geralmente são chamadas de splits. Depois que seus dados estiverem em um formato divisível, os AWS Glue leitores otimizados poderão recuperar divisões de um objeto do S3 fornecendo à GetObject API a Range opção de recuperar somente blocos específicos. Considere o diagrama a seguir para ver como isso funcionaria na prática.

Os dados compactados podem acelerar significativamente sua aplicação, desde que os arquivos tenham um tamanho ideal ou sejam divisíveis. Os tamanhos de dados menores reduzem os dados verificados do Amazon S3 e o tráfego de rede do Amazon S3 para seu cluster do Spark. Por outro lado, é necessária mais CPU para compactar e descompactar dados. A quantidade de computação necessária é escalada com a taxa de compactação do seu algoritmo de compactação. Considere essa compensação ao escolher seu formato de compactação divisível.

Formato de arquivo: use um formato colunar. O Apache Parquet e o Apache ORC são formatos de dados colunares conhecidos. O Parquet e o ORC armazenam dados de forma eficiente empregando compactação baseada em colunas, codificando e compactando cada coluna com base em seu tipo de dados. Para obter mais informações sobre codificações do Parquet, consulte Parquet encoding definitions. Os arquivos do Parquet também são divisíveis.

Os formatos colunares agrupam valores por coluna e os armazenam juntos em blocos. Ao usar formatos colunares, você pode ignorar blocos de dados que correspondam às colunas que você não planeja usar. As aplicações do Spark podem recuperar somente as colunas de que você precisa. Geralmente, melhores taxas de compactação ou ignorar blocos de dados significa ler menos bytes do Amazon S3, o que resulta em uma melhor performance. Ambos os formatos também são compatíveis com as seguintes abordagens de pushdown para reduzir a E/S:

Layout do arquivo: ao armazenar seus dados do S3 em objetos em caminhos diferentes com base em como os dados serão usados, você pode recuperar dados relevantes com eficiência. Para obter mais informações, consulte Organizar objetos usando prefixos na documentação do Amazon S3. O AWS Glue tem suporte para armazenamento de chaves e valores nos prefixos do Amazon S3 no formato key=value, particionando seus dados pelo caminho do Amazon S3. Ao particionar seus dados, você pode restringir a quantidade de dados verificados por cada aplicação de analytics downstream, melhorando a performance e reduzindo o custo. Para obter mais informações, consulte Gerenciando partições para saída ETL em. AWS Glue

O particionamento divide sua tabela em partes diferentes e mantém os dados relacionados em arquivos agrupados com base nos valores das colunas, como ano, mês e dia, conforme mostrado no exemplo a seguir.

# Partitioning by /YYYY/MM/DD
s3://<YourBucket>/year=2023/month=03/day=31/0000.gz
s3://<YourBucket>/year=2023/month=03/day=01/0000.gz
s3://<YourBucket>/year=2023/month=03/day=02/0000.gz
s3://<YourBucket>/year=2023/month=03/day=03/0000.gz
...

Você pode definir partições para seu conjunto de dados modelando-o com uma tabela no AWS Glue Data Catalog. Você pode então restringir a quantidade de dados verificados usando o descarte de partições da seguinte forma:

Muitas partições do Amazon S3: evite particionar seus dados do Amazon S3 em colunas que contenham uma grande variedade de valores, como uma coluna de ID com milhares de valores. Isso pode aumentar substancialmente o número de partições em seu bucket, porque o número de partições possíveis é o produto de todos os campos pelos quais você particionou. Muitas partições podem causar o seguinte:

Por exemplo, quando você define um tipo de data em partitionBy ou partitionKeys, o particionamento em nível de data como yyyy/mm/dd é bom para muitos casos de uso. No entanto, yyyy/mm/dd/<ID> pode gerar tantas partições que isso afetaria negativamente a performance como um todo.

Por outro lado, alguns casos de uso, como aplicações de processamento em tempo real, exigem muitas partições, como yyyy/mm/dd/hh. Se seu caso de uso exigir partições substanciais, considere usar índices de partição do AWS Glue para reduzir a latência na recuperação de metadados de partições do Catálogo de Dados.

Para DynamicFrame, use a opção SampleQuery. Ao usar create_dynamic_frame.from_catalog, configure o argumento additional_options como a seguir.

query = "SELECT * FROM <TableName> where id = 'XX' AND"
datasource0 = glueContext.create_dynamic_frame.from_catalog(
  database = db,
  table_name = table,
  additional_options={
      "sampleQuery": query,
      "hashexpression": key,
      "hashpartitions": 10,
      "enablePartitioningForSampleQuery": True
    },
  transformation_ctx = "datasource0"
)

Ao using create_dynamic_frame.from_options, configure o argumento connection_options como a seguir.

query = "SELECT * FROM <TableName> where id = 'XX' AND"
datasource0 = glueContext.create_dynamic_frame.from_options(
    connection_type = connection, 
    connection_options={
        "url": url,
        "user": user,
        "password": password,
        "dbtable": table,
        "sampleQuery": query,
        "hashexpression": key,
        "hashpartitions": 10,
        "enablePartitioningForSampleQuery": True
    }
)

Para DataFrame, use a opção de consulta.

query = "SELECT * FROM <TableName> where id = 'XX'"
jdbcDF = spark.read \
.format('jdbc') \
.option('url', url) \
.option('user', user) \
.option('password', pwd) \
.option('query', query) \
.load()

Para o Amazon Redshift, use a AWS Glue versão 4.0 ou posterior para aproveitar o suporte a push down no conector Amazon Redshift Spark.

dyf = glueContext.create_dynamic_frame.from_catalog(
    database = "redshift-dc-database-name", 
    table_name = "redshift-table-name", 
    redshift_tmp_dir = args["temp-s3-dir"], 
    additional_options = {"aws_iam_role": "arn:aws:iam::role-account-id:role/rs-role-name"}
)

Para outros bancos de dados, consulte a documentação do banco de dados.

Para evitar uma verificação completa de todas as execuções contínuas de trabalhos e processar somente os dados que não estavam presentes durante a última execução do trabalho, habilite os marcadores de tarefas.

Para limitar a quantidade de dados de entrada a serem processados, habilite a execução delimitada com marcadores de tarefas. Isso ajuda a reduzir a quantidade de dados verificados para cada execução de trabalho.