MaxCompute:Hash Clustering

Quand utiliser le hash clustering

Le hash clustering est particulièrement efficace lorsque votre table remplit toutes les conditions suivantes :

• La table contient un volume de données important.

• Les requêtes filtrent, joignent ou agrègent fréquemment les mêmes colonnes.

• Le même schéma de jointure ou d'agrégation s'exécute de manière répétée sur la table.

• La table possède un cycle de vie long, ce qui rend les économies de stockage pertinentes.

Si les requêtes adressant la table utilisent des modèles d'accès variés et imprévisibles, le hash clustering n'apporte qu'un bénéfice limité.

Fonctionnement

Dans de nombreux scénarios, il est nécessaire de joindre des tables. MaxCompute propose les méthodes de jointure suivantes :

• Broadcast hash join : si l'une des tables à joindre est de petite taille, MaxCompute diffuse cette table vers toutes les instances de tâche de jointure et effectue une jointure par hachage entre la petite table et la grande table.

• Shuffle hash join : si les deux tables sont volumineuses, un hash shuffle est appliqué aux deux tables en fonction des clés de jointure. Les enregistrements partageant les mêmes valeurs de clé sont distribués à la même instance de tâche de jointure, qui exécute ensuite la jointure.

• Sort merge join : si les deux tables sont très volumineuses (la mémoire disponible est insuffisante pour une table de hachage), MaxCompute effectue un hash shuffle sur les deux tables selon les clés de jointure, trie les résultats, puis les fusionne. Dans un job comportant M mappeurs et R réducteurs, cela génère M × R opérations d'E/S. Il s'agit de la méthode de jointure la plus courante dans MaxCompute.

Le plan d'exécution physique d'un job Fuxi pour une sort merge join nécessite deux étapes de map et une étape de jointure. Les opérations de shuffle et de tri sont mises en évidence ci-dessous.

Avec le hash clustering, les données sont distribuées dans des buckets par valeur de hachage et triées au sein de chaque bucket lors de l'écriture. Les requêtes ultérieures qui effectuent des jointures ou des filtres sur la clé de cluster ignorent entièrement le shuffle et le tri, réduisant ainsi un job Fuxi multi-étapes à une seule étape.

Trois optimisations spécifiques s'appliquent :

Bucket pruning et recherche d'index : une requête comportant une clause WHERE sur la clé de cluster lit uniquement le bucket correspondant au lieu de tous les buckets. MaxCompute génère également des index automatiquement lorsque SORTED BY est défini, permettant à la recherche d'index de localiser les enregistrements correspondants par page. Par exemple, une requête qui correspond à 26 enregistrements sur 42,7 milliards peut passer de 1 111 mappeurs analysant toutes les données à 4 mappeurs analysant 10 000 enregistrements, réduisant ainsi le temps d'exécution de 1 minute et 48 secondes à 6 secondes.

Optimisation des agrégations : une requête GROUP BY sur la clé de cluster ignore le shuffle et le tri. MaxCompute effectue une agrégation en flux directement sur les données pré-triées.

Optimisation du stockage : MaxCompute utilise un stockage en colonnes au niveau de la couche de stockage. Les données triées se compressent nettement mieux que les données non triées, car les enregistrements ayant des valeurs de clé similaires sont stockés ensemble. En pratique, le même jeu de données avec hash clustering utilise environ 10 % de stockage en moins ; dans les cas extrêmes, les économies peuvent atteindre 50 %.

• Le hash clustering n'est pas utilisé.

• Le hash clustering est utilisé.

Résultats du benchmark TPC-H : sur 1 To de données réparties dans 500 buckets, le hash clustering a réduit le temps CPU total d'environ 17,3 % et la durée totale des jobs d'environ 12,8 %. Pour les requêtes qui exploitent efficacement le clustering, telles que TPC-H Q4, Q12 et Q10, l'amélioration était respectivement de 68 %, 62 % et 47 %.

La figure suivante présente le plan d'exécution d'un job Fuxi pour les requêtes TPC-H Q4 sur une table standard. La figure suivante montre le plan d'exécution après activation du hash clustering. Le graphe orienté acyclique (DAG) est considérablement simplifié, ce qui constitue la clé de l'amélioration des performances.

Créer une table avec hash clustering

CREATE TABLE [IF NOT EXISTS] <table_name>
             [(<col_name> <data_type> [COMMENT <col_comment>], ...)]
             [COMMENT <table_comment>]
             [PARTITIONED BY (<col_name> <data_type> [COMMENT <col_comment>], ...)]
             [CLUSTERED BY (<col_name> [, <col_name>, ...])
             [SORTED BY (<col_name> [ASC | DESC] [, <col_name> [ASC | DESC] ...])]
             INTO <number_of_buckets> BUCKETS] [AS <select_statement>]

CREATE TABLE T1 (a string, b string, c bigint)
             CLUSTERED BY (c)
             SORTED BY (c) INTO 1024 BUCKETS;

CREATE TABLE T1 (a string, b string, c bigint)
       PARTITIONED BY (dt string)
       CLUSTERED BY (c)
       SORTED BY (c) INTO 1024 BUCKETS;

Modifier les propriétés de clustering d'une table

Pour les tables partitionnées, utilisez ALTER TABLE pour ajouter ou supprimer les propriétés de hash clustering.

-- Add hash clustering to an existing partitioned table
ALTER TABLE <table_name> [CLUSTERED BY (<col_name> [, <col_name>, ...])
                       [SORTED BY (<col_name> [ASC | DESC] [, <col_name> [ASC | DESC] ...])]
                       INTO <number_of_buckets> BUCKETS];

-- Remove hash clustering from a partitioned table
ALTER TABLE <table_name> NOT CLUSTERED;

Vérifier les propriétés de hash clustering

Après avoir créé une table avec hash clustering, exécutez DESC EXTENDED pour confirmer les propriétés de clustering. Recherchez la configuration du hash clustering dans Extended Info de la sortie.

DESC EXTENDED <table_name>;

La figure suivante montre un exemple du résultat renvoyé.

Pour les tables partitionnées, vérifiez également une partition spécifique :

DESC EXTENDED <table_name> PARTITION (<pt_spec>);

La figure suivante montre un exemple du résultat renvoyé.

Détails de l'optimisation du stockage

Une expérience simple a été réalisée en utilisant la table lineitem avec 100 Go de données issues d'un jeu de données TPC-H. La table contient des données de divers types tels que INT, DOUBLE et STRING. Avec les mêmes données et la même méthode de compression, la table avec hash clustering économise environ 10 % d'espace de stockage. Dans certains cas de test extrêmes, une table triée peut économiser jusqu'à 50 % d'espace de stockage par rapport à une table non triée.

Étapes suivantes

• Tables partitionnées — gérer les partitions de table dans MaxCompute

• Optimiser les performances des requêtes — techniques supplémentaires pour améliorer l'efficacité des requêtes