Cette rubrique explique comment utiliser le connecteur Paimon pour un data lakehouse en streaming. Pour des résultats optimaux, nous vous recommandons d'utiliser le connecteur avec le catalogue Paimon.
Informations générales
Apache Paimon est un format de stockage de data lake unifié pour les données en streaming et par lots, qui prend en charge les écritures à haut débit et les requêtes à faible latence. Paimon s'intègre parfaitement aux moteurs de calcul populaires disponibles dans Alibaba Cloud E-MapReduce, tels que Flink, Spark, Hive et Trino. Vous pouvez utiliser Apache Paimon pour créer rapidement un data lake sur HDFS ou OSS et le connecter à ces moteurs de calcul pour l'analytique des data lakes. Pour plus d'informations, consultez Apache Paimon.
|
Catégorie |
Description |
|
Types pris en charge |
Table source, table de dimension, table de résultat et cible pour l'ingestion de données |
|
Mode d'exécution |
Mode streaming et mode batch |
|
Format de données |
Non pris en charge |
|
Métriques de surveillance |
Aucune |
|
Types d'API |
SQL et YAML pour l'ingestion de données |
|
Mises à jour et suppressions sur les tables de résultat |
Oui |
Fonctionnalités clés
Apache Paimon offre les fonctionnalités principales suivantes :
Créez un data lake léger et peu coûteux sur HDFS ou sur un stockage objet.
Lisez et écrivez des jeux de données à grande échelle en modes streaming et batch.
Exécutez des requêtes batch et OLAP avec une fraîcheur des données allant de quelques minutes à quelques secondes.
Ingérez et produisez des données incrémentielles, servant de couche de stockage pour les entrepôts de données traditionnels hors ligne et les entrepôts de données en streaming modernes.
Pré-agrégez les données pour réduire les coûts de stockage et la charge de calcul en aval.
Accédez aux versions historiques des données.
Filtrez les données efficacement.
Activez l'évolution du schéma.
Limites et recommandations
Le connecteur Paimon nécessite le moteur de calcul Flink VVR version 6.0.6 ou ultérieure.
-
Le tableau suivant répertorie la compatibilité des versions entre Paimon et VVR.
Version d'Apache Paimon
VVR
1.3.1
11,5, 11,6, 11,7, 11,8
1,3
11,4
1,2
11,2, 11,3
1,1
11,1
1,0
8.0.11
-
Recommandations de stockage pour les écritures concurrentes
Lorsque plusieurs jobs effectuent des écritures concurrentes sur la même table Paimon, l'utilisation du stockage OSS standard (oss://) peut occasionnellement provoquer des conflits de validation ou des échecs de job en raison des limitations liées aux opérations atomiques sur les fichiers.
Pour garantir la stabilité et la cohérence des écritures, utilisez un service de métadonnées ou de stockage offrant de fortes garanties atomiques. L'option privilégiée est Data Lake Formation (DLF), qui propose une gestion unifiée des métadonnées et du stockage Paimon. Vous pouvez également utiliser OSS-HDFS ou HDFS.
-
Prise en compte des modifications de configuration
Les modifications apportées aux paramètres de configuration d'une table Paimon ne prennent effet qu'après le redémarrage des jobs associés. Les jobs en cours d'exécution ne chargent pas dynamiquement ces modifications.
-
Suppression physique différée des partitions supprimées
Lorsque vous exécutez une opération DROP PARTITION, le système ne supprime pas immédiatement les fichiers de données physiques sous-jacents. Cette opération effectue une suppression logique. Paimon retire les métadonnées de la partition cible uniquement du dernier snapshot. Étant donné que Paimon prend en charge la fonctionnalité de voyage dans le temps (time travel), les snapshots historiques font toujours référence aux fichiers de données de cette partition. Les fichiers de données physiques sont définitivement supprimés uniquement lorsque tous les snapshots historiques référençant la partition atteignent leur limite de rétention et sont nettoyés par le mécanisme d'expiration des snapshots.
SQL
Utilisez le connecteur Paimon dans les jobs SQL comme table source ou table de résultat.
Syntaxe
-
Si vous créez une table Paimon dans un catalogue Paimon, vous n'avez pas besoin de spécifier le paramètre
connector. La syntaxe est la suivante :CREATE TABLE `<YOUR-PAIMON-CATALOG>`.`<YOUR-DB>`.paimon_table ( id BIGINT, data STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( ... );RemarqueSi vous avez déjà créé une table Paimon dans un catalogue Paimon, vous pouvez l'utiliser directement.
-
Si vous créez une table temporaire Paimon dans un autre catalogue, vous devez spécifier les paramètres 'connector' et 'path'. La syntaxe est la suivante :
CREATE TEMPORARY TABLE paimon_table ( id BIGINT, data STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'paimon', 'path' = '<path-to-paimon-table-files>', 'auto-create' = 'true', -- If Paimon table data files do not exist at the specified path, they are automatically created. ... );RemarqueExemple de chemin :
'path' = 'oss://<bucket>/test/order.db/orders'. N'omettez pas le suffixe.db. Paimon s'appuie sur ce suffixe pour identifier la base de données.Plusieurs jobs écrivant dans la même table doivent partager la même configuration de chemin.
Si deux configurations de chemin diffèrent, Paimon ne les reconnaît pas comme appartenant à la même table. Même si le chemin physique est identique, des configurations de catalogue incohérentes peuvent entraîner des conflits d'écriture concurrents, des échecs d'opérations de compactage et une perte de données. Par exemple, Paimon considère
oss://b/testetoss://b/test/comme des tables différentes en raison de la barre oblique finale, bien qu'elles puissent pointer vers le même emplacement physique.
Paramètres WITH
|
Parameter |
Description |
Type |
Required |
Default |
Remarks |
|
connector |
Spécifie le connecteur de la table. |
String |
No |
None |
|
|
path |
Le chemin de stockage de la table. |
String |
No |
None |
|
|
auto-create |
Indique s'il faut créer automatiquement les fichiers de la table s'ils n'existent pas au chemin spécifié. |
Boolean |
No |
false |
Valeurs valides :
|
|
file.format |
Le format des fichiers de données. |
String |
No |
parquet |
Valeurs valides :
|
|
bucket |
Le nombre de buckets par partition. |
Integer |
No |
1 |
Paimon distribue les données dans les buckets en fonction de la clé Remarque
Nous recommandons que chaque bucket contienne moins de 5 Go de données. |
|
bucket-key |
La ou les colonnes utilisées comme clé de bucket. |
String |
No |
None |
Spécifie les colonnes utilisées pour distribuer les données dans les buckets. Séparez plusieurs noms de colonnes par une virgule (,). Par exemple, Remarque
|
|
changelog-producer |
Le mécanisme de production du journal des modifications (changelog). |
String |
No |
none |
Paimon peut produire un journal des modifications complet pour tout flux d'entrée, ce qui signifie que chaque enregistrement
Pour plus d'informations sur le choix d'un producteur de journal des modifications, consultez la section Production de journaux des modifications. |
|
full-compaction.delta-commits |
Le nombre maximal de validations entre deux compactages intégraux consécutifs. |
Integer |
No |
None |
Spécifie le nombre maximal de validations d'instantanés autorisées avant le déclenchement d'un compactage intégral. |
|
lookup.cache-max-memory-size |
La taille du cache mémoire pour une table de dimension Paimon. |
String |
No |
256 MB |
Ce paramètre contrôle la taille du cache pour les recherches dans les tables de dimension ainsi que pour le producteur de journal des modifications |
|
merge-engine |
Le mécanisme de fusion des enregistrements partageant la même clé primaire. |
String |
No |
deduplicate |
Valeurs valides :
Pour une analyse détaillée des moteurs de fusion, consultez la section Moteur de fusion. |
|
partial-update.ignore-delete |
Indique s'il faut ignorer les messages de suppression (-D). |
Boolean |
No |
false |
Valeurs valides :
Remarque
|
|
ignore-delete |
Indique s'il faut ignorer les messages de suppression (-D). |
Boolean |
No |
false |
Les valeurs valides sont identiques à celles de partial-update.ignore-delete. Remarque
|
|
partition.default-name |
Le nom de partition par défaut. |
String |
No |
__DEFAULT_PARTITION__ |
Le nom de partition à utiliser lorsque la valeur d'une colonne de partition est nulle ou une chaîne vide. |
|
partition.expiration-check-interval |
La fréquence à laquelle le système vérifie les partitions expirées. |
String |
No |
1h |
Pour plus de détails, consultez la section Configuration de l'expiration automatique des partitions. |
|
partition.expiration-time |
La durée après laquelle les partitions expirent. |
String |
No |
None |
Une partition expire lorsque son âge dépasse cette valeur. Par défaut, les partitions n'expirent jamais. Le système calcule l'âge d'une partition à partir de sa valeur de partition. Pour plus de détails, consultez la section Configuration de l'expiration automatique des partitions. |
|
partition.timestamp-formatter |
La chaîne de format pour convertir une chaîne de temps en horodatage. |
String |
No |
None |
Spécifie le format d'extraction de l'âge de la partition à partir de la valeur de partition. Pour plus de détails, consultez la section Configuration de l'expiration automatique des partitions. |
|
partition.timestamp-pattern |
La chaîne de format pour convertir une valeur de partition en chaîne de temps. |
String |
No |
None |
Spécifie le modèle d'extraction d'une chaîne de temps à partir de la valeur de partition. Pour plus de détails, consultez la section Configuration de l'expiration automatique des partitions. |
|
scan.bounded.watermark |
La valeur de watermark signalant la fin d'une analyse. La table source arrête de produire des données lorsque son watermark dépasse cette valeur. |
Long |
No |
None |
N/A |
|
scan.mode |
Spécifie la position de consommation pour la table source Paimon. |
String |
No |
default |
Pour plus de détails, consultez la section Définition de la position de consommation pour une table source Paimon. |
|
scan.snapshot-id |
Spécifie l'instantané à partir duquel la table source Paimon commence la consommation. |
Integer |
No |
None |
Pour plus de détails, consultez la section Définition de la position de consommation pour une table source Paimon. |
|
scan.timestamp-millis |
Spécifie le point dans le temps à partir duquel la table source Paimon commence la consommation. |
Integer |
No |
None |
Pour plus de détails, consultez la section Définition de la position de consommation pour une table source Paimon. |
|
snapshot.num-retained.max |
Le nombre maximal d'instantanés récents à conserver. |
Integer |
No |
2147483647 |
L'expiration des instantanés est déclenchée si cette condition ou la condition |
|
snapshot.num-retained.min |
Le nombre minimal d'instantanés récents à conserver. |
Integer |
No |
10 |
N/A |
|
snapshot.time-retained |
La période de rétention des instantanés. |
String |
No |
1h |
L'expiration des instantanés est déclenchée si cette condition ou la condition |
|
write-mode |
Le mode d'écriture pour la table Paimon. |
String |
No |
change-log |
Valeurs valides :
Pour plus d'informations sur les modes d'écriture, consultez la section Mode d'écriture. |
|
scan.infer-parallelism |
Indique s'il faut déduire automatiquement le parallélisme pour une table source Paimon. |
Boolean |
No |
true |
Valeurs valides :
|
|
scan.parallelism |
Le parallélisme pour la table source Paimon. |
Integer |
No |
None |
Remarque
Ce paramètre est ignoré si Resource Mode est défini sur Mode dans l'onglet du job. |
|
sink.parallelism |
Le parallélisme pour la table de destination Paimon. |
Integer |
No |
None |
Remarque
Ce paramètre est ignoré si Resource Mode est défini sur Mode dans l'onglet du job. |
|
sink.clustering.by-columns |
Spécifie les colonnes de regroupement pour l'écriture dans la table de destination Paimon. |
String |
No |
None |
Pour les tables Paimon en ajout seul (tables sans clé primaire), ce paramètre active l'écriture regroupée dans les jobs par lots. Ce processus améliore les performances des requêtes en regroupant les données sur les colonnes spécifiées. Séparez plusieurs noms de colonnes par une virgule (,), par exemple Pour plus d'informations sur le regroupement, consultez la documentation officielle d'Apache Paimon. |
|
sink.delete-strategy |
Spécifie une stratégie de validation pour garantir que le système gère correctement les messages de rétractation (-D/-U). |
Enum |
No |
NONE |
Valeurs valides et comportement attendu de l'opérateur de destination lors de la gestion des messages de rétractation :
Remarque
|
|
blob-as-descriptor |
Indique s'il faut générer les octets du descripteur Blob lors de la lecture d'une colonne Blob. |
Boolean |
No |
false |
Lorsque ce paramètre est défini sur true, les requêtes renvoient les octets sérialisés du BlobDescriptor au lieu du contenu réel du Blob. Utilisez ce paramètre conjointement avec la fonction d'URL présignée Blob. Il n'est pas nécessaire de définir ce paramètre lors de la création de la table. Vous pouvez le définir dynamiquement au moment de la lecture à l'aide d'un indice SQL. Ce paramètre est pris en charge dans VVR 11.9-preview1 et versions ultérieures. Remarque
Pris en charge uniquement dans VVR 11.9-preview1 et versions ultérieures. |
Pour plus d'informations sur les options de configuration, consultez la documentation officielle d'Apache Paimon.
Paramètres de table vectorielle
Les paramètres suivants sont pris en charge uniquement dans VVR 11.8 et versions ultérieures.
|
Parameter |
Description |
Data type |
Required |
Default value |
Remarks |
|
|
Le nombre de clusters IVF à sonder pendant la recherche. |
Integer |
No |
16 |
Des valeurs plus élevées améliorent généralement le rappel, mais augmentent la latence. |
|
|
Récupère les candidats IVF top_k × refine_factor et les reclasse à l'aide des vecteurs d'origine stockés dans la table Paimon. |
Float |
No |
Disabled |
Désactivé par défaut pour toutes les variantes IVF. Idéal pour les index compressés (tels que ivf-pq et ivf-hnsw-sq) dans les scénarios où le rappel est prioritaire par rapport à la latence. |
|
|
La largeur de recherche HNSW pendant la recherche. |
Integer |
No |
0 |
Des valeurs plus élevées améliorent généralement le rappel, mais augmentent la latence. 0 indique que la valeur par défaut de la bibliothèque native est utilisée. |
|
|
La taille de la liste de recherche Lumina DiskANN. |
Integer |
No |
max(1.5 × top_k, 16) |
Des valeurs plus élevées améliorent généralement le rappel, mais augmentent la latence. |
|
|
La largeur de faisceau de recherche Lumina DiskANN. |
Integer |
No |
4 |
— |
|
|
Le nombre de recherches Lumina parallèles. |
Integer |
No |
5 |
— |
Détails des fonctionnalités
Fraîcheur et cohérence des données
La table de destination Paimon utilise le protocole de validation en deux phases pour valider les données lors de chaque point de contrôle (checkpoint) du job Flink. Par conséquent, la fraîcheur des données dépend de l'intervalle de checkpoint du job Flink. Chaque validation produit au maximum deux snapshots.
Lorsque deux jobs Flink écrivent simultanément dans la même table Paimon, ils atteignent une cohérence sérialisable s'ils écrivent dans des buckets différents. S'ils écrivent dans le même bucket, ils n'atteignent que l'isolation par snapshot. Cela signifie que les données de la table peuvent être un mélange des résultats des deux jobs, mais aucune perte de données ne se produit.
Moteur de fusion
Lorsqu'une table de destination Paimon reçoit plusieurs enregistrements ayant la même clé primaire, elle les fusionne en un seul enregistrement pour garantir l'unicité. Vous pouvez contrôler ce comportement en définissant le paramètre merge-engine. Le tableau suivant décrit les moteurs de fusion disponibles.
|
Moteur de fusion |
Description |
|
Deduplicate |
Le moteur de déduplication est activé par défaut. Pour plusieurs enregistrements partageant la même clé primaire, la table de destination Paimon conserve uniquement le dernier enregistrement et ignore les autres. Remarque
Si le dernier enregistrement est un message de suppression, tous les enregistrements associés à cette clé primaire sont ignorés. |
|
Partial Update |
Le moteur de mise à jour partielle permet de construire un enregistrement complet en appliquant des mises à jour incrémentielles via plusieurs messages. Lorsqu'un nouvel enregistrement avec la même clé primaire arrive, ses valeurs non nulles écrasent les champs correspondants de l'enregistrement existant. Le moteur ignore les champs nuls du nouvel enregistrement et conserve les valeurs existantes. Par exemple, supposons qu'une table de destination Paimon reçoive les trois enregistrements suivants dans l'ordre :
Si la première colonne est la clé primaire, l'enregistrement fusionné final est <1, 25,2, 10, 'This is a book'>. Remarque
|
|
Aggregation |
Dans certains cas d'utilisation, seule la valeur agrégée des enregistrements est nécessaire. Le moteur d'agrégation combine les enregistrements ayant la même clé primaire en utilisant les fonctions d'agrégation que vous spécifiez. Pour chaque colonne qui n'est pas une clé primaire, vous devez spécifier une fonction d'agrégation à l'aide de l'option La colonne
Remarque
|
Générateur de journal des modifications (Changelog producer)
Définissez le paramètre changelog-producer pour configurer Paimon afin qu'il génère un journal des modifications complet (où chaque enregistrement update_after possède un enregistrement update_before correspondant) pour tout flux d'entrée. Le tableau suivant décrit les générateurs de journal des modifications disponibles. Pour plus de détails, consultez la documentation officielle d'Apache Paimon.
|
Générateur |
Description |
|
None |
Lorsque vous définissez Par exemple, si un consommateur en aval doit calculer la somme d'une colonne et ne voit que la dernière valeur de 5, il ne peut pas déterminer comment mettre à jour le total. Si la valeur précédente était 4, la somme devrait augmenter de 1 ; si la valeur précédente était 6, la somme devrait diminuer de 1. Les consommateurs sensibles aux enregistrements Remarque
Si votre consommateur en aval, tel qu'une base de données, n'est pas sensible aux données |
|
Input |
Lorsque vous définissez Utilisez ce générateur uniquement lorsque le flux d'entrée est déjà un journal des modifications complet, comme les données issues de la capture des données modifiées (CDC). |
|
Lookup |
Lorsque vous définissez Par rapport au générateur Privilégiez cette option pour les cas d'utilisation nécessitant une grande fraîcheur des données (par exemple, au niveau de la minute). |
|
Full Compaction |
Lorsque vous définissez Par rapport au générateur Privilégiez cette option pour les cas d'utilisation ayant des exigences de fraîcheur des données faibles (par exemple, au niveau de l'heure). |
Mode d'écriture
Les tables Paimon prennent en charge les modes d'écriture suivants.
|
Mode |
Description |
|
Change-log |
Le mode d'écriture |
|
Append-only |
Le mode d'écriture Pour une description détaillée du mode d'écriture
|
Cible pour l'ingestion de données Flink CDC
Les tables Paimon prennent en charge la synchronisation en temps réel des données provenant de tables uniques ou de bases de données entières. Les modifications de schéma en amont sont synchronisées vers les tables Paimon en temps réel. Pour plus d'informations, consultez la section Ingestion de données.
Élagage de lecture Variant (Read pruning)
Lit uniquement les champs Variant référencés par la requête, réduisant ainsi la charge d'E/S et de mémoire.
Comment activer
|
Paramètre |
Valeur par défaut |
Description |
|
|
false |
Définissez sur |
Prérequis
Les données Variant ne peuvent être écrites que par VVR 11.6 ou version ultérieure. Les données Variant écrites par des versions antérieures ou d'autres moteurs ne prennent pas en charge l'élagage de lecture.
Pour que les données en amont prennent effet, vous devez activer l'option variant.inferShreddingSchema pour permettre l'inférence automatique du découpage Variant, ou configurer explicitement variant.shreddingSchema et les options associées pour spécifier quels membres Variant sont adaptés au découpage. Pour plus d'informations, consultez Coreoptions.
Scénarios pris en charge
L'élagage de lecture prend effet lorsque SQL accède aux champs Variant par clé de chaîne, par exemple :
SELECT v['a'] FROM t; -- Single level
SELECT v['a']['b'] FROM t; -- Nested
SELECT v['a'], v['b'] FROM t; -- Multiple fields
SELECT id, v['a'] FROM t; -- Mixed with regular columns
SELECT v['a'] + 1 FROM t; -- Referenced in expressionsScénarios non pris en charge
Référencer directement l'intégralité du Variant, par exemple
SELECT v FROM t.Combiner la référence directe avec l'accès aux champs, par exemple
SELECT v, v['a'] FROM t.Accès par index de tableau, par exemple
v[0].Le schéma de la table contient des champs Row imbriqués. Dans ce cas, l'élagage de lecture ne s'applique à aucun champ Variant de la table.
URL présignées pour les blobs
Paimon peut générer des URL HTTPS GET présignées publiques à courte durée de vie pour les données de colonne Blob stockées dans OSS. Les services externes, tels que les services de modèles multimodaux, peuvent utiliser ces URL pour récupérer directement le contenu du fichier, sans transférer les octets du fichier au sein d'un job Flink.
Limites
Seul le moteur de calcul Flink VVR 11.9-preview1 et les versions ultérieures prennent en charge cette fonctionnalité.
Seules les tables Paimon stockées dans OSS sont prises en charge. Le type de metastore n'a pas d'importance. Les types de metastore Filesystem, DLF et autres sont tous pris en charge.
L'URL générée est une URL HTTPS publique et doit être consommée par un service externe disposant d'un accès au réseau public. Si le catalogue utilise un endpoint interne, nous vous recommandons de le remplacer par un endpoint HTTPS public afin que le service externe puisse accéder à l'URL générée.
Pour lire une colonne Blob standard, vous devez d'abord obtenir les octets BlobDescriptor en utilisant l'option
blob-as-descriptor, puis transmettre les octets à la fonction.
Syntaxe
descriptor_to_presigned_url(source_table, descriptor, validity)
try_descriptor_to_presigned_url(source_table, descriptor, validity)Paramètres
|
Paramètre |
Type |
Description |
|
|
Littéral STRING |
La table Paimon contenant la colonne Blob, au format |
|
|
BYTES |
Les octets BlobDescriptor, qui correspondent au résultat de la lecture de la colonne Blob après activation de |
|
|
INTERVAL |
La période de validité de l'URL. La valeur doit être un nombre positif de secondes, par exemple |
Valeur de retour
|
Type |
Description |
|
STRING |
Une URL HTTPS GET présignée publique à courte durée de vie. |
Exemple
-- Use an SQL hint to read the Blob column as a descriptor and generate a presigned URL
SELECT sys.descriptor_to_presigned_url(
'default.image_table',
image,
INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;
-- Fault-tolerant mode: returns NULL on row-level errors, with other behavior unchanged
SELECT sys.try_descriptor_to_presigned_url(
'default.image_table',
image,
INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;descriptor_to_presigned_urllève une exception en cas d'erreur au niveau de la ligne.try_descriptor_to_presigned_urlrenvoie NULL en cas d'erreur au niveau de la ligne.L'URL générée ne contient pas d'extension de fichier, et les octets vers lesquels pointe l'URL sont identiques aux données Blob. Les services en aval doivent identifier le format de fichier en fonction des octets renvoyés plutôt que du suffixe de l'URL.
Une URL présignée équivaut à des identifiants d'accès temporaires. Une fois l'URL présignée générée, soumettez-la immédiatement au service en aval. Ne l'écrivez pas dans les journaux ou les tables de résultats, et ne la persistez pas.
Table chaînée
Une table chaînée reconstruit un jeu de données complet à partir d'une branche snapshot et des modifications qui la suivent dans une branche delta. Écrivez périodiquement les données complètes dans la branche snapshot et les modifications entre deux chargements complets dans la branche delta. Une lecture utilise la dernière partition snapshot comme ancre et fusionne les partitions delta ultérieures pour reconstituer les données complètes, ce qui évite de réécrire l'intégralité de la table. Cette approche est utile lorsque seule une petite fraction des lignes change entre les exécutions, comme lors d'un vidage binlog ODS. Pour plus d'informations, consultez la rubrique Chain Table dans la documentation Apache Paimon. Les tables chaînées sont prises en charge uniquement à partir de VVR 11.9.
Activer une table chaînée
Définissez chain-table.enabled sur true lors de la création de la table, créez les branches snapshot et delta, puis configurez les options de secours (fallback) sur la table principale et sur les deux tables de branche.
CREATE TABLE `default`.`t` (
`t1` STRING,
`t2` STRING,
`t3` STRING,
`date` STRING,
PRIMARY KEY (`date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`date`) WITH (
'chain-table.enabled' = 'true',
'sequence.field' = 't2',
'bucket-key' = 't1',
'bucket' = '2',
'partition.timestamp-pattern' = '$date',
'partition.timestamp-formatter' = 'yyyyMMdd'
);
-- mycat is the name of the Paimon catalog.
CALL `mycat`.sys.create_branch('default.t', 'snapshot');
CALL `mycat`.sys.create_branch('default.t', 'delta');
ALTER TABLE `default`.`t` SET (
'scan.fallback-snapshot-branch' = 'snapshot',
'scan.fallback-delta-branch' = 'delta');
ALTER TABLE `default`.`t$branch_snapshot` SET (
'scan.fallback-snapshot-branch' = 'snapshot',
'scan.fallback-delta-branch' = 'delta');
ALTER TABLE `default`.`t$branch_delta` SET (
'scan.fallback-snapshot-branch' = 'snapshot',
'scan.fallback-delta-branch' = 'delta');Écrire des données
Écrivez les données complètes dans t$branch_snapshot. Chaque partition de cette branche offre une vue complète à un instant donné. Écrivez les données incrémentielles dans t$branch_delta. Chaque partition de cette branche contient uniquement les modifications de la période correspondante.
-- Full write
INSERT OVERWRITE `default`.`t$branch_snapshot` PARTITION (date = '20250810') VALUES ('1', '1', '1');
-- Incremental write
INSERT INTO `default`.`t$branch_delta` PARTITION (date = '20250811') VALUES ('2', '1', '1');Lire des données
|
Chemin de lecture |
Opération |
Résultat |
|
Requête complète |
Interrogez une partition de la table principale. |
Les données complètes de la partition. La partition est lue directement si la branche snapshot la contient. Sinon, elle est reconstituée à partir d'un snapshot ancre et des partitions delta ultérieures. |
|
Requête incrémentielle |
Interrogez une partition de |
Uniquement les modifications stockées dans la partition delta sélectionnée. L'état complet n'est pas reconstitué. |
|
Requête hybride |
Combinez une requête complète et une requête incrémentielle avec |
La combinaison des résultats complets et incrémentiels. La même ligne peut apparaître deux fois. |
|
Lecture en streaming |
Lisez la table principale en mode streaming. |
Le résultat de la phase de chargement initial est émis en premier, puis les nouveaux commits de la branche delta sont consommés en continu. |
|
Jointure de recherche (Lookup join) |
Utilisez la table chaînée comme table de dimension. |
L'état fusionné de la dernière partition snapshot et des partitions delta ultérieures de chaque groupe, actualisé de manière incrémentielle selon |
-- Full query
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811';
-- Incremental query
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';
-- Hybrid query
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811'
UNION ALL
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';Lecture en streaming
Un job de lecture en streaming s'exécute en deux phases :
Phase de chargement initial : par défaut, le job lit la dernière partition snapshot de chaque groupe et les partitions delta qui la suivent. Les partitions snapshot plus anciennes sont exclues. Pour fusionner les données snapshot et delta par ancre durant cette phase et résoudre les enregistrements DELETE inter-branches, définissez
chain-table.streaming.merge-snapshotsurtrue. Cela implique toutefois un balayage plus lourd au démarrage du job.Phase incrémentielle : le job lit en continu les nouveaux commits de la branche
delta.
SET 'execution.runtime-mode' = 'streaming';
INSERT INTO downstream_sink SELECT * FROM `default`.`t`;La phase incrémentielle surveille uniquement la branche delta. Les écritures dans la branche snapshot ne sont pas détectées tant que le job n'a pas été redémarré. Exécutez régulièrement la procédure compact_chain_table pour fusionner les données incrémentielles d'une partition dans la branche snapshot. Après la compaction, seules les modifications delta arrivées après la compaction doivent être fusionnées lors de la phase de chargement initial.
Clés de partition multiples
Si la table possède plusieurs clés de partition, utilisez chain-table.chain-partition-keys pour séparer la dimension de chaîne de la dimension de groupe. La valeur doit être un suffixe contigu des clés de partition. Les champs du suffixe forment la chaîne ordonnée dans le temps, et les champs de partition précédents constituent la dimension de groupe. Chaque groupe de valeurs de partition de groupe maintient sa propre chaîne et son ancre. Si cette option n'est pas définie, toutes les partitions appartiennent à un seul groupe implicite.
CREATE TABLE `default`.`t` (
`t1` STRING,
`t2` STRING,
`t3` STRING,
`region` STRING,
`date` STRING,
PRIMARY KEY (`region`, `date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`region`, `date`) WITH (
'chain-table.enabled' = 'true',
'sequence.field' = 't2',
'bucket-key' = 't1',
'bucket' = '2',
'partition.timestamp-pattern' = '$date',
'partition.timestamp-formatter' = 'yyyyMMdd',
-- Only date forms the chain. Each region has its own chain.
'chain-table.chain-partition-keys' = 'date'
);Limitations
Les tables chaînées sont prises en charge uniquement pour les tables à clé primaire. Vous devez spécifier
bucketetbucket-key.Le schéma de chaque branche doit être cohérent.
Une table chaînée nécessite
sequence.field. Les moteurs de fusionfirst-rowetaggregationne sont pas pris en charge.La lecture en streaming et la jointure de recherche (lookup join) exigent que la branche
deltautilise le moteur de fusiondeduplicate(valeur par défaut). La lecture par lot n'est pas affectée.Les vecteurs de suppression sont pris en charge uniquement pour les tables chaînées utilisant le moteur de fusion
deduplicate.Seules les valeurs
none(valeur par défaut) etinputsont prises en charge pourchangelog-producer. Les modeslookupetfull-compactionne sont pas pris en charge. Si la valeur estnone, les enregistrements sont normalisés par la clé primaire complète incluant la partition de chaîne, et les enregistrements-Det-Udont les enregistrements d'origine résident dans une autre partition sont supprimés. Définissez la valeur surinputsi les consommateurs en aval doivent recevoir des enregistrements de changelog inter-partitions.La lecture en streaming et la jointure de recherche (lookup join) prennent en charge uniquement le mode de démarrage par défaut (
latest-full). Une erreur est générée si vous spécifiez une position de départ telle quescan.snapshot-id,scan.timestamp-millis,scan.modedéfini surlatest, ouconsumer-id. Pour effectuer une lecture en streaming standard sans la logique de table chaînée, lisez une table de branche spécifique telle quet$branch_delta.Les filtres de partition ne sont pas pris en charge dans les lectures en streaming de tables chaînées et les jointures de recherche. Cela inclut les colonnes de partition dans la clause
WHEREet l'optionscan.partitions, car un filtre interfère avec la logique de fusion inter-branches. Utilisez le mode batch pour lire une partition spécifique.La clé de jointure d'une jointure de recherche (lookup join) ne doit pas contenir de clés de partition. Sinon, une erreur est renvoyée.
partition.timestamp-formatterdoit spécifier une date complète (année-mois-jour ou année-jour-de-l'année), utiliser uniquement des champs de date et d'heure réversibles (y/u,M/L,d,D,H,k,mets), et avoir un pas minimum d'au moins une seconde.
Format de fichier Mosaic
Mosaic est un format hybride colonne-bucket optimisé pour les tables larges contenant des milliers ou des dizaines de milliers de colonnes. Les colonnes sont triées par nom et distribuées dans des buckets par plage, stockées au format orienté colonne dans chaque bucket, et compressées indépendamment à l'aide de ZSTD. Le pushdown de projection fonctionne au niveau du bucket, de sorte qu'une requête lit uniquement les buckets contenant les colonnes demandées. Les colonnes ayant des préfixes de nom similaires atterrissent dans le même bucket, ce qui améliore à la fois le taux de compression et la localité des projections. Pour plus d'informations, consultez la rubrique Mosaic. Le format Mosaic est pris en charge uniquement à partir de VVR 11.9.
Activer le format
Définissez file.format sur mosaic lors de la création de la table. Dans un job d'ingestion de données, l'option correspondante est table.properties.file.format.
CREATE TABLE `default`.`wide_table` (
`id` BIGINT,
`col_a` STRING,
`col_b` DOUBLE
) WITH (
'file.format' = 'mosaic'
);Options de format
|
Option |
Valeur par défaut |
Description |
|
|
auto |
Le nombre de buckets de colonnes pour les E/S parallèles. Si la valeur est 0 ou si l'option n'est pas spécifiée, le nombre de buckets est déterminé automatiquement. |
|
|
(vide) |
Les noms des colonnes pour lesquelles les statistiques min/max sont collectées pour le pushdown de prédicat, séparés par des virgules (,). Aucune statistique n'est collectée si la valeur est vide. |
|
|
8 |
Le nombre de groupes de lignes qu'un lecteur ouvre avant le groupe de lignes en cours de consommation. Le préchargement permet de masquer la latence de lecture de plage avec le décodage. Chaque groupe de lignes préchargé conserve son lot décodé en mémoire et utilise un flux d'entrée distinct. La valeur 0 désactive le préchargement. |
|
|
64 mb |
La limite supérieure de la taille décodée estimée des groupes de lignes préchargés, calculée à partir de leur nombre de lignes et des types de colonnes projetés. Une projection large ou un grand groupe de lignes réduit la valeur effective de |
Limitations
Mosaic ne prend pas en charge les types complexes : ARRAY, MAP, MULTISET, ROW, VARIANT, BLOB et VECTOR.
Utiliser Paimon comme table de dimension
Les tables Paimon peuvent être utilisées comme tables de dimension. Pour la syntaxe JOIN, consultez la rubrique Dimension table JOIN statements.
Par défaut, la recherche (lookup) charge toutes les données dans chaque instance parallèle. Cette approche convient uniquement aux petites tables de dimension. Pour les grandes tables de dimension, utilisez les solutions Shuffle Lookup décrites ci-dessous.
Tables de dimension partitionnées
Si votre table de dimension est partitionnée et que vous avez uniquement besoin des données de la ou des deux dernières partitions, vous pouvez utiliser la fonctionnalité de chargement dynamique de partitions :
SELECT * FROM T
JOIN DIM /*+ OPTIONS('lookup.dynamic-partition'='max_pt()', 'lookup.dynamic-partition.refresh-interval'='1 h') */
FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col = D.col;|
Paramètre |
Type de données |
Valeur par défaut |
Description |
|
lookup.dynamic-partition |
String |
N/A |
|
|
lookup.dynamic-partition.refresh-interval |
Duration |
1 h |
L'intervalle auquel le système vérifie les mises à jour de partition dans la table de dimension. |
Grandes tables de dimension : tables à buckets fixes
Pris en charge uniquement à partir de VVR 8.0.8. Pour les tables à buckets fixes (bucket > 0), vous pouvez utiliser Shuffle Lookup pour distribuer les données par clé de bucket across les instances parallèles, de sorte que chaque instance charge uniquement les données de ses buckets assignés :
SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;La clé de jointure doit être la clé de bucket. La clé de bucket est par défaut la clé primaire.
Seules les tables à buckets fixes (bucket > 0) prennent en charge cette fonctionnalité.
Grandes tables de dimension : tables à buckets non fixes
Pris en charge uniquement à partir de VVR 8.0.10. Pour les tables à buckets dynamiques ou les tables d'ajout (append), vous pouvez utiliser SHUFFLE_HASH ou REPLICATED_SHUFFLE_HASH afin que chaque instance parallèle lise toutes les données mais ne conserve que la partie dont elle a besoin :
-- Shuffle Hash
SELECT /*+ SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;
-- Replicated Shuffle Hash
SELECT /*+ REPLICATED_SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;Pour plus d'informations sur SHUFFLE_HASH et REPLICATED_SHUFFLE_HASH, consultez la rubrique Dimension table JOIN statements.
Ingestion de données
Vous pouvez utiliser le connecteur Paimon comme sink dans les jobs YAML d'ingestion de données.
Syntaxe
sink:
type: paimon
name: Paimon Sink
catalog.properties.metastore: filesystem
catalog.properties.warehouse: /path/warehouseParamètres
|
Paramètre |
Description |
Obligatoire |
Type |
Valeur par défaut |
Notes |
|
type |
Le type de connecteur. |
Oui |
STRING |
Aucune |
La valeur doit être |
|
name |
Le nom du récepteur (sink). |
Non |
STRING |
Aucune |
|
|
catalog.properties.metastore |
Le type du catalogue Paimon. |
Non |
STRING |
filesystem |
Valeurs valides :
|
|
catalog.properties.* |
Paramètres pour la création d'un catalogue Paimon. |
Non |
STRING |
Aucune |
Pour plus d'informations, consultez Gérer le catalogue Paimon. |
|
table.properties.* |
Paramètres pour la création d'une table Paimon. |
Non |
STRING |
Aucune |
Pour plus d'informations, consultez Options de table Paimon. |
|
table.properties.file.format |
Le format de stockage des fichiers de données dans la table. |
String |
Non |
parquet |
Valeurs valides :
|
|
catalog.properties.warehouse |
Le répertoire racine pour le stockage des fichiers. |
Non |
STRING |
Aucune |
Ce paramètre s'applique uniquement lorsque |
|
commit.user-prefix |
Le préfixe du nom d'utilisateur pour la validation (commit) des fichiers de données. |
Non |
STRING |
Aucune |
Remarque
Nous vous recommandons de définir des noms d'utilisateurs différents pour chaque tâche. Cela facilite l'identification de la tâche à l'origine d'un conflit de validation. |
|
partition.key |
Les clés de partition pour une table partitionnée. |
Non |
STRING |
Aucune |
Les différentes tables sont séparées par |
|
sink.cross-partition-upsert.tables |
Répertorie les tables nécessitant une opération upsert inter-partitions, où la clé primaire n'inclut pas toutes les clés de partition. |
Non |
STRING |
Aucune |
S'applique aux tables avec des mises à jour inter-partitions.
Important
|
|
sink.commit.parallelism |
Spécifie le parallélisme de l'opérateur Commit. |
Non |
INTEGER |
Aucune |
Si l'opérateur Commit constitue un goulot d'étranglement, utilisez ce paramètre pour augmenter son parallélisme et améliorer les performances. Ce paramètre est pris en charge uniquement dans Realtime Compute for Apache Flink 11.6 et versions ultérieures. Remarque
La définition de ce paramètre modifie le parallélisme de l'opérateur. Lors du redémarrage d'une tâche avec état, vous devez spécifier |
|
sink.existing-table.schema-check.enabled |
Indique s'il faut vérifier la compatibilité du schéma de table existant et adapter automatiquement le schéma lors de l'écriture des données dans une table précréée. |
Boolean |
Non |
true |
La liste suivante décrit les valeurs valides et le comportement lors de l'écriture des données dans une table précréée :
Remarque
Ce paramètre est pris en charge uniquement dans Realtime Compute for Apache Flink VVR 11.8 et versions ultérieures. |
Réutiliser un catalogue existant
À partir de Realtime Compute for Apache Flink 11.5, vous pouvez référencer directement un catalogue Paimon intégré depuis la page Data Management dans une tâche d'ingestion de données Flink CDC. Cela réduit la configuration manuelle.
sink:
type: paimon
using.built-in-catalog: paimon_dlf_catalog
catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.comLes tâches d'ingestion de données peuvent réutiliser automatiquement tous les paramètres d'un catalogue Paimon. Cela équivaut à configurer manuellement les paramètres préfixés par catalog.properties. dans une tâche YAML.
Pour remplacer un paramètre réutilisé automatiquement, définissez-le explicitement dans la tâche YAML. La configuration YAML explicite a une priorité plus élevée. Par exemple, dans l'échantillon précédent, le paramètre fs.oss.endpoint utilise la valeur de la tâche YAML, remplaçant celle de paimon_dlf_catalog.
Modifications de schéma
Une tâche de pipeline YAML CDC peut utiliser différentes stratégies pour gérer les modifications de schéma. Pour configurer la stratégie, définissez le paramètre au niveau du pipeline schema.change.behavior comme décrit dans Configurations d'évolution de schéma. schema.change.behavior peut être défini sur IGNORE, LENIENT, TRY_EVOLVE, EVOLVE ou EXCEPTION. Les sections suivantes décrivent comment les modes LENIENT et EVOLVE, qui impliquent des modifications de schéma, gèrent différents événements de modification de schéma.
LENIENT (par défaut)
Les modifications de schéma suivantes sont prises en charge en mode LENIENT :
Ajout d'une colonne acceptant les valeurs NULL : la colonne est automatiquement ajoutée au schéma de la table de résultat, et les données de la nouvelle colonne sont automatiquement synchronisées.
Suppression d'une colonne acceptant les valeurs NULL : la colonne n'est pas supprimée de la table de résultat. À la place, les données de la colonne sont automatiquement remplacées par des valeurs NULL.
Ajout d'une colonne NOT NULL : la colonne est automatiquement ajoutée au schéma de la table de résultat, et les données de la nouvelle colonne sont automatiquement synchronisées. La nouvelle colonne accepte les valeurs NULL par défaut, et les données générées avant l'ajout de la colonne sont définies sur NULL.
Renommage d'une colonne : l'opération est traitée comme l'ajout d'une colonne et la suppression d'une colonne. La colonne renommée est ajoutée à la fin de la table de résultat, et les données de la colonne d'origine sont automatiquement remplacées par des valeurs NULL. Par exemple, si col_a est renommé en col_b, col_b est ajouté à la fin de la table de résultat, et les données de col_a sont remplacées par des valeurs NULL.
Modification du type d'une colonne : modifie le type de données de la colonne correspondante dans la table de résultat.
-
Les modifications de schéma suivantes ne sont pas prises en charge :
Modifications des types de clé primaire.
Passage de NULLABLE à NOT NULL.
EVOLVE
Les modifications de schéma suivantes sont prises en charge en mode EVOLVE :
Ajout d'une colonne acceptant les valeurs NULL : pris en charge.
Suppression d'une colonne acceptant les valeurs NULL : non pris en charge.
Ajout d'une colonne NOT NULL : une colonne acceptant les valeurs NULL est ajoutée à la table de résultat.
Renommage d'une colonne : pris en charge. La colonne d'origine est renommée dans la table de résultat.
Modification du type d'une colonne : modifie le type de données de la colonne correspondante dans la table de résultat.
-
Les modifications de schéma suivantes ne sont pas prises en charge :
Modifications des types de clé primaire.
Passage de NULLABLE à NOT NULL.
Pour un exemple d'activation du mode EVOLVE, consultez Activation du mode EVOLVE.
Si la table Paimon aval existe déjà, la tâche écrit les données dans le schéma de table existant et ne tente pas de recréer la table. Si le schéma de table existant diffère du schéma de la table amont, les colonnes acceptant les valeurs NULL qui existent dans la table amont mais sont absentes de la table aval sont automatiquement ajoutées à la table existante. Si un cas ne peut pas être adapté automatiquement, tel qu'un type de colonne incompatible ou une colonne NOT NULL manquante ou redondante, la vérification échoue et une erreur est signalée. Vous pouvez exécuter les instructions SQL recommandées dans le message d'erreur pour modifier le schéma de table. Vous pouvez également définir sink.existing-table.schema-check.enabled sur false dans le récepteur (sink) pour ignorer la vérification et l'ajout automatique de colonnes, afin que les données soient écrites en fonction du schéma de la table aval.
Exemples
Les exemples suivants montrent des configurations dans des scénarios typiques.
Écriture dans un catalogue Rest
L'exemple suivant montre comment écrire des données dans Data Lake Formation (DLF) lorsque le catalogue Paimon est de type rest :
source:
type: mysql
name: MySQL Source
hostname: ${secret_values.mysql.hostname}
port: ${mysql.port}
username: ${secret_values.mysql.username}
password: ${secret_values.mysql.password}
tables: ${mysql.source.table}
server-id: 8601-8604
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
catalog.properties.metastore: rest
catalog.properties.uri: dlf_uri
catalog.properties.warehouse: your_warehouse
catalog.properties.token.provider: dlf
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: truePour plus d'informations sur les paramètres préfixés par catalog.properties, consultez Paramètres de configuration du catalogue Flink CDC.
Écriture dans un catalogue FileSystem
Exemple de configuration pour l'écriture dans Object Storage Service (OSS) avec un catalogue Paimon filesystem :
source:
type: mysql
name: MySQL Source
hostname: ${secret_values.mysql.hostname}
port: ${mysql.port}
username: ${secret_values.mysql.username}
password: ${secret_values.mysql.password}
tables: ${mysql.source.table}
server-id: 8601-8604
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
catalog.properties.metastore: filesystem
catalog.properties.warehouse: oss://default/test
catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
catalog.properties.fs.oss.accessKeyId: xxxxxxxx
catalog.properties.fs.oss.accessKeySecret: xxxxxxxx
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: truePour plus d'informations sur les paramètres préfixés par catalog.properties, consultez Créer un catalogue Filesystem Paimon.
Écriture dans une table partitionnée
Convertit le champ create_time de type TIMESTAMP dans la table amont en un champ DATE, qui est utilisé comme clé de partition de la table Paimon.
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.test_source_table
server-id: 5401-5499
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
transform:
- source-table: test_db.test_source_table
projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
primary-keys: id, create_time, partition_key
partition-keys: partition_key
description: add partition key
pipeline:
name: MySQL to Paimon PipelineSynchroniser une table unique
Nous vous recommandons de configurer des tâches de synchronisation pour une table unique lorsque les exigences en matière de performances en temps réel et de stabilité sont élevées. La synchronisation d'une table unique évite la complexité liée à la synchronisation de plusieurs tables. L'exemple suivant illustre la configuration :
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.test_source_table
server-id: 5401-5499
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
pipeline:
name: MySQL to Paimon PipelineSynchroniser une base de données entière
La synchronisation d'une base de données entière réduit le nombre de tâches à configurer et à exécuter, ce qui diminue les coûts d'exploitation et de maintenance. L'exemple suivant illustre la configuration :
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.\.*
server-id: 5401-5499
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
pipeline:
name: MySQL to Paimon PipelineSynchroniser une base de données entière et remplacer les noms de bases de données et de tables
Pour remplacer les noms de bases de données et de tables par lot, ajoutez la section route à une tâche de synchronisation de base de données entière. L'exemple suivant illustre la configuration :
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.\.*
server-id: 5401-5499
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
route:
# Synchronize all tables in the MySQL test_db database to the Paimon test_db2 database and retain the table names.
- source-table: test_db.\.*
sink-table: test_db2.<>
replace-symbol: <>
pipeline:
name: MySQL to Paimon PipelineFusionner des bases de données et des tables fragmentées
Pour écrire les données provenant de plusieurs tables fragmentées dans une seule table en aval, utilisez le modèle suivant :
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.user\.*
server-id: 5401-5499
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
route:
# All sharded tables in the MySQL test_db database are merged into a single Paimon test_db.user table.
- source-table: test_db.user\.*
sink-table: test_db.user
pipeline:
name: MySQL to Paimon PipelineAjouter des tables existantes lors du redémarrage d'une tâche
Pour ajouter des tables existantes à la synchronisation, définissez scan.newly-added-table.enabled sur true et redémarrez la tâche.
Si la tâche définit initialement scan.binlog.newly-added-table.enabled sur true pour capturer les nouvelles tables, ne définissez pas ensuite scan.newly-added-table.enabled sur true avant de redémarrer la tâche afin de capturer les tables existantes. Sinon, les données seront livrées plusieurs fois.
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.\.*
server-id: 5401-5499
scan.startup.mode: initial
# On job restart, check the new tables captured by the tables parameter and take a snapshot.
# Note: Must be used together with scan.startup.mode: initial.
scan.newly-added-table.enabled: true
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
pipeline:
name: MySQL to Paimon PipelineExclure des tables spécifiques lors de la synchronisation d'une base de données entière
Pour exclure des tables spécifiques, telles que des tables de test ou des tables provoquant des basculements répétés de la tâche, d'une tâche de synchronisation de base de données entière, utilisez la configuration suivante :
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.\.*
# Tables that match this regular expression are not synchronized.
tables.exclude: test_db.table1
server-id: 5401-5499
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
pipeline:
name: MySQL to Paimon PipelineSynchroniser une base de données entière vers des tables Paimon au format de fichier Lance
Lance est un format de stockage de données conçu pour les données vectorielles et multimodales. Définissez le paramètre de création de table file.format pour écrire les données dans des tables Paimon utilisant le format de fichier Lance. L'exemple suivant illustre la configuration :
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.\.*
server-id: 5401-5499
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
# Specify the file format of output files as lance
table.properties.file.format: lance
pipeline:
name: MySQL to Paimon PipelineActiver le mode EVOLVE
Le mode EVOLVE maintient une stricte cohérence entre le schéma de la table en aval et celui de la table en amont, y compris pour les opérations telles que la suppression de champs et de tables. Dans ce mode, si le système en aval ne peut pas appliquer tous les événements de modification de schéma, la tâche peut subir un basculement et ne pourra pas récupérer sans intervention. L'exemple suivant illustre la configuration de la tâche :
source:
type: mysql
name: MySQL Source
hostname: <yourHostname>
port: 3306
username: flink
password: ${secret_values.password}
tables: test_db.test_source_table
server-id: 5401-5499
#(Optional) Synchronize data from newly added tables during the incremental phase.
scan.binlog.newly-added-table.enabled: true
#(Optional) Synchronize table and field comments.
include-comments.enabled: true
#(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
scan.incremental.snapshot.unbounded-chunk-first.enabled: true
#(Optional) Enable parsing and filtering to accelerate data reads.
scan.only.deserialize.captured.tables.changelog.enabled: true
sink:
type: paimon
name: Paimon Sink
using.built-in-catalog: paimon_dlf_catalog
#(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
commit.user: your_job_name
#(Optional) Enable deletion vectors to improve read performance.
table.properties.deletion-vectors.enabled: true
pipeline:
name: MySQL to Paimon Pipeline
schema.change.behavior: evolveFAQ
Configuration du décalage du consommateur pour une table source Paimon
Pourquoi une tâche Paimon peut-elle échouer avec l'erreur « Heartbeat of TaskManager timed out » ?
Pourquoi une tâche Paimon peut-elle échouer avec l'erreur « File xxx not found, Possible causes » ?
La visibilité des données du connecteur Paimon dépend-elle des intervalles de point de contrôle ?