Ridurre la quantità di scansione dei dati

Dimensioni del file: consigliamo di mantenere le dimensioni dei file di input e output entro un intervallo moderato (ad esempio, 128 MB). File troppo piccoli e file troppo grandi possono causare problemi.

Un numero elevato di file di piccole dimensioni causa i seguenti problemi:

D'altra parte, se il tipo di file non è divisibile (come gzip) e i file sono troppo grandi, l'applicazione Spark deve attendere il completamento di una singola operazione di lettura dell'intero file.

Per ridurre l'eccessivo parallelismo che si verifica quando viene creata un'attività di Apache Spark per ogni file di piccole dimensioni, utilizzate il raggruppamento di file per. DynamicFrames Questo approccio riduce le possibilità di un'OOMeccezione dal driver Spark. Per configurare il raggruppamento dei file, impostate i parametri groupFiles andgroupSize. L'esempio di codice seguente utilizza lo script AWS Glue DynamicFrame API in uno ETL script con questi parametri.

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

Compressione: se i tuoi oggetti S3 misurano centinaia di megabyte, valuta la possibilità di comprimerli. Esistono vari formati di compressione, che possono essere generalmente classificati in due tipi:

Per Spark (e altri motori di elaborazione distribuita comuni), suddividerai il file di dati sorgente in blocchi che il tuo motore può elaborare in parallelo. Queste unità vengono spesso chiamate split. Dopo che i dati sono in un formato divisibile, i AWS Glue lettori ottimizzati possono recuperare le suddivisioni da un oggetto S3 offrendo la Range possibilità di recuperare solo blocchi specifici. GetObject API Considerate il diagramma seguente per vedere come funzionerebbe in pratica.

I dati compressi possono velocizzare notevolmente l'applicazione, purché i file abbiano una dimensione ottimale o siano divisibili. Le dimensioni dei dati più piccole riducono i dati scansionati da Amazon S3 e il traffico di rete da Amazon S3 al cluster Spark. D'altra parte, CPU è necessario fare di più per comprimere e decomprimere i dati. La quantità di calcolo richiesta varia in base al rapporto di compressione dell'algoritmo di compressione. Considerate questo compromesso quando scegliete il formato di compressione divisibile.

Formato di file: utilizza un formato colonnare. Apache Parquet e Apache ORC sono i formati di dati colonnari più diffusi. Parquet e ORC archivia i dati in modo efficiente utilizzando la compressione, la codifica e la compressione basate su colonne in base al tipo di dati. Per ulteriori informazioni sulle codifiche Parquet, consulta Definizioni di codifica Parquet. I file Parquet sono anche divisibili.

I formati colonnari raggruppano i valori per colonna e li memorizzano in blocchi. Quando usi i formati colonnari, puoi saltare i blocchi di dati che corrispondono a colonne che non intendi utilizzare. Le applicazioni Spark possono recuperare solo le colonne di cui hai bisogno. In genere, rapporti di compressione migliori o il salto di blocchi di dati implicano la lettura di un numero inferiore di byte da Amazon S3, con conseguente miglioramento delle prestazioni. Entrambi i formati supportano anche i seguenti approcci pushdown per ridurre l'I/O:

Layout dei file: archiviando i dati S3 su oggetti in percorsi diversi in base a come verranno utilizzati i dati, puoi recuperare in modo efficiente i dati pertinenti. Per ulteriori informazioni, consulta Organizzazione degli oggetti utilizzando i prefissi nella documentazione di Amazon S3. AWS Glue supporta la memorizzazione di chiavi e valori nei prefissi Amazon S3 nel formatokey=value, partizionando i dati in base al percorso Amazon S3. Partizionando i dati, puoi limitare la quantità di dati scansionati da ciascuna applicazione di analisi downstream, migliorando le prestazioni e riducendo i costi. Per ulteriori informazioni, consulta Gestione delle partizioni per l'output in. ETL AWS Glue

Il partizionamento divide la tabella in diverse parti e mantiene i dati correlati in file raggruppati in base ai valori delle colonne come anno, mese e giorno, come illustrato nell'esempio seguente.

# 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
...

È possibile definire le partizioni per il set di dati modellandolo con una tabella in. AWS Glue Data CatalogÈ quindi possibile limitare la quantità di dati scansionati utilizzando la rimozione delle partizioni come segue:

Troppe partizioni Amazon S3: evita di partizionare i dati Amazon S3 su colonne che contengono un'ampia gamma di valori, ad esempio una colonna ID con migliaia di valori. Ciò può aumentare notevolmente il numero di partizioni nel bucket, poiché il numero di partizioni possibili è il prodotto di tutti i campi in base ai quali hai partizionato. Troppe partizioni potrebbero causare quanto segue:

Ad esempio, quando si imposta un tipo di data in partitionBy orpartitionKeys, il partizionamento a livello di data yyyy/mm/dd è utile per molti casi d'uso. Tuttavia, yyyy/mm/dd/<ID> potrebbe generare così tante partizioni da influire negativamente sulle prestazioni complessive.

D'altra parte, alcuni casi d'uso, come le applicazioni di elaborazione in tempo reale, richiedono molte partizioni come. yyyy/mm/dd/hh Se il tuo caso d'uso richiede partizioni sostanziali, prendi in considerazione l'utilizzo di indici di AWS Glue partizione per ridurre la latenza per il recupero dei metadati delle partizioni dal Data Catalog.

Per DynamicFrame, utilizzate l'sampleQueryopzione. Durante l'utilizzocreate_dynamic_frame.from_catalog, configura l'additional_optionsargomento come segue.

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

Quandousing create_dynamic_frame.from_options, configura l'connection_optionsargomento come segue.

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

Per DataFrame, usa l'opzione di interrogazione.

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

Per Amazon Redshift, usa la AWS Glue versione 4.0 o successiva per sfruttare il supporto pushdown nel connettore 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"}
)

Per altri database, consulta la documentazione relativa a quel database.

Per evitare una scansione completa per tutte le esecuzioni continue dei lavori ed elaborare solo i dati che non erano presenti durante l'ultima esecuzione del processo, abilita i segnalibri dei lavori.

Per limitare la quantità di dati di input da elaborare, abilita l'esecuzione limitata con i segnalibri dei lavori. Questo aiuta a ridurre la quantità di dati scansionati per ogni esecuzione di lavoro.