MaxCompute améliore le format Append Delta Table avec la prise en charge du cluster de hachage, ce qui améliore les performances des requêtes tout en permettant le traitement incrémentiel des données. Cette rubrique présente les différences par rapport aux autres types de tables, les détails de la syntaxe et des exemples d'utilisation pour SQL et le SDK Data Tunnel.
Cas d'utilisation
Le cluster de hachage est recommandé dans les scénarios suivants :
Requêtes avec filtres d'égalité : utilisez cette méthode pour les recherches ponctuelles ou les filtres d'égalité sur des colonnes spécifiques afin de réduire l'analyse des données.
Opérations Equi-joins et GROUP BY : réduisez le brassage des données lors de la jointure ou de l'agrégation de plusieurs tables sur la même clé.
Comparaison avec les types de tables similaires
|
Type de table
|
Méthode de clustering
|
Écriture incrémentielle (ACID)
|
|
|
Table à cluster de hachage
|
Hachage
|
Non pris en charge
|
Fournit une optimisation par cluster de hachage (Shuffle + Sort), mais ne prend pas en charge ACID.
|
|
Table PK Delta
|
Hachage
|
Pris en charge
|
Offre des fonctionnalités ACID et une optimisation par cluster de hachage. Convient aux données disposant d'une clé primaire. Les performances en lecture et en écriture sont inférieures à celles des tables sans clé primaire.
|
|
Table d'ajout Delta - Cluster de plage
|
Plage
|
Pris en charge
|
Offre des fonctionnalités ACID et prend en charge le reclustering, mais sa méthode de clustering par plage entraîne des performances d'écriture inférieures à celles du hachage.
|
|
Table d'ajout Delta - Cluster de hachage
|
Hachage
|
Pris en charge
|
Combine l'optimisation par cluster de hachage (Shuffle + Sort) avec des fonctionnalités ACID complètes. Elle prend également en charge le reclustering incrémentiel et complet en arrière-plan, ce qui en fait l'option la plus complète.
|
Prérequis
Avant de créer une table, activez les paramètres de session suivants :
SET odps.table.append2.enable=true;
SET odps.table.hash.delta.enable=true; -- Enables the trial feature for creating hash delta tables.
Syntaxe
CREATE TABLE [IF NOT EXISTS] <table_name>
[(<col_name> <data_type> [comment <col_comment>], ...)]
[PARTITIONED BY (<col_name> <data_type> [comment <col_comment>], ...)]
CLUSTERED BY (<col_name> [, <col_name>, ...])
[SORTED BY (<col_name> [, <col_name>, ...])] -- Only ascending order is supported.
INTO <number_of_buckets> BUCKETS
TBLPROPERTIES ('table.format.version' = '2');
Paramètres
|
Paramètre
|
Description
|
|
CLUSTERED BY
|
Spécifie les colonnes de regroupement (bucketing). Choisissez des colonnes fréquemment utilisées dans les requêtes avec filtres d'égalité, les equi-joins, les clauses GROUP BY ou WINDOW PARTITION BY. Pour obtenir les meilleurs résultats, sélectionnez des colonnes à forte cardinalité afin de garantir une répartition uniforme des données entre les buckets.
|
|
SORTED BY
|
Facultatif. Spécifie les colonnes de tri au sein de chaque bucket. Seul l'ordre croissant est actuellement pris en charge. Nous vous recommandons de choisir des colonnes utilisées pour les filtres de plage ou d'égalité, les calculs de fenêtre ou les horodatages de versioning.
|
|
INTO ... BUCKETS
|
Spécifie le nombre de buckets logiques. Nous vous recommandons de définir ce nombre en fonction du volume de vos données, de la concurrence des requêtes et de la cardinalité des colonnes de regroupement. Le nombre de buckets affecte le parallélisme des opérations d'écriture et les optimisations de shuffle lors des opérations de lecture.
|
|
table.format.version
|
Définissez la valeur sur 2 pour créer une table avec le format de données requis pour le cluster de hachage.
|
Exemples SQL
Cet exemple utilise une table de statut de produit et de version de prix. Dans la logique métier, les recherches ponctuelles ou les jointures sont généralement effectuées sur les produits en fonction de item_id. Par conséquent, item_id est désigné comme colonne de regroupement par hachage. L'heure d'efficacité de la version, event_time, est utilisée pour suivre les modifications historiques, donc event_time est désignée comme colonne de tri. Cette conception convient aux scénarios tels que les tables de dimension à évolution lente (SCD), les tables de version de prix des produits et les tables de détails des changements de statut.
Préparation
SET odps.sql.type.system.odps2=true;
SET odps.table.append2.enable=true;
SET odps.table.hash.delta.enable=true;
Créer une table
Table non partitionnée
CREATE TABLE hash_delta_sales_demo (
item_id BIGINT,
event_time TIMESTAMP,
price DOUBLE,
status STRING
)
CLUSTERED BY (item_id)
SORTED BY (event_time)
INTO 256 BUCKETS
TBLPROPERTIES ('table.format.version' = '2');
Table partitionnée
Si vous devez gérer les données par date, vous pouvez également créer une table partitionnée :
CREATE TABLE hash_delta_sales_demo_pt (
item_id BIGINT,
event_time TIMESTAMP,
price DOUBLE,
status STRING
)
PARTITIONED BY (ds STRING)
CLUSTERED BY (item_id)
SORTED BY (event_time)
INTO 256 BUCKETS
TBLPROPERTIES ('table.format.version' = '2');
Exécutez DESC EXTENDED hash_delta_sales_demo; pour afficher les informations de la table. Les définitions de regroupement et de tri de la table sont les suivantes :
ClusterType: hash
BucketNum: 256
ClusterColumns: [item_id]
SortColumns: [event_time ASC]
Écritures incrémentielles
Les exemples suivants utilisent une table non partitionnée pour illustrer les écritures incrémentielles et le reclustering.
-
Écriture initiale des données
INSERT INTO TABLE hash_delta_sales_demo VALUES
(1001, TIMESTAMP '2026-05-01 10:00:00', 10.00, 'active'),
(1001, TIMESTAMP '2026-05-03 10:00:00', 13.00, 'active'),
(1002, TIMESTAMP '2026-05-01 11:00:00', 20.00, 'active');
DESC EXTENDED hash_delta_sales_demo;
Résultat d'exécution exemple
Champs clés :
+------------------------------------------------------------------------------------+
| Owner: ALIYUN$***_com |
| Project: test |
| TableComment: |
+------------------------------------------------------------------------------------+
| CreateTime: 2026-07-08 16:38:34 |
| LastDDLTime: 2026-07-08 16:38:34 |
| LastModifiedTime: 2026-07-08 16:39:38 |
+------------------------------------------------------------------------------------+
| InternalTable: YES | Size: 4823 |
+------------------------------------------------------------------------------------+
| Native Columns: |
+------------------------------------------------------------------------------------+
| Field | Type | Label | ExtendedLabel | Nullable | DefaultValue | Comment |
+------------------------------------------------------------------------------------+
| item_id | bigint | | | true | NULL | |
| event_time | timestamp | | | true | NULL | |
| price | double | | | true | NULL | |
| status | string | | | true | NULL | |
+------------------------------------------------------------------------------------+
| Extended Info: |
+------------------------------------------------------------------------------------+
| TableID: 65**8e |
| IsArchived: false |
| PhysicalSize: 14469 |
| FileNum: 5 |
| ColdStorageStatus: N/A |
| CompressionStrategy: normal |
| DataFullySorted: true |
| DataPhysicalClustered: true |
| IsolationMin: NONSTRICT_SNAPSHOT_ISOLATION |
| OverlapDepth: 2 |
| OverlapRatio: 1.000000 |
| StoredAs: AliOrc |
| Transactional: true |
| encryption_enable: false |
| odps.timemachine.retention.days: 1 |
| ClusterType: hash |
| BucketNum: 256 |
| ClusterColumns: [item_id] |
| SortColumns: [event_time ASC] |
| StorageTier: Standard |
| StorageTierLastModifiedTime: 2026-07-08 16:39:38 |
+------------------------------------------------------------------------------------+
-
Opération de suppression
Vérifiez l'état de la table après avoir supprimé certaines données :
DELETE FROM hash_delta_sales_demo WHERE item_id = 1002;
DESC EXTENDED hash_delta_sales_demo;
Résultat d'exécution exemple
Champs clés :
+------------------------------------------------------------------------------------+
| Owner: ALIYUN$***_com |
| Project: test |
| TableComment: |
+------------------------------------------------------------------------------------+
| CreateTime: 2026-07-08 16:38:34 |
| LastDDLTime: 2026-07-08 16:38:34 |
| LastModifiedTime: 2026-07-08 16:41:02 |
| LastAccessTime: 2026-07-08 16:40:57 |
+------------------------------------------------------------------------------------+
| InternalTable: YES | Size: 7082 |
+------------------------------------------------------------------------------------+
| Native Columns: |
+------------------------------------------------------------------------------------+
| Field | Type | Label | ExtendedLabel | Nullable | DefaultValue | Comment |
+------------------------------------------------------------------------------------+
| item_id | bigint | | | true | NULL | |
| event_time | timestamp | | | true | NULL | |
| price | double | | | true | NULL | |
| status | string | | | true | NULL | |
+------------------------------------------------------------------------------------+
| Extended Info: |
+------------------------------------------------------------------------------------+
| TableID: 65**8e |
| IsArchived: false |
| PhysicalSize: 21246 |
| FileNum: 10 |
| ColdStorageStatus: N/A |
| CompressionStrategy: normal |
| DataFullySorted: true |
| DataPhysicalClustered: true |
| IsolationMin: NONSTRICT_SNAPSHOT_ISOLATION |
| OverlapDepth: 2 |
| OverlapRatio: 1.000000 |
| StoredAs: AliOrc |
| Transactional: true |
| encryption_enable: false |
| odps.timemachine.retention.days: 1 |
| ClusterType: hash |
| BucketNum: 256 |
| ClusterColumns: [item_id] |
| SortColumns: [event_time ASC] |
| StorageTier: Standard |
| StorageTierLastModifiedTime: 2026-07-08 16:41:02 |
+------------------------------------------------------------------------------------+
-
Rétrochargement des données historiques
Rétrochargez une version historique. L'event_time de cette version se situe entre les horodatages existants pour item_id=1001 :
INSERT INTO TABLE hash_delta_sales_demo VALUES
(1001, TIMESTAMP '2026-05-02 09:00:00', 12.00, 'active');
DESC EXTENDED hash_delta_sales_demo;
Résultat d'exécution exemple
Champs clés :
DataPhysicalClustered: true -- Les données sont physiquement clusterisées par item_id.
DataFullySorted: false -- Les données du bucket ne sont plus entièrement triées car de nouveaux fichiers ont été ajoutés.
+------------------------------------------------------------------------------------+
| Owner: ALIYUN$***_com |
| Project: test |
| TableComment: |
+------------------------------------------------------------------------------------+
| CreateTime: 2026-07-08 16:38:34 |
| LastDDLTime: 2026-07-08 16:38:34 |
| LastModifiedTime: 2026-07-08 16:42:54 |
| LastAccessTime: 2026-07-08 16:40:57 |
+------------------------------------------------------------------------------------+
| InternalTable: YES | Size: 10705 |
+------------------------------------------------------------------------------------+
| Native Columns: |
+------------------------------------------------------------------------------------+
| Field | Type | Label | ExtendedLabel | Nullable | DefaultValue | Comment |
+------------------------------------------------------------------------------------+
| item_id | bigint | | | true | NULL | |
| event_time | timestamp | | | true | NULL | |
| price | double | | | true | NULL | |
| status | string | | | true | NULL | |
+------------------------------------------------------------------------------------+
| Extended Info: |
+------------------------------------------------------------------------------------+
| TableID: 65**8e |
| IsArchived: false |
| PhysicalSize: 32115 |
| FileNum: 13 |
| ColdStorageStatus: N/A |
| CompressionStrategy: normal |
| DataFullySorted: false |
| DataPhysicalClustered: true |
| IsolationMin: NONSTRICT_SNAPSHOT_ISOLATION |
| OverlapDepth: 2 |
| OverlapRatio: 1.000000 |
| StoredAs: AliOrc |
| Transactional: true |
| encryption_enable: false |
| odps.timemachine.retention.days: 1 |
| ClusterType: hash |
| BucketNum: 256 |
| ClusterColumns: [item_id] |
| SortColumns: [event_time ASC] |
| StorageTier: Standard |
| StorageTierLastModifiedTime: 2026-07-08 16:42:54 |
+------------------------------------------------------------------------------------+
Élagage des buckets
Lorsque vous exécutez une requête avec un filtre d'égalité sur une colonne de bucketing, MaxCompute utilise la distribution de hachage pour localiser directement le bucket cible, ce qui évite l'analyse de tous les autres buckets. La requête suivante applique un filtre sur item_id = 1001 et lit uniquement le bucket logique contenant cette valeur, évitant ainsi une analyse complète de la table :
SELECT * FROM hash_delta_sales_demo
WHERE item_id = 1001
ORDER BY event_time
LIMIT 10;
-- Returns:
+------------+---------------------+------------+--------+
| item_id | event_time | price | status |
+------------+---------------------+------------+--------+
| 1001 | 2026-05-01 10:00:00 | 10.0 | active |
| 1001 | 2026-05-02 09:00:00 | 12.0 | active |
| 1001 | 2026-05-03 10:00:00 | 13.0 | active |
+------------+---------------------+------------+--------+
Reclustering complet
Si vous devez réorganiser les données existantes, exécutez RECLUSTER FULL. Cette opération préserve la sémantique des données historiques de la table et réorganise les données stockées selon la définition actuelle de la table.
ALTER TABLE hash_delta_sales_demo RECLUSTER FULL;
DESC EXTENDED hash_delta_sales_demo;
Résultat d'exécution exemple
Champs clés :
+------------------------------------------------------------------------------------+
| Owner: ALIYUN$***_com |
| Project: test |
| TableComment: |
+------------------------------------------------------------------------------------+
| CreateTime: 2026-07-08 16:38:34 |
| LastDDLTime: 2026-07-08 16:38:34 |
| LastModifiedTime: 2026-07-08 16:42:54 |
| LastAccessTime: 2026-07-08 16:40:57 |
+------------------------------------------------------------------------------------+
| InternalTable: YES | Size: 20218 |
+------------------------------------------------------------------------------------+
| Native Columns: |
+------------------------------------------------------------------------------------+
| Field | Type | Label | ExtendedLabel | Nullable | DefaultValue | Comment |
+------------------------------------------------------------------------------------+
| item_id | bigint | | | true | NULL | |
| event_time | timestamp | | | true | NULL | |
| price | double | | | true | NULL | |
| status | string | | | true | NULL | |
+------------------------------------------------------------------------------------+
| Extended Info: |
+------------------------------------------------------------------------------------+
| TableID: 65**8e |
| IsArchived: false |
| PhysicalSize: 60654 |
| FileNum: 20 |
| ColdStorageStatus: N/A |
| CompressionStrategy: normal |
| DataFullySorted: true |
| DataPhysicalClustered: true |
| IsolationMin: NONSTRICT_SNAPSHOT_ISOLATION |
| OverlapDepth: 2 |
| OverlapRatio: 1.000000 |
| StoredAs: AliOrc |
| Transactional: true |
| encryption_enable: false |
| odps.timemachine.retention.days: 1 |
| ClusterType: hash |
| BucketNum: 256 |
| ClusterColumns: [item_id] |
| SortColumns: [event_time ASC] |
| StorageTier: Standard |
| StorageTierLastModifiedTime: 2026-07-08 16:42:54 |
+------------------------------------------------------------------------------------+
Une Append Delta Table avec Hash Cluster prend en charge les opérations d'écriture incrémentielles telles que INSERT, UPDATE, DELETE et MERGE INTO, tout en préservant la distribution de hachage. L'optimiseur choisit un plan d'exécution en fonction de l'état actuel des données. Il exploite le stockage trié lorsque les données sont ordonnées et revient à l'utilisation du bucketing par hachage lorsque les données ne sont pas entièrement triées. Vous pouvez exécuter RECLUSTER FULL à tout moment pour restaurer l'ordre de tri complet.
Exemple avec le SDK Data Tunnel
Cette section montre comment utiliser le SDK Data Tunnel pour charger et télécharger des données depuis la table hash_delta_sales_demo.
-
Importez la dépendance du SDK
Utilisez la version 0.59 ou ultérieure. Pour plus de détails, consultez les Notes de version.
-
Exemple de code
Exemple de code
/**
* This example demonstrates how to upload and download data from the hash_delta_sales_demo table
* by using MaxStorageClient.
*
* Table Schema:
* CREATE TABLE hash_delta_sales_demo (
* item_id BIGINT,
* event_time TIMESTAMP,
* price DOUBLE,
* status STRING
* );
*/
public class MaxStorageClientExample {
private static final String ENDPOINT = "<your-endpoint>";
private static final String TUNNEL_ENDPOINT = "<your-tunnel-endpoint>";
private static final String PROJECT = "<your-project>";
private static final String ACCESS_ID = "<your-access-id>";
private static final String ACCESS_KEY = "<your-access-key>";
private static final String TABLE_NAME = "hash_delta_sales_demo";
public static void main(String[] args) throws Exception {
RootAllocator allocator = new RootAllocator(Long.MAX_VALUE);
// 1. Build the MaxStorageClient.
MaxStorageClient client = MaxStorageClient.builder()
.endpoint(ENDPOINT)
.tunnelEndpoint(TUNNEL_ENDPOINT)
.credentialsProvider(
new StaticCredentialProvider(new AliyunAccount(ACCESS_ID, ACCESS_KEY).getCredentials()))
.project(PROJECT)
.bufferAllocator(allocator)
.build();
try {
// 2. Upload data.
uploadData(client);
Thread.sleep(5000);
// 3. Download data.
downloadData(client);
} finally {
client.close();
allocator.close();
}
}
/**
* Upload data to the hash_delta_sales_demo table.
*/
private static void uploadData(MaxStorageClient client) throws Exception {
TableIdentifier tableId = TableIdentifier.of(PROJECT, TABLE_NAME);
// Create a write session. withOverwrite(true) indicates that the table will be overwritten.
TableWriteSession writeSession = client
.createTableWriteSessionBuilder(tableId)
.withOverwrite(true)
.build();
System.out.println("Write session created: " + writeSession.getId());
// Use RecordWriter to write data (a high-level API for row-based writing).
try (RecordWriter writer = writeSession.createWriterBuilder("stream-1", 1)
.build()
.getAsRecordWriter(1024)) {
for (int i = 0; i < 1000; i++) {
Record record = writer.newRecord(false);
record.set(0, (long) i); // item_id: BIGINT
record.set(1, LocalDateTime.of(2025, 5, 18, 10, 30, i % 60).atZone(ZoneId.systemDefault())
.toInstant()); // event_time: TIMESTAMP
record.set(2, 99.9 + i * 0.1); // price: DOUBLE
record.set(3, i % 2 == 0 ? "paid" : "pending"); // status: STRING
writer.write(record);
}
}
// Commit the session to make the data visible.
writeSession.commit();
System.out.println("Successfully uploaded 1000 records to " + TABLE_NAME);
}
/**
* Download data from the hash_delta_sales_demo table.
*/
private static void downloadData(MaxStorageClient client) throws Exception {
TableIdentifier tableId = TableIdentifier.of(PROJECT, TABLE_NAME);
// Create a read session. You can select specific columns and apply filters.
TableReadSession readSession = client.createTableReadSessionBuilder(tableId)
.withColumns(Arrays.asList("item_id", "event_time", "price", "status"))
.withSplitOptions(SplitOptions.newBuilder()
.withSplitMode(SplitMode.ROW_OFFSET)
.build())
.build();
System.out.println("Read session created: " + readSession.getId());
// Get the input splits.
List<InputSplit> splits = readSession.getSplits();
System.out.println("Total splits: " + splits.size());
int totalRecords = 0;
// Iterate through each split to read data.
for (InputSplit split : splits) {
try (ArrowReader reader = readSession.createReaderBuilder(split).build()) {
Schema schema = reader.getSchema();
System.out.println("Schema: " + schema);
while (reader.nextBatch()) {
VectorSchemaRoot root = reader.getCurrentValue();
int rowCount = root.getRowCount();
totalRecords += rowCount;
// Print the first 5 rows as an example.
int printCount = Math.min(rowCount, 5);
for (int i = 0; i < printCount; i++) {
System.out.printf(" item_id=%s, event_time=%s, price=%s, status=%s%n",
root.getVector("item_id").getObject(i),
root.getVector("event_time").getObject(i),
root.getVector("price").getObject(i),
root.getVector("status").getObject(i));
}
if (rowCount > 5) {
System.out.println(" ... (" + (rowCount - 5) + " more rows in this batch)");
}
}
}
}
System.out.println("Total records downloaded: " + totalRecords);
}
}
FAQ
Choix de la taille de stockage d'un bucket
La taille de stockage recommandée pour un seul bucket se situe entre plusieurs centaines de mégaoctets et quelques dizaines de gigaoctets.
Des buckets de petite taille augmentent les frais de stockage et les coûts de shuffle.
Des buckets de grande taille allongent les temps d'écriture et réduisent l'efficacité de l'élagage des buckets et des optimisations de shuffle.
Définissez le nombre de buckets en fonction de la croissance prévue des données, et non pas uniquement du volume actuel, afin d'éviter des modifications fréquentes de la structure de la table.
Si votre volume de données est exceptionnellement important, un seul bucket peut prendre en charge un stockage plus élevé, ou vous pouvez définir un nombre de buckets plus élevé. Toutefois, vous devez évaluer l'impact sur les performances d'écriture et de requête en fonction de votre cas d'utilisation spécifique.