Realtime Compute for Apache Flink:Paramètres clés

Remarques d'utilisation

Avant le démarrage d'une tâche, si vous ne définissez pas explicitement ce paramètre dans la section Parameters, la mémoire gérée sert à mettre les données en mémoire tampon en mode mini-lot. L'une des conditions suivantes déclenche le calcul final et la sortie des données :

• Réception d'un message de watermark depuis l'opérateur MiniBatchAssigner.

• La mémoire gérée est pleine.

• Avant le début d'un point de contrôle (checkpoint).

• Arrêt de la tâche.

Valeurs disponibles

• -1 (par défaut) : Indique que la mémoire gérée sert à mettre les données en mémoire tampon.

• Autres valeurs Long négatives : Identique au paramètre par défaut.

• Autres valeurs Long positives : Indique que la mémoire heap sert à mettre les données en mémoire tampon. Lorsque le nombre d'enregistrements en mémoire tampon atteint cette valeur (N), le système déclenche automatiquement l'opération de sortie.

Exemple

1. Créez une tâche de streaming SQL, copiez le code SQL suivant et deploy la tâche.

   CREATE TEMPORARY TABLE s1 (
     a INT,
     b INT,
     ts TIMESTAMP(3),
     PRIMARY KEY (a) NOT ENFORCED,
     WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
   ) WITH (
     'connector'='datagen',
     'rows-per-second'='1',
     'fields.ts.kind'='random',
     'fields.ts.max-past'='5s',
     'fields.b.kind'='random',
     'fields.b.min'='0',
     'fields.b.max'='10'
   );
   CREATE TEMPORARY TABLE sink (
     a INT,
     b BIGINT,
     PRIMARY KEY (a) NOT ENFORCED
   ) WITH (
     'connector'='print'
   );
   INSERT INTO sink SELECT a, sum(b) FROM s1 GROUP BY a;

2. Sur l'onglet Configuration de la page Deployments , dans le champ Other Configuration de la section Parameters , définissez les paramètres table.exec.mini-batch.enabled: true et table.exec.mini-batch.allow-latency: 2s , et ne définissez pas table.exec.mini-batch.size afin d'utiliser sa valeur par défaut (-1).

3. Start la tâche. Sous l'onglet Status , la topologie de la tâche inclut les opérateurs MiniBatchAssigner, LocalGroupAggregate et GlobalGroupAggregate.

Remarques d'utilisation

• Ce paramètre est effectif uniquement dans VVR 8.0.8 et versions ultérieures. Dans les versions antérieures à VVR 8.0.8, le comportement équivaut à la définition de ce paramètre sur false.

• Lors du changement de la valeur de false à true, la quantité de données envoyées en aval depuis les opérateurs MinibatchGlobalAgg et MinibatchAgg peut augmenter, ce qui exerce une pression accrue sur les opérateurs en aval.

Valeurs disponibles

• false (par défaut) : Lorsque state TTL est activé, les opérateurs MinibatchGlobalAgg et MinibatchAgg n'émettent pas de données en aval si le résultat de l'agrégation ne change pas.

• true : Lorsque state TTL est activé, les opérateurs MinibatchGlobalAgg et MinibatchAgg émettent toujours des enregistrements mis à jour (en double) en aval, même si le résultat de l'agrégation ne change pas.

Exemple

1. Créez une tâche de streaming SQL, copiez le code SQL suivant et deploy la tâche.

   create temporary table src(
       a int,
       b string
   ) with (
       'connector' = 'datagen',
       'rows-per-second' = '10',
       'fields.a.min' = '1',
       'fields.a.max' = '1',
       'fields.b.length' = '3'
   );
   create temporary table snk(
       a int,
       max_length_b bigint
   ) with (
       'connector' = 'blackhole'
   );
   insert into snk select a, max(CHAR_LENGTH(b)) from src group by a; 

2. Dans la section Other Configuration de la zone Parameters de l'onglet Configuration de la page Deployments , définissez les paramètres table.exec.mini-batch.enabled: true et table.exec.mini-batch.allow-latency: 2s pour activer l'optimisation Minibatch Aggregate.

3. Start la tâche. Sous l'onglet Status , la tâche inclut un opérateur MinibatchGlobalAggregate. Cliquez sur le signe « + » de l'opérateur pour constater que l'opérateur GlobalGroupAggregate n'envoie pas de données en aval lorsque le résultat de l'agrégation est inchangé.

   L'opérateur affiche RecordsIn à 19 et RecordsOut à 1, ce qui signifie que 19 enregistrements d'entrée n'ont produit qu'une seule sortie agrégée.

4. Arrêtez la tâche et ajoutez le paramètre table.exec.agg.mini-batch.output-identical-enabled: true à Other Configuration dans la section Parameters de la page Configuration de la page Deployments .

5. Start la tâche. Sous l'onglet Status , vous pouvez voir que la tâche inclut un opérateur MinibatchGlobalAggregate. Cliquez sur le signe « + » de l'opérateur pour observer que l'opérateur GlobalGroupAggregate envoie désormais des données en aval, même lorsque le résultat de l'agrégation est inchangé. Après le redémarrage de la tâche, l'onglet Status indique que les valeurs RecordsIn et RecordsOut de l'opérateur GlobalGroupAggregate sont toutes deux de 94. Cela indique qu'avec table.exec.agg.mini-batch.output-identical-enabled: true activé, l'opérateur envoie des données en aval même si le résultat de l'agrégation ne change pas.

Cas d'utilisation

• Utilisez cette optimisation pour préserver l'ordre de traitement par clé dans une jointure de table de dimension lorsque le flux contient peu de messages avec la même clé de mise à jour au fil du temps.

• Dans un flux CDC (Change Data Capture) avec une clé primaire, vous effectuez un enrichissement des données via une jointure de table de dimension et écrivez dans un récepteur dont la clé primaire correspond à la clé primaire source. La clé de jointure pour la jointure de table de dimension est différente de la clé primaire, et la clé de jointure côté dimension est la clé primaire. Cette optimisation effectue un brassage par la clé primaire CDC, qui est dérivée en tant que clé de mise à jour. Par rapport à l'activation de l'optimisation SHUFFLE_HASH pour le même scénario, elle évite la génération d'un opérateur SinkMaterializer avant le récepteur à un parallélisme plus élevé. Cela permet d'éviter les problèmes de performance potentiels liés à cet opérateur, notamment l'état volumineux qui peut s'accumuler lors d'exécutions longues. Pour plus d'informations sur SinkUpsertMaterializer, consultez Recommendations .

• La clé de jointure pour la jointure de table de dimension est différente de la clé primaire, la clé de jointure côté dimension est la clé primaire, et un opérateur de rang suit. Cette optimisation effectue un brassage par la clé primaire CDC, qui est dérivée en tant que clé de mise à jour. Par rapport à l'activation de l'optimisation SHUFFLE_HASH pour le même scénario, elle empêche UpdateFastRank de se dégrader en RetractRank. Pour savoir comment optimiser RetractRank en UpdateFastRank, consultez TopN optimization techniques .

Remarques d'utilisation

• Si le flux ne possède pas de clé de mise à jour, la ligne entière est utilisée comme clé.

• Le débit diminue si la même clé de mise à jour est fréquemment mise à jour sur une courte période, car les enregistrements pour une même clé de mise à jour sont traités dans un ordre strict.

• Par rapport à la jointure de table de dimension asynchrone d'origine, le mode Key-Ordered introduit un état clé. L'activation ou la désactivation de ce mode affecte la compatibilité de l'état.

• Cette fonctionnalité prend effet uniquement pour VVR 8.0.10 et versions ultérieures lorsque l'entrée de la jointure de table de dimension est un flux de mises à jour et que vous configurez table.exec.async-lookup.output-mode='ALLOW_UNORDERED' et table.exec.async-lookup.key-ordered-enabled='true' .

Valeurs disponibles

• false (par défaut) : Désactive le mode Key-Ordered.

• true : Active le mode Key-Ordered.

Exemple

1. L'exemple suivant utilise une jointure de table de dimension Hologres asynchrone. Créez une tâche de streaming SQL, copiez le code SQL suivant et deploy la tâche.

   Pour plus d'informations sur le connecteur Hologres, consultez Hologres .

   create TEMPORARY table bid_source(
     auction  BIGINT,
     bidder  BIGINT,
     price  BIGINT,
     channel  VARCHAR,
     url  VARCHAR,
     dateTime  TIMESTAMP(3),
     extra  VARCHAR,
     proc_time as proctime(),
     WATERMARK FOR dateTime AS dateTime - INTERVAL '4' SECOND
   ) with (
     'connector' = 'kafka',  -- A non-insert-only stream connector
     'topic' = 'user_behavior',
     'properties.bootstrap.servers' = 'localhost:9092',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'earliest-offset',
     'format' = 'csv'
   );
   CREATE TEMPORARY TABLE users (
       user_id STRING PRIMARY KEY NOT ENFORCED, -- Define the primary key
       user_name VARCHAR(255) NOT NULL, 
       age INT NOT NULL
   ) WITH (
       'connector' = 'hologres',    -- A connector that supports asynchronous lookup
       'async' = 'true',
       'dbname' = 'holo db name', --The name of your Hologres database
       'tablename' = 'schema_name.table_name', --The name of the Hologres table to receive data
       'username' = 'access id', --The AccessKey ID of your Alibaba Cloud account
       'password' = 'access key', --The AccessKey Secret of your Alibaba Cloud account
       'endpoint' = 'holo vpc endpoint', --The VPC endpoint of your Hologres instance
   );
   CREATE TEMPORARY TABLE bh ( 
       auction  BIGINT,
       age int
   ) WITH (
       'connector' = 'blackhole'
   );
   insert into bh
   SELECT
       bid_source.auction,
       u.age
   FROM bid_source
       JOIN users FOR SYSTEM_TIME AS OF bid_source.proc_time AS u
       ON bid_source.channel = u.user_id;
   

2. Sur la page Deployments , sous l'onglet Configuration , dans la section Other Configuration de la zone Parameters , définissez les paramètres table.exec.async-lookup.output-mode='ALLOW_UNORDERED' et table.exec.async-lookup.key-ordered-enabled='true' .

3. Start la tâche. Sous l'onglet Status , vous pouvez voir que l'attribut async de la tâche est KEY_ORDERED:true.

table.optimizer.window-join-enabled

Ce paramètre contrôle l'activation des opérations de window join . Lorsqu'il est activé, Flink optimise le plan d'exécution correspondant en tant que window join . Pour les petites fenêtres, cela réduit la surcharge d'état et améliore les performances. Par rapport à une regular join , une window join peut également éviter d'émettre des messages de mise à jour vers les opérateurs en aval, ce qui est utile pour les cas d'utilisation nécessitant des jointures avec des conditions de fenêtre temporelle réduite.