MaxCompute:Architecture d'intégration de données complètes et incrémentielles en quasi-temps réel avec les tables Delta

Création de table

MaxCompute introduit les tables Delta avec un format de données de table unifié pour prendre en charge l'architecture intégrée. Les tables Delta prennent en charge toutes les fonctionnalités des workflows de traitement par lots existants et des nouveaux workflows tels que le stockage et le traitement des données incrémentielles en quasi-temps réel.

Pour créer une table Delta, spécifiez les clés primaires et définissez "transactional"="true" dans l'instruction CREATE TABLE :

CREATE TABLE tt2 (pk BIGINT NOT NULL PRIMARY KEY, val STRING) tblproperties ("transactional"="true");
CREATE TABLE par_tt2 (pk BIGINT NOT NULL PRIMARY KEY, val STRING)  PARTITIONED BY (pt STRING) tblproperties ("transactional"="true");

Les clés primaires garantissent l'unicité des lignes. La propriété transactional active le mécanisme de transaction ACID (atomicité, cohérence, isolation et durabilité) avec isolation par snapshot pour les opérations de lecture et d'écriture. Pour plus de détails, consultez la section Opérations sur les tables.

Paramètres clés pour les tables Delta

Pour la référence complète des paramètres, consultez la section « Paramètres des tables Delta » dans Opérations sur les tables.

write.bucket.num

Spécifie le nombre de buckets par table partitionnée ou non partitionnée, ainsi que le nombre de nœuds d'écriture concurrents. La valeur par défaut est 16. Valeurs valides : (0, 4096].

• Pour les tables partitionnées : la valeur peut être modifiée et s'applique automatiquement aux nouvelles partitions.

• Pour les tables non partitionnées : la valeur ne peut pas être modifiée après la création.

acid.data.retain.hours

Spécifie la plage de temps des données historiques disponibles pour les requêtes de voyage dans le temps. La valeur par défaut est 24. Valeurs valides : [0, 168] (unité : heures).

• Définissez sur 0 pour désactiver les requêtes de voyage dans le temps et réduire considérablement les coûts de stockage des données historiques.

• Pour les données historiques datant de plus de 168 heures (7 jours), contactez le support technique MaxCompute.

Évolution du schéma

Les tables Delta prennent en charge l'évolution complète du schéma, y compris l'ajout et la suppression de colonnes. Lors de l'interrogation de données historiques avec le voyage dans le temps, le système lit les données selon le schéma de cette version historique.

L'exemple suivant ajoute une colonne :

ALTER TABLE tt2 ADD columns (val2 string);

Pour la syntaxe DDL complète, consultez la section Opérations sur les tables.

Formats de données de table

La figure ci-dessus illustre la structure de données d'une table partitionnée. Les fichiers de données sont physiquement isolés par partition (stockés dans des répertoires distincts) et les données de chaque partition sont divisées en buckets. Les tables Delta utilisent deux types de fichiers de données :

Gouvernance et optimisation automatiques des données

Problème : prolifération des petits fichiers

Les tables Delta prennent en charge l'importation de données incrémentielles en quasi-temps réel en quelques minutes. Dans les scénarios d'écriture à fort trafic avec de nombreux buckets, le nombre de petits fichiers de données incrémentielles peut augmenter rapidement, entraînant un nombre excessif de demandes d'accès, des coûts élevés et une faible efficacité des E/S. Les opérations UPDATE et DELETE intensives aggravent encore ce problème en générant un grand nombre d'enregistrements historiques intermédiaires redondants.

Solution : quatre services de gouvernance automatisés

Le moteur de stockage MaxCompute gouverne et optimise automatiquement les données stockées sans configuration manuelle. Le moteur de stockage identifie intelligemment les caractéristiques des données selon plusieurs dimensions et applique automatiquement les politiques appropriées.

Auto partial compact efface uniquement les enregistrements historiques dont l'heure de création dépasse la fenêtre de rétention du voyage dans le temps.

Pour les scénarios nécessitant des performances de requête optimales, déclenchez manuellement une compaction majeure :

SET odps.merge.task.mode=service;
ALTER TABLE tt2 compact major;

Pour plus d'informations, consultez la section COMPACTION.

Upserts en quasi-temps réel en quelques minutes

Pourquoi les tables Delta : Le traitement par lots traditionnel importe les données incrémentielles vers une nouvelle table ou partition sur plusieurs heures ou jours, puis déclenche un processus ETL hors ligne pour joindre et fusionner ces données incrémentielles avec les données de table existantes. Cette approche présente une latence élevée ainsi que des coûts de ressources et de stockage importants.

Avec les tables Delta, le pipeline d'upsert maintient une latence de 5 à 10 minutes entre l'écriture des données et leur interrogation. Aucun processus complexe de fusion ETL n'est requis, ce qui réduit à la fois les coûts de calcul et de stockage.

Une variété de sources de données est courante en production : bases de données, systèmes de journaux, files d'attente de messages. MaxCompute fournit un plug-in connecteur Flink open source qui fonctionne avec DataWorks Data Integration et d'autres outils d'importation de données. Il prend en charge la conception personnalisée et l'optimisation du développement pour les scénarios de haute concurrence, de tolérance aux pannes et de soumission de transactions.

Capacités clés de l'intégration du connecteur Flink :

Le débit du trafic varie selon l'environnement et la configuration. Estimez le débit maximal en fonction de la capacité de traitement d'un seul bucket (1 Mo/s). Le groupe de ressources Tunnel partagé est utilisé par défaut pour MaxCompute Tunnel, ce qui peut entraîner un débit instable en cas de forte contention des ressources. Des limites sont également imposées sur la consommation de ressources.

Synchronisation de données en temps réel depuis des bases de données à l'aide de DataWorks Data Integration

De nombreux systèmes de production combinent le traitement transactionnel en ligne (OLTP), le traitement analytique en ligne (OLAP) et des moteurs d'analyse hors ligne. Un workflow courant consiste à synchroniser les nouveaux enregistrements d'une table unique ou d'une base de données entière vers MaxCompute en temps réel pour analyse.

La figure ci-dessus contraste deux approches :

• Gauche (traitement par lots) : Les données incrémentielles sont importées vers une nouvelle table ou partition sur plusieurs heures ou jours. Un processus ETL hors ligne joint ensuite et fusionne les données incrémentielles avec les données de table existantes. Cette approche présente une latence élevée ainsi que des coûts de ressources et de stockage importants.

• Droite (architecture intégrée) : Les nouveaux enregistrements sont lus depuis les bases de données en quelques minutes. Aucune extraction ou fusion périodique des données n'est nécessaire ; les tables Delta gèrent directement les mises à jour, minimisant ainsi les coûts de calcul et de stockage.

Traitement par lots à l'aide d'instructions SQL DML et d'upserts

Les modules Compiler, Optimizer et Runtime du moteur SQL sont modifiés et optimisés pour les opérations sur les tables Delta. Cela inclut l'analyse syntaxique, les plans d'optimisation, la logique de déduplication basée sur la clé primaire et les upserts au moment de l'exécution pour fournir une prise en charge complète de la syntaxe SQL.

Comportements clés :

• Cohérence des transactions : Une fois le traitement des données terminé, le service de métadonnées effectue la détection des conflits de transactions et les mises à jour atomiques des métadonnées, garantissant l'isolation en lecture/écriture et la cohérence des transactions.

• Upserts simplifiés : Le système fusionne automatiquement les enregistrements basés sur les clés primaires lors des requêtes sur les tables Delta. Pour les scénarios mélangeant les opérations INSERT et UPDATE, utilisez INSERT INTO au lieu de la syntaxe complexe UPDATE ou MERGE INTO ; cela réduit les E/S de lecture et économise les ressources de calcul.

Pour la syntaxe DML SQL complète, consultez la section Opérations DML.

Requêtes de voyage dans le temps (Time travel)

Les requêtes de voyage dans le temps vous permettent d'interroger les versions historiques d'une table Delta. Les cas d'utilisation courants incluent :

• Récupération de données : Restaurez les données vers une version historique spécifiée après une modification ou une suppression accidentelle.

• Retour en arrière historique : Auditez ou réanalysez les données métier à partir d'un point dans le passé.

Exemples de requêtes :

-- Query historical data at a specific timestamp.
SELECT * FROM tt2 TIMESTAMP AS OF '2024-04-01 01:00:00';
-- Query historical data from 5 minutes before the current time.
SELECT * FROM tt2 TIMESTAMP AS OF CURRENT_TIMESTAMP() - 300;
-- Query historical data from the second-to-last commit.
SELECT * FROM tt2 TIMESTAMP AS OF GET_LATEST_TIMESTAMP('tt2', 2);

La figure suivante montre le fonctionnement d'une requête de voyage dans le temps :

L'exemple utilise une table transactionnelle nommée src :

• Côté gauche (processus de mise à jour des données) : Les transactions t1 à t5 génèrent chacune un fichier de données delta. COMPACTION s'exécute aux étapes t2 et t4, produisant les fichiers compactés c1 et c2. Dans c1, l'enregistrement historique intermédiaire (2,a) est supprimé et le dernier enregistrement (2,b) est conservé.

• Résolution de requête : Pour interroger les données historiques à t1, le système lit uniquement le fichier delta d1. Pour interroger à t2, il lit le fichier compacté c1 et renvoie trois enregistrements. Pour interroger à t3, il lit c1 et le fichier delta d3, puis les fusionne pour la sortie. Une fréquence plus élevée de COMPACTION accélère les requêtes mais augmente la surcharge opérationnelle ; choisissez une politique de déclenchement adaptée à vos besoins.

La syntaxe SQL prend en charge les constantes, les fonctions courantes et les clauses TIMESTAMP AS OF expr et VERSION AS OF expr pour des requêtes historiques précises. Pour plus de détails, consultez la section Requêtes de voyage dans le temps.

Requêtes incrémentielles

MaxCompute conçoit et développe une nouvelle syntaxe SQL de requête incrémentielle pour optimiser les requêtes et le calcul incrémentiels pour les tables Delta. Après soumission d'une instruction SQL de requête incrémentielle, le moteur MaxCompute analyse les versions historiques des données incrémentielles à interroger, récupère les fichiers de données compactés pertinents, fusionne les données des fichiers et renvoie le résultat.

La figure suivante montre le processus de requête incrémentielle :

L'exemple utilise la même table transactionnelle src avec les transactions t1 à t5 et les fichiers compactés c1 (à t2) et c2 (à t4) :

• Si begin est t1-1 et end est t1, le système lit uniquement le fichier delta d1 à t1.

• Si end est t2, le système lit les fichiers delta d1 et d2.

• Si begin est t1 et end est t2-1, la plage de requête s'étend de t1 à t2. Aucune donnée incrémentielle n'existe dans cette plage, donc des lignes vides sont renvoyées.

Les données des fichiers compactés c1 et c2 générés par COMPACTION ne sont pas considérées comme de nouvelles données pour la sortie des requêtes incrémentielles.

Pour la syntaxe des requêtes incrémentielles et les limites des paramètres, consultez la section « Paramètres et limites des requêtes incrémentielles » dans Requêtes de voyage dans le temps et requêtes incrémentielles.

Optimisation du saut de données basé sur la clé primaire

La distribution des données et les index des tables Delta sont construits sur les valeurs des colonnes de clé primaire. Lorsque vous interrogez par clé primaire, le système filtre à plusieurs niveaux pour réduire considérablement la quantité de données lues, améliorant l'efficacité des requêtes de centaines à milliers de fois.

Exemple : Une table Delta contient 100 millions d'enregistrements. Le filtrage par une seule valeur de clé primaire peut nécessiter la lecture de seulement 10 000 enregistrements.

Le processus de filtrage à trois niveaux :

1. Élagage des buckets (Bucket pruning) : Localise le bucket contenant la clé primaire cible, éliminant ainsi l'analyse de tous les autres buckets.

2. Élagage des fichiers de données : Au sein du bucket cible, identifie uniquement les fichiers de données contenant la valeur de la clé primaire.

3. Filtrage de plage au niveau des blocs : Applique un filtrage précis basé sur la distribution des valeurs de clé primaire au sein des blocs de fichiers, extrayant uniquement les blocs contenant la valeur cible.

Optimisation des plans de requête et d'analyse SQL

Chaque bucket d'une table Delta stocke des données uniques et triées par valeur de clé primaire. L'optimiseur SQL exploite ces propriétés pour éliminer les opérations coûteuses :

Après élimination de DISTINCT, du tri et du brassage global, les performances des requêtes s'améliorent de 100 %.