Gestion des partitions pour la sortie ETL dans AWS Glue

Le partitionnement est une technique importante technique pour organiser les ensembles de données afin qu'ils puissent être interrogés de manière efficace. Il organise les données en une structure de répertoires hiérarchique fondée sur les valeurs distinctes d'une ou de plusieurs colonnes.

Par exemple, vous pouvez décider de partitionner vos journaux d'application dans Amazon Simple Storage Service (Amazon S3) par date, réparties par année, par mois et par jour. Les fichiers qui correspondent à une seule journée de données sont ensuite placés sous un préfixe, tel que s3://my_bucket/logs/year=2018/month=01/day=23/. Les systèmes comme Amazon Athena, Amazon Redshift Spectrum, et maintenant AWS Glue peuvent utiliser ces partitions pour filtrer les données en fonction de la valeur de partition, sans avoir à lire toutes les données sous-jacentes depuis Amazon S3.

Les crawlers déduisent non seulement les types de fichiers et les schémas, mais ils identifient aussi automatiquement la structure de partition de votre ensemble de données lorsqu'ils remplissent AWS Glue Data Catalog. Les colonnes de partition résultantes sont disponibles pour l'interrogation dans les tâches ETL AWS Glue ou les moteurs de requête tels qu'Amazon Athena.

Après avoir analysé une table, vous pouvez afficher les partitions créées par l'crawler. Dans la console AWS Glue, dans le panneau de navigation de gauche, sélectionnez Tables. Sélectionnez la table créée par l'crawler, puis View partitions (Afficher les partitions).

Pour les chemins partitionnés de style Apache Hive de type key=val, les crawlers remplissent automatiquement le nom de la colonne à l'aide du nom de clé. Sinon, ils utilisent des noms par défaut comme partition_0, partition_1, etc. Vous pouvez modifier les noms par défaut sur la console. Pour ce faire, accédez à la table. Vérifiez si des index existent sous l'onglet Index. Si tel est le cas, vous devez les supprimer pour continuer (vous pourrez ensuite les recréer en utilisant les nouveaux noms de colonnes). Choisissez ensuite Modifier le schéma et modifiez les noms des colonnes de partition qui s'y trouvent.

Dans vos scripts ETL, vous pouvez ensuite filtrer sur les colonnes de partition. Étant donné que les informations de partition sont stockées dans le catalogue de données, utilisez les appels d'API from_catalog pour inclure les colonnes de partition dans le fichier DynamicFrame. Par exemple, utilisez create_dynamic_frame.from_catalog plutôt que create_dynamic_frame.from_options.

Le partitionnement est une technique d'optimisation qui réduit l'analyse des données. Pour plus d'informations sur le processus permettant d'identifier le moment où cette technique est appropriée, consultez la section Réduire la quantité de données analysées dans le guide des meilleures pratiques pour le réglage des performances d'AWS Glue pour les tâches Apache Spark sur les Recommandations AWS.

Préfiltrage à l'aide des prédicats pushdown

Dans de nombreux cas, vous pouvez utiliser un prédicat pushdown pour filtrer les partitions sans avoir à répertorier et à lire tous les fichiers de votre ensemble de données. Au lieu de lire l'ensemble des données, puis de filtrer dans un DynamicFrame, vous pouvez appliquer le filtre directement sur les métadonnées de la partition de Data Catalog. Ensuite, vous affichez et lisez uniquement ce dont vous avez réellement besoin dans un DynamicFrame.

Par exemple, en Python, vous pouvez écrire ce qui suit.

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

Cela crée un DynamicFrame qui charge uniquement les partitions de Data Catalog qui satisfont l'expression du prédicat. En fonction de la taille d'un sous-ensemble de vos données de chargement, vous pouvez économiser beaucoup de temps de traitement.

L'expression de prédicat peut être n'importe quelle expression booléenne prise en charge par Spark SQL. Tout ce que vous pouvez placer dans une clause WHERE d'une requête SQL Spark fonctionne. Par exemple, l'expression de prédicat pushDownPredicate = "(year=='2017' and month=='04')" charge uniquement les partitions dans Data Catalog qui disposent à la fois d'une year égale à 2017 et un month égal à 04. Pour plus d'informations, consultez la documentation Apache Spark SQL, en particulier le document Référence des fonctions Scala SQL.

Filtrage côté serveur à l'aide de prédicats de partition de catalogue

L'option push_down_predicate est appliquée après avoir répertorié toutes les partitions du catalogue et avant de répertorier les fichiers d'Amazon S3 pour ces partitions. Si vous disposez d'un grand nombre de partitions pour une table, la liste des partitions du catalogue peut encore entraîner une surcharge de temps supplémentaire. Pour remédier à cette surcharge, vous pouvez utiliser l'élagage de partition côté serveur avec l'option catalogPartitionPredicate qui utilise les index de partition dans AWS Glue Data Catalog. Cela accélère considérablement le filtrage des partitions lorsque vous avez des millions de partitions dans une table. Vous pouvez utiliser à la fois push_down_predicate et catalogPartitionPredicate dans additional_options si votre catalogPartitionPredicate a besoin d'une syntaxe de prédicat qui n'est pas encore prise en charge avec les index de partition de catalogue.

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

Note

push_down_predicate et catalogPartitionPredicate utilisent des syntaxes différentes. Le premier utilise la syntaxe standard SQL Spark et le dernier utilise l'analyseur JSQL.

Écriture des partitions

Par défaut, un DynamicFrame n'est pas partitionné lorsqu'il est écrit. Tous les fichiers de sortie sont écrits au niveau supérieur du chemin de sortie spécifié. Jusqu'à récemment, le seul moyen d'écrire un DynamicFrame en partitions était de le convertir en un DataFrame Spark SQL avant d'écrire.

Toutefois, les DynamicFrames prennent désormais en charge le partitionnement natif à l'aide d'une séquence de clés, à l'aide de l'option partitionKeys lorsque vous créez un récepteur. Par exemple, le code Python suivant écrit un ensemble de données sur Amazon S3 au format Parquet, dans les répertoires partitionnés par le type de champ. À partir de là, vous pouvez traiter ces partitions à l'aide d'autres systèmes, comme Amazon Athena.

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