Gerenciar partições para saída de ETL no AWS Glue

O particionamento é uma técnica importante para organizar os conjuntos de dados para que possam ser consultados com eficiência. Ele organiza dados em uma estrutura hierárquica de diretórios com base nos valores distintos de uma ou mais colunas.

Por exemplo, você pode decidir particionar os logs da sua aplicação no Amazon Simple Storage Service (Amazon S3) por data, divididos em ano, mês e dia. Os arquivos que correspondem a um único dia de dados são colocados com um prefixo, como s3://my_bucket/logs/year=2018/month=01/day=23/. Sistemas como o Amazon Athena, o Amazon Redshift Spectrum e agora o AWS Glue podem usar essas partições para filtrar dados por valor de partição sem precisar ler todos os dados subjacentes do Amazon S3.

Os crawlers não apenas inferem tipos de arquivo e esquemas, como também identificam automaticamente a estrutura da partição do seu conjunto de dados quando eles preenchem o AWS Glue Data Catalog. As colunas resultantes da partição ficam disponíveis para consulta em trabalhos de ETL do AWS Glue ou em mecanismos de consulta, como o Amazon Athena.

Após rastrear uma tabela, você pode visualizar as partições que o crawler criou. No console do AWS Glue, escolha Tables (Tabelas) no painel de navegação esquerdo. Escolha a tabela criada pelo crawler e, em seguida, selecione View Partitions (Exibir partições).

Para os caminhos particionados do Apache Hive no estilo key=val, os crawlers preenchem automaticamente o nome da coluna usando o nome da chave padrão. Caso contrário, ele usa nomes padrão, como partition_0, partition_1 e assim por diante. Você pode alterar os nomes padrão no console. Para fazer isso, navegue até a tabela. Verifique se os índices existem na guia Índices. Se for esse o caso, você precisa excluí-las para continuar (você poderá recriá-las posteriormente usando os novos nomes das colunas). Em seguida, escolha Editar esquema e modifique os nomes das colunas de partição.

Nos scripts de ETL, você pode filtrar as colunas da partição. Uma vez que as informações de partição são armazenadas no Data Catalog, utilize as chamadas de API from_catalog para incluir as colunas de partição no DynamicFrame. Por exemplo, use create_dynamic_frame.from_catalog em vez de create_dynamic_frame.from_options.

Particionamento é uma técnica de otimização que reduz as varreduras de dados. Para obter mais informações sobre o processo para identificar quando essa técnica é adequada, consulte Reduce the amount of data scan no guia Best practices for performance tuning AWS Glue for Apache Spark jobs em AWS Prescriptive Guidance.

Pré-filtragem usando a aplicação de predicados

Em muitos casos, você pode usar uma aplicação de predicado para filtrar partições sem precisar listar e ler todos os arquivos do seu conjunto de dados. Em vez de ler todo o conjunto de dados e, em seguida, filtrá-lo em um DynamicFrame, você pode aplicar o filtro diretamente nos metadados da partição no Data Catalog. Em seguida, você lista e lê somente o que você realmente precisa em um DynamicFrame.

Por exemplo, em Python, você pode gravar o seguinte.

glue_context.create_dynamic_frame.from_catalog(
    database = "my_S3_data_set",
    table_name = "catalog_data_table",
    push_down_predicate = my_partition_predicate)

Isso cria um DynamicFrame que carrega somente as partições do Data Catalog que satisfazem a expressão do predicado. Dependendo do tamanho do subconjunto dos dados que você está carregando, isso pode economizar muito tempo de processamento.

A expressão do predicado pode ser qualquer expressão booleana compatível com Spark SQL. Tudo que você pode colocar em uma cláusula WHERE de uma consulta do SQL Spark funcionará. Por exemplo, a expressão de predicado pushDownPredicate = "(year=='2017' and month=='04')" carrega apenas as partições no Data Catalog que tiverem tanto year igual a 2017 quanto month igual a 04. Para obter mais informações, consulte a Documentação do Apache Spark SQL e, em especial, a Referência das funções SQL em Scala.

Filtragem do lado do servidor usando predicados de partição de catálogo

A opção push_down_predicate é aplicada depois de listar todas as partições do catálogo e antes de listar arquivos do Amazon S3 para essas partições. Se você tiver muitas partições para uma tabela, a listagem de partições de catálogo ainda poderá incorrer em sobrecarga de tempo adicional. Para resolver essa sobrecarga, você pode usar a remoção de partição do lado do servidor com a opção catalogPartitionPredicate, que usaíndices de partição no AWS Glue Data Catalog. Isso torna a filtragem de partição muito mais rápida quando você tem milhões de partições em uma tabela. Você pode usar push_down_predicate e catalogPartitionPredicate em additional_options juntos, se seu catalogPartitionPredicate exigir sintaxe de predicado que ainda não é suportada com os índices de partição de catálogo.

Python:

dynamic_frame = glueContext.create_dynamic_frame.from_catalog(
    database=dbname, 
    table_name=tablename,
    transformation_ctx="datasource0",
    push_down_predicate="day>=10 and customer_id like '10%'",
    additional_options={"catalogPartitionPredicate":"year='2021' and month='06'"}
)

Scala:

val dynamicFrame = glueContext.getCatalogSource(
    database = dbname,
    tableName = tablename, 
    transformationContext = "datasource0",
    pushDownPredicate="day>=10 and customer_id like '10%'",
    additionalOptions = JsonOptions("""{
        "catalogPartitionPredicate": "year='2021' and month='06'"}""")
    ).getDynamicFrame()

nota

push_down_predicate e catalogPartitionPredicate usam sintaxes diferentes. O primeiro usa a sintaxe padrão do Spark SQL e o outro usa o analisador JSQL.

Gravar partições

Por padrão, um DynamicFrame não é particionado quando é gravado. Todos os arquivos de saída são gravados no nível superior do caminho de saída especificado. Até recentemente, a única maneira de gravar um DynamicFrame em partições era convertê-lo em um DataFrame Spark SQL antes da gravação.

No entanto, DynamicFrames agora oferecem suporte ao particionamento nativo usando uma sequência de chaves, usando a opção partitionKeys ao criar um depósito. Por exemplo, o seguinte código Python grava um conjunto de dados no Amazon S3 no formato Parquet em diretórios particionados, de acordo com o tipo de campo. A partir daí, você pode processar essas partições usando outros sistemas, como o Amazon Athena.

glue_context.write_dynamic_frame.from_options(
    frame = projectedEvents,
    connection_type = "s3",    
    connection_options = {"path": "$outpath", "partitionKeys": ["type"]},
    format = "parquet")