Tous les produits
Search

Ce produit

    Realtime Compute for Apache Flink:Tables à clé primaire et tables en ajout seul

    Centre de documentation

    Realtime Compute for Apache Flink:Tables à clé primaire et tables en ajout seul

    Dernière mise à jour :Aug 09, 2026

    Apache Paimon prend en charge deux types de tables : les tables à clé primaire et les tables en ajout seul. Chaque type possède une sémantique distincte pour les écritures, les mises à jour et la consommation en streaming. Choisissez le type adapté aux exigences de votre pipeline de données.

    Choisir un type de table

    Scénario Configuration recommandée
    Synchronisation CDC, upserts standard Table à clé primaire, moteur deduplicate
    Déduplication où seul le premier enregistrement importe Table à clé primaire, moteur first-row
    Agrégation en temps réel (totaux cumulés, suivi des max/min) Table à clé primaire, moteur aggregation
    Assemblage de tables larges à partir de plusieurs sources de streaming Table à clé primaire, moteur partial-update
    Ingestion de logs, streaming d'événements Table en ajout seul (évolutive)
    Remplacement de file d'attente de messages préservant l'ordre Table de file d'attente en ajout

    Tables à clé primaire

    Une table à clé primaire nécessite une ou plusieurs clés primaires, spécifiées lors de sa création. Les enregistrements partageant les mêmes clés primaires sont fusionnés selon le moteur de fusion configuré.

    CREATE TABLE T (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket' = '4'
);

    Dans cet exemple, la clé de partition est dt, les clés primaires sont dt, shop_id et user_id. La table utilise 4 buckets fixes.

    Modes de bucket

    Le bucket est la plus petite unité de lecture et d'écriture dans une table Paimon. Les partitions (ou la table entière pour les tables non partitionnées) sont subdivisées en buckets, ce qui permet des lectures et des écritures parallèles.

    Mode Configuration Notes
    Mode de bucket dynamique (par défaut) Omettre bucket ou définir 'bucket' = '-1' Ne prend pas en charge les écritures concurrentes depuis plusieurs déploiements Flink ; prend en charge les mises à jour inter-partitions
    Mode de bucket fixe Définir 'bucket' = '<num>' (entier > 0) <num> correspond au nombre de buckets pour toute la table (non partitionnée) ou par partition (partitionnée) ; prend en charge la modification du nombre de buckets

    Dimensionnement des buckets : Nous recommandons de limiter la taille totale des données dans chaque bucket à 2 Go et de ne pas dépasser 5 Go. Un nombre insuffisant de buckets limite le parallélisme d'écriture, tandis qu'un nombre excessif génère trop de petits fichiers.

    Mode de bucket dynamique : mise à jour des données

    Les mises à jour inter-partitions se produisent lorsque les clés primaires n'incluent pas toutes les clés de partition. Paimon ne pouvant pas déterminer la partition cible à partir de la seule clé primaire, il utilise RocksDB pour maintenir une correspondance entre la clé primaire et la partition/bucket. Sur les grandes tables, cela peut réduire considérablement les performances par rapport au mode de bucket fixe. L'initialisation du déploiement prend également plus de temps, le temps que la correspondance soit chargée dans RocksDB.

    Le comportement de mise à jour dépend du moteur de fusion :

    • deduplicate : supprime l'enregistrement existant et insère un nouvel enregistrement dans la partition cible.

    • aggregation ou partial-update : met à jour l'enregistrement existant sur place dans la partition actuelle.

    • first-row : conserve l'enregistrement existant et ignore l'enregistrement entrant.

    Les mises à jour intra-partition se produisent lorsque les clés primaires incluent toutes les clés de partition. Paimon peut déterminer la partition à partir de la clé primaire, mais pas le bucket. Il maintient donc un index en mémoire mappant les clés primaires aux buckets. Chaque tranche de 100 millions d'entrées de mappage consomme environ 1 Go de mémoire heap, imputée uniquement aux partitions en cours d'écriture active.

    Mode de bucket dynamique : attribution des buckets

    Les nouveaux enregistrements sont d'abord écrits dans les buckets existants. Si tous les buckets sont saturés, un nouveau bucket est créé automatiquement.

    Configurez ce comportement avec les paramètres suivants dans la clause WITH :

    Paramètre Description Valeur par défaut
    dynamic-bucket.target-row-num Nombre maximal d'enregistrements par bucket 2000000
    dynamic-bucket.initial-buckets Nombre initial de buckets Parallélisme de l'opérateur d'écriture

    Mode de bucket fixe : attribution des buckets

    Par défaut, Paimon attribue les enregistrements aux buckets en hachant les valeurs de clé primaire. Pour utiliser un autre ensemble de colonnes, définissez bucket-key dans la clause WITH. Spécifiez plusieurs colonnes séparées par des virgules. Les colonnes de bucket-key doivent être un sous-ensemble des clés primaires.

    Par exemple, 'bucket-key' = 'c1,c2' achemine les enregistrements selon le hachage de c1 et c2.

    En mode de bucket fixe, incluez toutes les clés de partition dans les clés primaires pour éviter les mises à jour inter-partitions.

    Modifier le nombre de buckets en mode de bucket fixe

    Le nombre de buckets détermine le parallélisme de lecture et d'écriture. Pour le modifier sur une table existante :

    1. Suspendez tous les déploiements qui lisent ou écrivent dans la table.

    2. Créez un script et exécutez l'instruction SQL suivante pour configurer le paramètre de bucket :

      ALTER TABLE `<catalog-name>`.`<database-name>`.`<table-name>` SET ('bucket' = '<bucket-num>');

    3. Réorganisez les données en exécutant un déploiement par lots.

      • Table non partitionnée : Créez un Blank Batch Draft, collez le code SQL suivant, puis cliquez sur Deploy et Start : ``sql INSERT OVERWRITE <catalog-name>.<database-name>.<table-name> SELECT * FROM <catalog-name>.<database-name>.<table-name>; ``

      • Table partitionnée : Exécutez un lot par partition à réorganiser :

        INSERT OVERWRITE `<catalog-name>`.`<database-name>`.`<table-name>`
PARTITION (<partition-spec>)
SELECT * FROM `<catalog-name>`.`<database-name>`.`<table-name>`
WHERE <partition-condition>;

        Exemple — réorganisation de la partition où dt = '20240312' et hh = '08' :

        INSERT OVERWRITE `<catalog-name>`.`<database-name>`.`<table-name>`
PARTITION (dt = '20240312', hh = '08')
SELECT * FROM `<catalog-name>`.`<database-name>`.`<table-name>`
WHERE dt = '20240312' AND hh = '08';

    4. Une fois le déploiement par lots terminé avec succès, reprenez les déploiements suspendus.

    Moteurs de fusion

    Lorsque plusieurs enregistrements partagent les mêmes clés primaires, Paimon les fusionne selon le paramètre merge-engine.

    Si votre flux d'entrée contient des enregistrements hors ordre, configurez le paramètre sequence.field pour contrôler l'ordre de fusion. Consultez la section Gestion des données hors ordre .

    deduplicate (par défaut)

    Paimon conserve uniquement l'enregistrement le plus récent pour chaque ensemble de clés primaires, en ignorant les précédents. Si le dernier enregistrement est une suppression (DELETE), tous les enregistrements ayant ces clés primaires sont retirés.

    CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'deduplicate' -- Optional: deduplicate is the default
);

    Exemples de résultats :

    • Écriture de +I(1, 2.0, 'apple'), +I(1, 4.0, 'banana'), +I(1, 8.0, 'cherry') — la requête renvoie (1, 8.0, 'cherry').

    • Écriture de +I(1, 2.0, 'apple'), +I(1, 4.0, 'banana'), -D(1, 4.0, 'banana') — la requête ne renvoie aucune ligne.

    À utiliser pour : Synchronisation CDC, pipelines d'upsert standard.

    first-row

    Paimon conserve uniquement le premier enregistrement pour chaque ensemble de clés primaires. Étant donné que les consommateurs en aval ne voient que des événements de journal des modifications de type INSERT, la production de journaux est plus efficace qu'avec deduplicate.

    CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'first-row'
);

    Exemple de résultat :

    Écriture de +I(1, 2.0, 'apple'), +I(1, 4.0, 'banana'), +I(1, 8.0, 'cherry') — la requête renvoie (1, 2.0, 'apple').

    Remarques d'utilisation :

    • Définissez changelog-producer sur lookup pour activer la consommation en streaming.

    • Le moteur first-row ne traite pas les modifications DELETE et UPDATE_BEFORE. Pour les ignorer au lieu de générer une erreur, définissez 'first-row.ignore-delete' = 'true'.

    • Les champs de séquence ne sont pas pris en charge.

    À utiliser pour : Pipelines de déduplication où seul le premier enregistrement rencontré importe.

    aggregation

    Paimon applique une fonction d'agrégation à chaque colonne non-clé primaire sur les enregistrements partageant les mêmes clés primaires. Spécifiez la fonction par colonne en utilisant fields.<field-name>.aggregate-function. Les colonnes sans fonction explicite utilisent par défaut last_non_null_value.

    CREATE TABLE T (
  product_id BIGINT,
  price DOUBLE,
  sales BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'merge-engine' = 'aggregation',
  'fields.price.aggregate-function' = 'max',
  'fields.sales.aggregate-function' = 'sum'
);

    Exemple de résultat :

    Écriture de +I(1, 23.0, 15) et +I(1, 30.2, 20) — la requête renvoie (1, 30.2, 35).

    Fonctions d'agrégation prises en charge :

    Fonction Types pris en charge
    sum DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE
    product DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE
    count INTEGER, BIGINT
    max, min CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ
    first_value, last_value Tous les types, y compris null
    first_not_null_value, last_non_null_value Tous les types
    listagg STRING
    bool_and, bool_or BOOLEAN
    Seules les fonctions sum , product et count prennent en charge les modifications de rétraction (UPDATE_BEFORE et DELETE). Pour ignorer les modifications de rétraction pour une colonne spécifique, définissez 'fields.<field-name>.ignore-retract' = 'true' .

    Définissez changelog-producer sur lookup ou full-compaction pour activer la consommation en streaming.

    À utiliser pour : Métriques d'agrégation en temps réel, totaux cumulés, suivi des valeurs maximales/minimales.

    partial-update

    Paimon met à jour des colonnes individuelles d'un enregistrement existant en utilisant la dernière valeur non nulle des enregistrements entrants partageant les mêmes clés primaires. Les valeurs nulles n'écrasent pas les valeurs existantes.

    CREATE TABLE T (
  k INT,
  v1 DOUBLE,
  v2 BIGINT,
  v3 STRING,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update'
);

    Exemple de résultat :

    Écriture de +I(1, 23.0, 10, NULL), +I(1, NULL, NULL, 'This is a book'), +I(1, 25.2, NULL, NULL) — la requête renvoie (1, 25.2, 10, 'This is a book').

    Remarques d'utilisation :

    • Définissez changelog-producer sur lookup ou full-compaction pour activer la consommation en streaming.

    • Le moteur partial-update ne traite pas les modifications DELETE et UPDATE_BEFORE. Pour les ignorer, définissez 'partial-update.ignore-delete' = 'true'.

    Groupes de séquence

    Lors de l'assemblage d'une table large à partir de plusieurs tables amont, utilisez des groupes de séquence pour contrôler indépendamment l'ordre de mise à jour de différents ensembles de colonnes, en gérant les données hors ordre provenant de diverses sources.

    Dans l'exemple suivant, les colonnes a et b sont mises à jour dans l'ordre croissant de g_1, et les colonnes c et d sont mises à jour dans l'ordre croissant de g_2 :

    CREATE TABLE T (
  k INT,
  a STRING,
  b STRING,
  g_1 INT,
  c STRING,
  d STRING,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence-group' = 'a,b',
  'fields.g_2.sequence-group' = 'c,d'
);

    Combinez les groupes de séquence avec des fonctions d'agrégation en ajoutant fields.<field-name>.aggregate-function pour toute colonne du groupe. L'exemple suivant suit la valeur non nulle la plus récente de a, la valeur maximale de b, la valeur non nulle la plus récente de c et la somme de d :

    CREATE TABLE T (
  k INT,
  a STRING,
  b INT,
  g_1 INT,
  c STRING,
  d INT,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence-group' = 'a,b',
  'fields.b.aggregate-function' = 'max',
  'fields.g_2.sequence-group' = 'c,d',
  'fields.d.aggregate-function' = 'sum'
);

    Pour plus d'informations, consultez la section Moteur de fusion.

    À utiliser pour : Assemblage de tables larges en joignant des colonnes provenant de plusieurs sources de streaming.

    Gestion des données hors ordre

    Par défaut, Paimon fusionne les enregistrements dans l'ordre d'arrivée : le dernier enregistrement reçu est le dernier à être fusionné. Pour les flux contenant des enregistrements hors ordre, définissez sequence.field dans la clause WITH. Les enregistrements partageant les mêmes clés primaires sont alors fusionnés dans l'ordre croissant de la valeur de la colonne spécifiée.

    Types de données pris en charge pour sequence.field : TINYINT, SMALLINT, INTEGER, BIGINT, TIMESTAMP, TIMESTAMP_LTZ.

    Lors de l'utilisation de MySQL comme source d'entrée avec la colonne de métadonnées op_t comme champ de séquence, les paires de modifications UPDATE_BEFORE et UPDATE_AFTER partagent la même valeur de séquence. Pour garantir que Paimon traite UPDATE_BEFORE avant UPDATE_AFTER, définissez 'sequence.auto-padding' = 'row-kind-flag' .

    Producteur de journal des modifications

    Une table à clé primaire doit générer un journal des modifications complet (couvrant les opérations INSERT, DELETE et UPDATE) pour prendre en charge la consommation en streaming en aval, à l'instar d'un journal binaire de base de données. Configurez la méthode de génération avec le paramètre changelog-producer.

    Valeur Description Latence Utilisation des ressources À utiliser lorsque
    none Aucun journal des modifications n'est généré. — La plus faible Aucune consommation en streaming nécessaire
    input Transmet directement les enregistrements d'entrée aux consommateurs en aval. La plus faible La plus faible L'entrée contient déjà un journal des modifications complet, tel qu'un journal binaire de base de données. Option la plus efficace : aucun calcul supplémentaire requis.
    lookup (recommandé) Exécute une recherche sur les résultats de compaction des petits fichiers pour produire un journal des modifications complet. Déclenché à chaque point de contrôle Flink. Niveau minute Plus élevée Tout type d'entrée ; exigences de latence au niveau de la minute
    full-compaction Génère un journal des modifications après chaque compaction complète des petits fichiers. Jusqu'à plusieurs heures Inférieure à lookup Tout type d'entrée ; une latence plus élevée est acceptable

    Pour full-compaction, définissez 'full-compaction.delta-commits' = '<num>' pour déclencher une compaction complète après chaque <num> points de contrôle Flink. La compaction complète est gourmande en ressources : définissez l'intervalle entre 30 minutes et 1 heure.

    Par défaut, Paimon génère un enregistrement de journal des modifications même lorsque la valeur mise à jour est identique à la précédente. Pour supprimer ces enregistrements sans effet, définissez 'changelog-producer.row-deduplicate' = 'true' . Cette option s'applique uniquement à lookup et full-compaction et nécessite un calcul supplémentaire pour comparer les valeurs avant/après ; activez-la uniquement lorsqu'un grand nombre d'enregistrements inutiles est attendu.

    Tables en ajout seul

    Une table en ajout seul ne possède pas de clés primaires et n'accepte que les opérations INSERT en mode streaming. Utilisez ce type de table pour les charges de travail ne nécessitant pas de mises à jour en streaming, telles que la synchronisation de données de journal.

    CREATE TABLE T (
  dt STRING,
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) PARTITIONED BY (dt) WITH (
  'bucket' = '-1'
);

    Sous-types

    Sous-type Configuration Caractéristiques
    Table évolutive en ajout 'bucket' = '-1' Débit d'écriture élevé ; aucun partitionnement par hachage requis ; tri des données pris en charge ; configuration flexible du parallélisme ; prise en charge de la conversion directe depuis les tables Hive ; compaction de fichiers complètement asynchrone ; l'ordre de consommation diffère de l'ordre d'écriture ; skew des données possible lorsque le parallélisme amont est égal au parallélisme de l'writer
    Table de file d'attente en ajout 'bucket' = '<num>' (entier > 0) Préservation de l'ordre par bucket ; latence de quelques minutes ; équivalent à une partition de topic Kafka ou au nombre de shards dans une instance ApsaraMQ for MQTT

    Attribution des buckets

    Table évolutive en ajout : Les données sont écrites directement dans une seule partition en parallèle. Le concept de bucket est ignoré et aucun partitionnement par hachage n'est requis, ce qui maximise les performances d'écriture.

    Table de file d'attente en ajout : Par défaut, les enregistrements sont attribués aux buckets en fonction des valeurs de toutes les colonnes de l'enregistrement de données. Pour contrôler l'attribution, définissez bucket-key dans la clause WITH avec une liste de noms de colonnes séparés par des virgules.

    Par exemple, 'bucket-key' = 'c1,c2' achemine les enregistrements selon le hachage de c1 et c2.

    Spécifiez bucket-key pour réduire le calcul d'attribution des buckets et améliorer l'efficacité d'écriture.

    Ordre de consommation des données

    Table évolutive en ajout : Les enregistrements peuvent être consommés dans un ordre différent de l'ordre d'écriture.

    Table de file d'attente en ajout : Les enregistrements au sein de chaque bucket sont consommés dans l'ordre d'écriture. Pour les enregistrements provenant de différentes partitions ou buckets :

    • Deux enregistrements provenant de partitions différentes : si 'scan.plan-sort-partition' = 'true', l'enregistrement ayant la valeur de partition la plus petite est consommé en premier ; sinon, l'enregistrement de la partition créée le plus tôt est consommé en premier.

    • Deux enregistrements provenant de la même partition, même bucket : l'enregistrement écrit le plus tôt est consommé en premier.

    • Deux enregistrements provenant de la même partition, buckets différents : l'ordre de consommation n'est pas garanti car différents buckets peuvent être traités simultanément.

    Étapes suivantes