Gestione delle partizioni per l'output ETL in AWS Glue

Il partizionamento è una tecnica importante per organizzare i set di dati in modo da poterne eseguire query efficaci. I dati sono organizzati in una struttura gerarchica di directory basata su valori distinti di una o più colonne.

Ad esempio, puoi decidere di partizionare i log di un'applicazione in Amazon Simple Storage Service (Amazon S3) per data, suddivisi per anno, mese e giorno. I file che corrispondono ai dati di un solo giorno vengono quindi posti sotto un prefisso, ad esempio s3://my_bucket/logs/year=2018/month=01/day=23/. Sistemi come Amazon Athena, Amazon Redshift Spectrum e ora AWS Glue possono usare queste partizioni per filtrare i dati in base al valore di partizione senza dover leggere tutti i dati sottostanti da Amazon S3.

I crawler non deducono solo i tipi di file e gli schemi, ma identificano automaticamente anche la struttura della partizione del set di dati quando popolano AWS Glue Data Catalog. Le colonne di partizione risultanti sono disponibili per le query nei processi ETL AWS Glue o per motori di query come Amazon Athena.

Dopo aver eseguito il crawling di una tabella, puoi visualizzare le partizioni create dal crawler. Nella console AWS Glue, scegli Applications (Applicazioni) nel pannello di navigazione sinistro. Scegli la tabella creata dal crawler, quindi scegli View Partitions (Visualizza partizioni).

Per i percorsi partizionati di tipo Apache Hive nello stile key=val, i crawler popolano automaticamente il nome della colonna usando il nome della chiave. Altrimenti, utilizzano i nomi predefiniti, ad esempio partition_0, partition_1 e così via. È possibile modificare i nomi predefiniti sulla console. A tale scopo, accedi alla tabella. Controlla se gli indici esistono nella scheda Indici. In tal caso, è necessario eliminarli per procedere; potrai ricrearli in seguito utilizzando i nuovi nomi di colonna. Quindi, scegli Modifica schema e modifica i nomi delle colonne delle partizioni.

Negli script ETL, puoi applicare un filtro alle colonne di partizione. Poiché le informazioni sulla partizione sono memorizzate nel catalogo dati, utilizza le chiamate API from_catalog per includere le colonne delle partizioni nel DynamicFrame. Ad esempio, utilizza create_dynamic_frame.from_catalog anziché create_dynamic_frame.from_options.

Il partizionamento è una tecnica di ottimizzazione che riduce la scansione dei dati. Per ulteriori informazioni sul processo di identificazione del momento in cui questa tecnica è appropriata, consulta Ridurre la quantità di scansione dei dati nella guida Best practice per l'ottimizzazione delle prestazioni dei processi AWS per Apache Spark nel Prontuario AWS.

Prefiltraggio con i predicati pushdown

In molti casi, puoi utilizzare un predicato pushdown per filtrare le partizioni senza dover elencare e leggere tutti i file del set di dati. Invece di leggere l'intero set di dati e quindi filtrarli in un oggetto DynamicFrame, puoi applicare il filtro direttamente sui metadati della partizione nel catalogo dati. Elenchi e leggi solo quello che effettivamente ti serve in un DynamicFrame.

Ad esempio, in Python puoi scrivere quanto segue.

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

Questo crea un oggetto DynamicFrame che carica solo le partizioni nel catalogo dati che soddisfano l'espressione del predicato. A seconda di quanto piccolo sia il sottoinsieme di dati che stai caricando, questo può far risparmiare moltissimo tempo nell'elaborazione.

L'espressione del predicato può essere qualsiasi espressione booleana supportata da Spark SQL. Funziona tutto ciò che puoi inserire in una clausola WHERE in una query Spark SQL. Ad esempio, l'espressione del predicato pushDownPredicate = "(year=='2017' and month=='04')" carica solo le partizioni in nel catalogo dati che hanno sia year uguale a 2017 che month uguale a 04. Per ulteriori informazioni, consulta la documentazione di Apache Spark SQL, in particolare il riferimento alle funzioni SQL Scala.

Filtraggio lato server utilizzando predicati delle partizioni di catalogo

L'opzione push_down_predicate viene applicata dopo aver elencato tutte le partizioni dal catalogo e prima di pubblicare i file da Amazon S3 per tali partizioni. Se per una tabella ci sono molte partizioni, l'elenco delle partizioni del catalogo può comunque essere soggetto a un sovraccarico di tempo aggiuntivo. Per risolvere questo sovraccarico, puoi eliminare le partizioni lato server con l'opzione catalogPartitionPredicate che utilizza indici delle partizioni in AWS Glue Data Catalog. Questo rende il filtraggio delle partizioni molto più veloce quando sono presenti milioni di partizioni in una tabella. Puoi utilizzare sia push_down_predicate che catalogPartitionPredicate insieme in additional_options se il catalogPartitionPredicate richiede sintassi del predicato che non è ancora supportata con gli indici delle partizioni del catalogo.

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 usano sintassi diverse. Il primo utilizza la sintassi standard Spark SQL e il secondo utilizza il parser JSQL.

Scrittura delle partizioni

Per impostazione predefinita, un DynamicFrame non è partizionato quando viene scritto. Tutti i file di output vengono scritti nel livello principale del percorso di output specificato. Fino a poco tempo fa, l'unico modo per scrivere un DynamicFrame nelle partizioni era convertirlo in un frame di dati SQL DataFrame prima della scrittura.

Ora però i DynamicFrames supportano il partizionamento nativo utilizzando una sequenza di chiavi, utilizzando l'opzione partitionKeys quando crei un sink. Ad esempio, il codice Python seguente scrive un set di dati in Amazon S3 in formato Parquet, in directory partizionate in base al campo type. Da qui puoi elaborare le partizioni usando altri sistemi, ad esempio Amazon Athena.

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