Écrire et consommer des données de table Paimon - Realtime Compute for Apache Flink

Prérequis

Créez d'abord un catalogue Paimon et une table Paimon. Pour plus d'informations, consultez la page Gérer les catalogues Paimon.

Limitations

Les tables Paimon sont prises en charge uniquement dans Ververica Runtime (VVR) 8.0.5 et versions ultérieures.

Écrire des données dans une table Paimon

Synchroniser les modifications de données et de schéma avec l'ingestion de données Flink CDC

Pour plus d'informations, consultez le connecteur Ingestion de données Paimon.

Insérer ou mettre à jour des données avec INSERT INTO

Utilisez une instruction INSERT INTO pour insérer ou mettre à jour directement des données dans une table Paimon.

Une table de clé primaire Paimon accepte tous les types de messages, notamment INSERT, UPDATE_BEFORE, UPDATE_AFTER et DELETE. Lors de l'écriture, elle fusionne les données partageant la même clé primaire, selon le mécanisme de fusion des données.
Une table append-only Paimon, également appelée table sans clé primaire, n'accepte que les messages de type INSERT.

Écraser des données avec INSERT OVERWRITE

L'écrasement efface les données existantes et écrit de nouvelles données. Utilisez une instruction INSERT OVERWRITE pour écraser une table Paimon entière ou une partition spécifique. Les exemples de code suivants illustrent cette opération.

Remarque

L'instruction INSERT OVERWRITE est prise en charge uniquement dans les jobs par lot.
Par défaut, une opération INSERT OVERWRITE ne génère pas de données de journal des modifications (changelog). Les jobs en flux continu en aval ne peuvent pas consommer les données supprimées et importées. Si vous devez consommer ce type de données, consultez la section Consommer en flux continu les résultats d'une instruction INSERT OVERWRITE.

Écrasez la table non partitionnée entière my_table.
```
INSERT OVERWRITE my_table SELECT ...;
```

Écrasez la partition dt=20240108,hh=06 de la table my_table.

INSERT OVERWRITE my_table PARTITION (`dt` = '20240108', `hh` = '06') SELECT ...;

Écrasez dynamiquement les partitions de la table my_table. Seules les partitions présentes dans le résultat de l'instruction SELECT sont écrasées, tandis que les autres restent inchangées.
```
INSERT OVERWRITE my_table SELECT ...;
```

Écrasez la table partitionnée entière my_table.

INSERT OVERWRITE my_table /*+ OPTIONS('dynamic-partition-overwrite' = 'false') */ SELECT ...;

Supprimer des données à l'aide de DELETE

Utilisez une instruction DELETE pour supprimer des données d'une table de clé primaire Paimon. Vous ne pouvez exécuter des instructions DELETE que dans l'outil Data Exploration.

-- Delete all data where currency = 'UNKNOWN' from the my_table table.
DELETE FROM my_table WHERE currency = 'UNKNOWN';

Filtrer les messages de suppression

Lorsque vous utilisez une table de clé primaire Paimon, les messages DELETE suppriment par défaut les données associées à la clé primaire correspondante. Pour empêcher la table Paimon de traiter ces messages, utilisez un indice SQL afin de définir le paramètre suivant sur true. Cela permet de filtrer les messages de suppression.

Paramètre

Description

Type

Valeur par défaut

ignore-delete

Indique s'il faut filtrer les messages de suppression.

Booléen

false

Ajuster le parallélisme du puits (sink)

Utilisez un indice SQL pour définir le paramètre suivant afin d'ajuster le parallélisme de l'opérateur de puits (sink).

Paramètre

Description

Type

Valeur par défaut

sink.parallelism

Définit le parallélisme de l'opérateur de puits Paimon.

Entier

Aucune

Par exemple, l'instruction SQL suivante définit le parallélisme de l'opérateur de puits Paimon sur 10.

INSERT INTO t /*+ OPTIONS('sink.parallelism' = '10') */ SELECT * FROM s;

Consommer des données d'une table Paimon

Jobs en flux continu

Remarque

Pour les tables de clé primaire Paimon consommées par des jobs en flux continu, configurez un producteur de journal des modifications (changelog).

Par défaut, un opérateur source Paimon dans un job en flux continu produit d'abord l'intégralité des données de la table au démarrage du job, puis les données incrémentielles à partir de ce point.

Consommer des données à partir d'un décalage spécifié

Consommez des données à partir d'un décalage spécifié de l'une des manières suivantes :

Si vous n'avez pas besoin de consommer l'intégralité des données d'une table Paimon au démarrage du job et souhaitez uniquement consommer les données incrémentielles ultérieures, définissez 'scan.mode' = 'latest' via un indice SQL.
```
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'latest') */;
```
Si vous ne souhaitez pas consommer l'intégralité des données, mais uniquement les données incrémentielles à partir d'un moment précis, utilisez un indice SQL pour définir le paramètre scan.timestamp-millis. La valeur du paramètre représente le nombre de millisecondes écoulées depuis l'époque Unix (1970-01-01 00:00:00 UTC) jusqu'au moment spécifié.
```
SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;
```
Pour consommer les données écrites après un moment donné, puis continuer à consommer les données incrémentielles ultérieures, utilisez l'une des méthodes suivantes.

Remarque
Cette méthode de consommation lit les fichiers de données modifiés après l'heure spécifiée. En raison de la compaction, les fichiers de données peuvent contenir une petite quantité de données écrites avant l'heure spécifiée. Ajoutez une clause WHERE à votre job SQL pour filtrer les données selon vos besoins.
- Ne définissez aucun indice SQL. Au démarrage du job, sélectionnez Specify source's start time. Dans la boîte de dialogue Job Start, sélectionnez Stateless start-up, activez l'option Specify source's start time et définissez l'heure cible.
- Définissez le paramètre scan.file-creation-time-millis à l'aide d'un indice SQL.
```
SELECT * FROM t /*+ OPTIONS('scan.file-creation-time-millis' = '1678883047356') */;
```
Si vous ne souhaitez pas consommer l'intégralité des données, mais uniquement les données incrémentielles à partir d'un fichier d'instantané spécifique, utilisez un indice SQL pour définir le paramètre scan.snapshot-id. La valeur de ce paramètre correspond à l'ID du fichier d'instantané spécifié.
```
SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;
```
Si vous souhaitez consommer l'intégralité des données d'un fichier d'instantané spécifique, puis continuer à consommer les données incrémentielles, utilisez un indice SQL pour définir les paramètres 'scan.mode' = 'from-snapshot-full' et scan.snapshot-id. La valeur du paramètre scan.snapshot-id correspond à l'ID du fichier d'instantané spécifié.
```
SELECT * FROM t /*+ OPTIONS('scan.mode' = 'from-snapshot-full', 'scan.snapshot-id' = '1') */;
```

Spécifier un ID de consommateur

Un ID de consommateur enregistre la progression de la consommation pour une table Paimon. Cette fonctionnalité est principalement utilisée dans les scénarios suivants :

La définition d'un ID de consommateur enregistre la progression de consommation correspondante dans le fichier de métadonnées de la table Paimon. Cela permet au job de reprendre la consommation à partir du point d'interruption, même s'il est démarré ultérieurement en mode sans état.
Après avoir défini un ID de consommateur, les instantanés qui n'ont pas été consommés ne sont pas supprimés lors de l'expiration. Cela évite les erreurs causées par une vitesse de consommation insuffisante par rapport à la vitesse d'expiration des instantanés.

Définissez le paramètre consumer-id pour attribuer un ID de consommateur à un opérateur source Paimon dans un job en flux continu. La valeur de l'ID de consommateur peut être n'importe quelle chaîne de caractères. Lors de la première création d'un ID de consommateur, son décalage de départ est déterminé selon les règles décrites dans la section Consommer une table Paimon à partir d'un décalage spécifié. Vous pouvez ensuite reprendre la consommation de la table Paimon en continuant à utiliser le même ID de consommateur.

Par exemple, l'instruction SQL suivante définit un ID de consommateur nommé test-id pour un opérateur source Paimon. Si vous souhaitez réinitialiser le décalage pour un ID de consommateur spécifique, vous pouvez également définir 'consumer.ignore-progress' = 'true'.

SELECT * FROM t /*+ OPTIONS('consumer-id' = 'test-id') */;

Remarque

Les fichiers d'instantané qui n'ont pas été consommés par un ID de consommateur ne sont pas supprimés lors de l'expiration. Si les ID de consommateur obsolètes ne sont pas nettoyés, les fichiers d'instantané correspondants et les fichiers de données historiques ne seront jamais supprimés et occuperont de l'espace de stockage. Définissez le paramètre de table consumer.expiration-time pour nettoyer les ID de consommateur inutilisés pendant une période donnée. Par exemple, 'consumer.expiration-time' = '3d' indique que les ID de consommateur inutilisés pendant 3 jours seront nettoyés.

Consommer en flux continu les résultats de INSERT OVERWRITE

Par défaut, les opérations INSERT OVERWRITE ne génèrent pas de données de journal des modifications (changelog), et les données supprimées et importées ne peuvent pas être consommées par les jobs en flux continu en aval. Si vous devez consommer ces données, définissez 'streaming-read-overwrite' = 'true' dans un job de consommation en flux continu à l'aide d'un indice SQL.

SELECT * FROM t /*+ OPTIONS('streaming-read-overwrite' = 'true') */;

Jobs par lot

Par défaut, un opérateur source Paimon dans un job par lot lit le dernier instantané et produit les données d'état les plus récentes de la table Paimon.

Voyage dans le temps par lot

Interrogez l'état d'une table Paimon à un moment précis en définissant le paramètre scan.timestamp-millis dans un indice SQL. La valeur du paramètre représente le nombre de millisecondes écoulées depuis l'époque Unix (1970-01-01 00:00:00 UTC) jusqu'au moment spécifié.

SELECT * FROM t /*+ OPTIONS('scan.timestamp-millis' = '1678883047356') */;

Interrogez l'état d'une table Paimon au moment de la création d'un instantané en utilisant un indice SQL pour définir le paramètre scan.snapshot-id sur l'ID de l'instantané spécifié.

SELECT * FROM t /*+ OPTIONS('scan.snapshot-id' = '3') */;

Interroger les modifications entre les instantanés

Pour interroger les modifications de données dans une table Paimon entre deux instantanés, définissez le paramètre incremental-between à l'aide d'un indice SQL. Par exemple, pour afficher toutes les données ayant changé entre l'instantané 20 et l'instantané 12, l'instruction SQL est la suivante.

SELECT * FROM t /*+ OPTIONS('incremental-between' = '12,20') */;

Remarque

Étant donné que les jobs par lot ne prennent pas en charge la consommation des messages DELETE, ces messages sont ignorés par défaut. Si vous souhaitez consommer des messages DELETE dans un job par lot, interrogez la table système Audit Log. Par exemple : SELECT * FROM .

Ajuster le parallélisme de la source

Par défaut, Paimon déduit automatiquement le parallélisme de l'opérateur source en fonction d'informations telles que le nombre de partitions et de buckets. Utilisez un indice SQL pour définir les paramètres suivants afin d'ajuster le parallélisme.

Paramètre

Type

Valeur par défaut

Description

scan.parallelism

Entier

Aucune

Définit le parallélisme de l'opérateur source Paimon.

scan.infer-parallelism

Booléen

true

Indique s'il faut déduire automatiquement le parallélisme de l'opérateur source Paimon.

scan.infer-parallelism.max

Entier

1024

Limite supérieure du parallélisme déduit automatiquement pour l'opérateur source Paimon.

L'instruction SQL suivante montre comment définir le parallélisme de l'opérateur source Paimon sur 10.

SELECT * FROM t /*+ OPTIONS('scan.parallelism' = '10') */;

Utiliser les tables Paimon comme tables de dimension

Les tables Paimon peuvent également être utilisées comme tables de dimension. Pour plus d'informations sur la syntaxe des jointures de tables de dimension, consultez la page Instruction JOIN de table de dimension.

Écrire et consommer le type VARIANT

Dans Ververica Runtime (VVR) 11.1 et versions ultérieures, les tables Paimon prennent en charge le type de données semi-structurées VARIANT. Ce type vous permet de convertir les chaînes JSON VARCHAR vers le type VARIANT à l'aide des fonctions PARSE_JSON ou TRY_PARSE_JSON. L'écriture et la consommation directes du type VARIANT améliorent considérablement les performances des requêtes et du traitement JSON.

Le code suivant fournit un exemple :

CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  k BIGINT,
  info VARIANT
);
INSERT INTO `my-catalog`.`my_db`.`my_tbl` 
SELECT k, PARSE_JSON(jsonStr) FROM T;

Documentation connexe

Lorsque vous écrivez et consommez des données dans les tables Paimon, vous pouvez utiliser des indices SQL pour modifier temporairement les paramètres de table. Pour plus d'informations, consultez la page Gérer les tables Paimon.
Pour plus d'informations sur les fonctionnalités de base des tables de clé primaire Paimon et des tables append-only Paimon, consultez la page Tables de clé primaire et append-only Paimon.
Pour plus d'informations sur les optimisations courantes des tables de clé primaire Paimon et des tables Append Scalable dans différents scénarios, consultez la page Optimisation des performances Paimon.
La consommation de données à partir des tables Paimon repose sur les fichiers d'instantané. Si le délai d'expiration des instantanés est trop court ou si un job de consommation est inefficace, le fichier d'instantané en cours de consommation peut être supprimé après expiration, ce qui entraîne l'erreur File xxx not found, Possible causes dans le job de consommation. Pour résoudre ce problème, consultez la page Résoudre l'erreur "File xxx not found, Possible causes" dans les jobs de lecture Paimon.