Tous les produits
Search

Ce produit

    Realtime Compute for Apache Flink:Flink CEP dynamique

    Centre de documentation

    Realtime Compute for Apache Flink:Flink CEP dynamique

    Dernière mise à jour :Aug 12, 2026

    Realtime Compute for Apache Flink prend en charge les tâches Flink CEP avec des règles mises à jour dynamiquement dans les jobs DataStream. Cette rubrique illustre la création d'un job Flink CEP qui charge dynamiquement des règles pour traiter les données issues d'un topic Kafka en amont, à travers un scénario de marketing en temps réel.

    Cas d'usage

    Grâce à son architecture distribuée, sa latence de l'ordre de la milliseconde et ses puissantes capacités d'expression de règles, Flink CEP se prête à de nombreuses applications. Trois scénarios typiques sont :

    • Contrôle des risques en temps réel : Flink CEP permet d'identifier les utilisateurs à risque. Par exemple, il peut analyser les journaux de comportement des clients pour signaler les utilisateurs effectuant plus de 10 virements totalisant plus de 10 000 en moins de 5 minutes.

    • Marketing en temps réel : Flink CEP optimise les stratégies marketing. En analysant les journaux de comportement utilisateur lors d'une promotion e-commerce, vous pouvez par exemple identifier les personnes ajoutant plus de trois articles à leur panier en 10 minutes sans finaliser l'achat, ce qui permet d'ajuster le ciblage marketing. Flink CEP s'avère également efficace dans les scénarios de lutte contre la fraude liés au marketing en temps réel.

    • Internet des objets (IoT) : Flink CEP détecte les états anormaux et envoie des alertes. Il peut ainsi émettre une alerte de risque si un vélo en libre-service sort d'une zone autorisée et n'y revient pas dans les 15 minutes. Combiné à des capteurs IoT, il détecte aussi les anomalies sur les lignes de production. Si un capteur de température signale par exemple des valeurs dépassant un seuil défini pendant trois périodes consécutives, une alerte se déclenche automatiquement.

    Exemple détaillé

    Cette rubrique montre comment utiliser le CEP dynamique pour répondre à ces scénarios. Dans cet exemple, les journaux de comportement client sont stockés dans ApsaraMQ for Kafka. Le job Flink CEP consomme ces données tout en interrogeant une table de règles dans une base de données ApsaraDB RDS for MySQL. Il récupère les dernières règles ajoutées par un administrateur de politiques et les utilise pour faire correspondre les événements. Lorsqu'une correspondance est trouvée, le job envoie une alerte ou écrit les informations pertinentes dans un autre magasin de données. La figure suivante illustre le pipeline de données global.

    Ce tutoriel commence par démarrer le job Flink CEP, puis insère la Règle 1. Celle-ci correspond à une séquence où trois événements consécutifs avec une action égale à 0 sont suivis d'un événement dont l'action n'est pas 1. Cela signifie qu'un utilisateur a consulté un produit trois fois sans effectuer d'achat.

    Prérequis

    Procédure

    Cette rubrique décrit comment écrire et mettre à jour dynamiquement un job Flink CEP qui surveille et enregistre les utilisateurs dont les journaux de comportement correspondent à des règles spécifiques.

    Étape 1 : Préparer les données de test

    Préparer le topic Kafka en amont

    1. Connectez-vous à la console ApsaraMQ for Kafka.

    2. Créez un topic nommé demo_topic pour stocker les journaux de comportement utilisateur simulés.

      Pour plus d'informations, consultez Étape 1 : Créer un topic.

    Préparer la base de données RDS

    Dans la console Data Management (DMS), préparez les données de test pour ApsaraDB RDS for MySQL.

    1. Connectez-vous à l'instance ApsaraDB RDS for MySQL avec un compte privilégié.

      Pour plus d'informations, consultez Se connecter à une instance ApsaraDB RDS for MySQL via DMS.

    2. Créez la table de règles rds_demo pour stocker les règles du job Flink CEP. Créez ensuite la table match_results pour conserver les données correspondant aux règles.

      Dans la fenêtre SQLConsole active, saisissez les commandes suivantes et cliquez sur Execute.

      CREATE DATABASE cep_demo_db;
USE cep_demo_db;
CREATE TABLE rds_demo (
  `id` VARCHAR(64),
  `version` INT,
  `pattern` VARCHAR(4096),
  `function` VARCHAR(512)
);
CREATE TABLE match_results (
    rule_id INT,
    rule_version INT,
    user_id INT,
    user_name VARCHAR(255),
    production_id INT,
    PRIMARY KEY (rule_id,rule_version,user_id,production_id)
);

      Chaque ligne de la table de règles rds_demo représente une règle unique. Elle comprend un id et une version pour distinguer les différentes règles et leurs versions, un champ pattern décrivant l'objet pattern de l'API CEP, ainsi qu'un champ function indiquant comment traiter la séquence d'événements correspondant au modèle.

      Chaque ligne de la table match_results correspond à une occurrence où le comportement d'un utilisateur pour un produit spécifique respecte une règle donnée. Cet enregistrement permet d'élaborer des stratégies commerciales adaptées, comme l'envoi de coupons pour des produits associés.

    Étape 2 : Configurer la liste d'autorisation d'adresses IP

    Pour permettre au job Flink d'accéder à l'instance ApsaraDB RDS for MySQL, ajoutez le bloc CIDR de l'espace de travail Realtime Compute for Apache Flink à la liste d'autorisation d'adresses IP de l'instance.

    1. Obtenez le bloc CIDR VPC de l'espace de travail Realtime Compute for Apache Flink.

      1. Connectez-vous à la console Realtime Compute for Apache Flink.

      2. Dans la colonne Actions de l'workspace cible, choisissez More > Workspace details.

      3. Dans la boîte de dialogue Workspace details, consultez le CIDR block du service Fully Managed Flink.

    2. Ajoutez le bloc CIDR de Fully Managed Flink à la liste d'autorisation d'adresses IP de votre instance ApsaraDB RDS for MySQL.

      Pour plus d'informations, consultez Configurer une liste d'autorisation d'adresses IP. Dans la boîte de dialogue Modify whitelist, saisissez le bloc CIDR Flink dans la zone de texte IP addresses in whitelist. Si vous avez plusieurs blocs CIDR, séparez-les par des virgules. Cliquez ensuite sur OK.

    Étape 3 : Développer et démarrer le job CEP

    Remarque

    Tout le code de cette rubrique est disponible dans notre référentiel GitHub. À des fins de démonstration, l'exemple de code présenté ici est légèrement modifié sur la branche timeOrMoreAndWindow. Vous pouvez télécharger le fichier complet ververica-cep-demo-master.zip pour référence.

    1. Ajoutez flink-cep comme dépendance de projet dans le fichier POM Maven du job.

      Pour plus d'informations sur la gestion des autres packages JAR liés à Flink et la résolution des conflits, consultez Configurer les dépendances de l'environnement Flink.

      <dependency>
    <groupId>com.alibaba.ververica</groupId>
    <artifactId>flink-cep</artifactId>
    <version>1.17-vvr-8.0.8</version>
    <scope>provided</scope>
</dependency>

    2. Développez le code du job.

      1. Construisez une source Kafka.

        Pour plus de détails sur l'écriture du code, consultez Connecteur Kafka DataStream.

      2. Utilisez l'API CEP.dynamicPatterns().

        Pour prendre en charge les modifications dynamiques de règles et la correspondance multi-règles pour le CEP, Realtime Compute for Apache Flink définit l'API CEP.dynamicPatterns(). Cette API est définie comme suit.

        public static <T, R> SingleOutputStreamOperator<R> dynamicPatterns(
         DataStream<T> input,
         PatternProcessorDiscovererFactory<T> discovererFactory,
         TimeBehaviour timeBehaviour,
         TypeInformation<R> outTypeInfo)

        Le tableau suivant décrit les paramètres de cette API. Vous pouvez adapter ces valeurs selon votre cas d'utilisation réel.

        Paramètre

        Description

        DataStream<T> input

        Flux d'événements en entrée.

        PatternProcessorDiscovererFactory<T> discovererFactory

        Fabrique créant un PatternProcessorDiscoverer. Ce découvreur récupère les règles les plus récentes et construit les instances PatternProcessor correspondantes.

        TimeBehaviour timeBehaviour

        Décrit la manière dont le job Flink CEP gère les attributs temporels des événements. Valeurs valides :

        • TimeBehaviour.ProcessingTime : Traite les événements selon le temps de traitement.

        • TimeBehaviour.EventTime : Traite les événements selon le temps d'événement.

        TypeInformation<R> outTypeInfo

        Décrit les informations de type du flux de sortie.

        Pour plus d'informations sur les concepts Flink courants tels que DataStream, TimeBehaviour et TypeInformation, consultez API DataStream, Temps d'événement et temps de traitement et TypeInformation.

        L'interface PatternProcessor constitue un composant clé. Un PatternProcessor contient un Pattern spécifique décrivant comment faire correspondre les événements, ainsi qu'une PatternProcessFunction définissant le traitement à appliquer en cas de correspondance, comme l'envoi d'une alerte. Il inclut également un id et une version pour identifier le PatternProcessor. Pour plus de contexte, consultez la proposition.

        Le patternProcessorDiscovererFactory crée un découvreur pour récupérer le dernier PatternProcessor. L'exemple de code inclut une classe abstraite montrant comment interroger périodiquement un stockage externe pour obtenir de nouvelles instances de PatternProcessor.

        public abstract class PeriodicPatternProcessorDiscoverer<T>
        implements PatternProcessorDiscoverer<T> {
    ...
    @Override
    public void discoverPatternProcessorUpdates(
            PatternProcessorManager<T> patternProcessorManager) {
        // Periodically discovers the pattern processor updates.
        timer.schedule(
                new TimerTask() {
                    @Override
                    public void run() {
                        if (arePatternProcessorsUpdated()) {
                            List<PatternProcessor<T>> patternProcessors = null;
                            try {
                                patternProcessors = getLatestPatternProcessors();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                            patternProcessorManager.onPatternProcessorsUpdated(patternProcessors);
                        }
                    }
                },
                0,
                intervalMillis);
    }
    ...
}

        Realtime Compute for Apache Flink fournit une implémentation de JDBCPeriodicPatternProcessorDiscoverer pour récupérer les règles les plus récentes depuis une base de données compatible avec le protocole JDBC, telle que ApsaraDB RDS for MySQL ou Hologres. Lors de son utilisation, vous devez spécifier les paramètres suivants.

        Paramètre

        Description

        jdbcUrl

        URL JDBC de la base de données.

        jdbcDriver

        Nom de classe du pilote de base de données.

        tableName

        Nom de la table de base de données.

        initialPatternProcessors

        PatternProcessor par défaut à utiliser lorsque la table de règles de la base de données est vide.

        intervalMillis

        Intervalle d'interrogation de la base de données, en millisecondes.

        Vous pouvez l'utiliser dans votre code comme suit. Le job imprimera les règles correspondantes dans la sortie du TaskManager Flink.

        // import ......
public class CepDemo {
    public static void main(String[] args) throws Exception {
        ......
        // DataStream Source
        DataStreamSource<Event> source =
                env.fromSource(
                        kafkaSource,
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner((event, ts) -> event.getEventTime()),
                        "Kafka Source");
        env.setParallelism(1);
        // Key by userId and productionId.
        // Note: Only events with the same key will be processed to see if there is a match.
        KeyedStream<Event, Tuple2<Integer, Integer>> keyedStream =
                source.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forGenerator(ctx -> new EventBoundedOutOfOrdernessWatermarks(Duration.ofSeconds(5)))
                ).keyBy(new KeySelector<Event, Tuple2<Integer, Integer>>() {
                    @Override
                    public Tuple2<Integer, Integer> getKey(Event value) throws Exception {
                        return Tuple2.of(value.getId(), value.getProductionId());
                    }
                });
        SingleOutputStreamOperator<String> output =
                CEP.dynamicPatterns(
                        keyedStream,
                        new JDBCPeriodicPatternProcessorDiscovererFactory<>(
                                params.get(JDBC_URL_ARG),
                                JDBC_DRIVE,
                                params.get(TABLE_NAME_ARG),
                                null,
                                Long.parseLong(params.get(JDBC_INTERVAL_MILLIS_ARG))),
                        Boolean.parseBoolean(params.get(USING_EVENT_TIME)) ? TimeBehaviour.EventTime : TimeBehaviour.ProcessingTime,
                        TypeInformation.of(new TypeHint<String>() {}));
        output.print();
        // Compile and submit the job
        env.execute("CEPDemo");
    }
}
        Remarque

        À des fins de démonstration, le code de l'exemple partitionne le flux de données d'entrée par id et product_id avant de le connecter à CEP.dynamicPatterns(). Cela signifie que seuls les événements partageant le même id et le même product_id sont pris en compte pour la correspondance des règles. Les événements ayant des clés différentes ne seront pas comparés entre eux.

    3. Dans la console Realtime Compute for Apache Flink, téléchargez le fichier JAR et déployez le job JAR. Pour plus d'informations, consultez Déployer un job.

      Pour démarrer rapidement, vous pouvez télécharger le fichier de test cep-demo.jar. Le tableau suivant décrit les paramètres à configurer lors du déploiement.

      Remarque

      Comme la source Kafka en amont est vide et que la table de règles de la base de données ne contient aucune donnée, le job ne produira aucune sortie après son démarrage.

      Paramètre

      Description

      Deployment mode

      Sélectionnez Stream Mode.

      Deployment name

      Saisissez un nom pour le job JAR.

      Engine version

      Pour plus d'informations sur les versions du moteur, consultez Versions du moteur et Politiques de cycle de vie. Nous recommandons d'utiliser une version recommandée ou stable. Les balises de version sont décrites ci-dessous :

      • Version recommandée : Dernière version mineure de la version majeure actuelle.

      • Version stable : Dernière version mineure d'une version majeure encore dans sa période de support produit et dont les défauts historiques ont été corrigés.

      • Version normale : Autres versions mineures encore dans leur période de support produit.

      • Version EOS : Version ayant dépassé sa période de support produit.

      JAR URL

      Téléchargez votre fichier JAR packagé ou le fichier JAR de test que nous fournissons.

      Entry Point Class

      Saisissez com.alibaba.ververica.cep.demo.CepDemo.

      Entry Point Main Arguments

      Si vous utilisez un job développé par vos soins et avez déjà configuré les informations de stockage en amont et en aval, laissez ce champ vide. En revanche, si vous utilisez le JAR de test fourni, vous devez configurer ce paramètre. Le code est le suivant.

      --kafkaBrokers YOUR_KAFKA_BROKERS 
--inputTopic YOUR_KAFKA_TOPIC 
--inputTopicGroup YOUR_KAFKA_TOPIC_GROUP 
--jdbcUrl jdbc:mysql://YOUR_DB_URL:port/DATABASE_NAME?user=YOUR_USERNAME&password=YOUR_PASSWORD
--tableName YOUR_TABLE_NAME  
--jdbcIntervalMs 3000
--usingEventTime false

      Le tableau suivant décrit ces paramètres.

      • kafkaBrokers : Adresse du broker Kafka.

      • inputTopic : Nom du topic Kafka.

      • inputTopicGroup : Groupe de consommateurs Kafka.

      • jdbcUrl : URL JDBC de la base de données.

        Remarque

        Le nom d'utilisateur et le mot de passe dans l'URL JDBC de cet exemple doivent correspondre à un compte standard, et le mot de passe ne peut contenir que des lettres et des chiffres. Dans un scénario réel, vous pouvez utiliser différentes méthodes d'authentification dans votre job selon vos besoins.

      • tableName : Nom de la table cible.

      • jdbcIntervalMs : Intervalle d'interrogation de la base de données.

      • usingEventTime : Spécifie s'il faut utiliser le temps d'événement pour le traitement (true/false).

      Remarque

      • Remplacez les valeurs de remplacement par vos informations réelles de stockage en amont et en aval.

      • Évitez d'utiliser des mots de passe en clair dans un environnement de production. Nous recommandons d'utiliser la fonctionnalité de gestion des variables. Pour plus d'informations, consultez Gestion des variables.

    4. Sur la page Configuration, dans la section Other configuration, ajoutez les paramètres d'exécution du job suivants.

      Dans les applications pratiques, le JAR flink-cep est chargé par le chargeur de classes système, tandis que les classes liées à aviator sont généralement packagées dans un JAR utilisateur et chargées par le chargeur de classes utilisateur. Grâce aux deux configurations ci-dessous, vous garantissez que le chargeur de classes système puisse accéder aux classes du JAR utilisateur lorsqu'il tente de charger une classe, évitant ainsi les échecs de chargement.

      kubernetes.application-mode.classpath.include-user-jar: 'true' 
classloader.resolve-order: parent-first

      Pour plus d'informations sur la configuration des paramètres d'exécution, consultez Configuration des paramètres d'exécution.

    5. Sur la page O&M > Deployments, localisez le déploiement cible et cliquez sur Start dans la colonne Actions.

      Pour plus d'informations sur la configuration des paramètres de démarrage du job, consultez Démarrer un job.

    Étape 4 : Insérer une règle

    Le job Flink CEP étant en cours d'exécution, insérez la Règle 1 : après trois événements consécutifs avec une action égale à 0, l'action de l'événement suivant n'est pas 1. Cela signifie qu'un utilisateur consulte un produit trois fois sans effectuer d'achat.

    1. Connectez-vous à la console ApsaraDB RDS for MySQL.

    2. Insérez la règle de mise à jour dynamique.

      Concaténez la chaîne JSON avec l'id, la version et le nom de la classe de fonction, puis insérez-la dans RDS.

      INSERT INTO rds_demo (
 `id`,
 `version`,
 `pattern`,
 `function`
) values(
  '1',
   1,
  '{"name":"end","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":null,"nodes":[{"name":"end","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.EndCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"start","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["LOOPING"],"times":{"from":3,"to":3,"windowTime":null},"untilCondition":null},"condition":{"expression":"action == 0","type":"AVIATOR"},"type":"ATOMIC"}],"edges":[{"source":"start","target":"end","type":"SKIP_TILL_NEXT"}],"window":null,"afterMatchStrategy":{"type":"SKIP_PAST_LAST_EVENT","patternName":null},"type":"COMPOSITE","version":1}',
  'com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction')
;

      Pour améliorer l'utilisabilité et la lisibilité du champ pattern dans la base de données, Realtime Compute for Apache Flink définit un format de règle basé sur JSON. Pour plus d'informations, consultez Format JSON pour les règles dans le CEP dynamique. Le champ pattern dans l'instruction SQL précédente contient une chaîne JSON sérialisée. Cette chaîne représente un modèle correspondant à la séquence suivante : après trois événements consécutifs avec une action égale à 0, l'action de l'événement suivant n'est pas 1.

      Remarque

      Dans le code EndCondition, la condition définie est action != 1.

      • La description correspondante de l'API CEP est la suivante.

        Pattern<Event, Event> pattern =
    Pattern.<Event>begin("start", AfterMatchSkipStrategy.skipPastLastEvent())
        .where(new StartCondition("action == 0"))
        .timesOrMore(3)
        .followedBy("end")
        .where(new EndCondition());

      • Vous pouvez la convertir en chaîne JSON correspondante en utilisant la méthode dans CepJsonUtils.

        public void printTestPattern(Pattern<?, ?> pattern) throws JsonProcessingException {
    System.out.println(CepJsonUtils.convertPatternToJSONString(pattern));
}

      • La chaîne JSON correspondante est la suivante.

        Exemple de chaîne JSON de règle CEP dynamique

        {
  "name": "end",
  "quantifier": {
    "consumingStrategy": "SKIP_TILL_NEXT",
    "properties": [
      "SINGLE"
    ],
    "times": null,
    "untilCondition": null
  },
  "condition": null,
  "nodes": [
    {
      "name": "end",
      "quantifier": {
        "consumingStrategy": "SKIP_TILL_NEXT",
        "properties": [
          "SINGLE"
        ],
        "times": null,
        "untilCondition": null
      },
      "condition": {
        "className": "com.alibaba.ververica.cep.demo.condition.EndCondition",
        "type": "CLASS"
      },
      "type": "ATOMIC"
    },
    {
      "name": "start",
      "quantifier": {
        "consumingStrategy": "SKIP_TILL_NEXT",
        "properties": [
          "LOOPING"
        ],
        "times": {
          "from": 3,
          "to": 3,
          "windowTime": null
        },
        "untilCondition": null
      },
      "condition": {
        "expression": "action == 0",
        "type": "AVIATOR"
      },
      "type": "ATOMIC"
    }
  ],
  "edges": [
    {
      "source": "start",
      "target": "end",
      "type": "SKIP_TILL_NEXT"
    }
  ],
  "window": null,
  "afterMatchStrategy": {
    "type": "SKIP_PAST_LAST_EVENT",
    "patternName": null
  },
  "type": "COMPOSITE",
  "version": 1
}

    3. Envoyez des messages au topic demo_topic à l'aide d'un client Kafka.

      Dans cette démonstration, vous pouvez également utiliser la page Start to send and consume message fournie par ApsaraMQ for Kafka pour envoyer des messages de test.

      1,Ken,0,1,1662022777000
1,Ken,0,1,1662022778000
1,Ken,0,1,1662022779000
1,Ken,0,1,1662022780000

      Choisissez la méthode d'envoi Console, saisissez 1 dans le champ Message key, collez les données de test dans le champ Message content, réglez Send to specified partition sur No, puis envoyez le message. La page affiche une notification Message sent successfully.

      Le tableau suivant décrit les champs du topic demo_topic.

      Paramètre

      Description

      id

      Identifiant de l'utilisateur.

      username

      Nom d'utilisateur.

      action

      Action de l'utilisateur. Valeurs valides :

      • 0 : opération de consultation

      • 1 : action d'achat

      product_id

      Identifiant du produit.

      event_time

      Heure de l'événement de comportement.

    4. Consultez la dernière règle dans les journaux du JobManager et la correspondance dans les journaux du TaskManager.

      • Dans les journaux du JobManager, recherchez JDBCPeriodicPatternProcessorDiscoverer pour voir la dernière règle.

        Pour trouver les journaux, accédez à Logs > JobManager, puis cliquez sur l'onglet Logs. Saisissez le mot-clé dans la zone de recherche pour localiser l'entrée de journal pertinente. Le message de journal PatternProcessors have been updated confirme que la règle a été mise à jour avec succès.

      • Dans le fichier journal du TaskManager se terminant par .out, recherchez A match for Pattern of (id, version): (1, 1) pour voir la correspondance résultante.

        Sur la page de détails du déploiement, cliquez sur l'onglet Logs, sélectionnez Running task managers, puis ouvrez le sous-onglet Logs du TaskManager correspondant. Recherchez le mot-clé dans le fichier flink.out pour localiser la séquence d'événements correspondante.

    5. Interrogez la table match_results en exécutant SELECT * FROM pour voir les résultats correspondant à la règle.

      La requête retourne un enregistrement avec les champs rule_id, rule_version, user_id, user_name et production_id, ayant respectivement pour valeurs 1, 1, 1, Ken et 1.

    Étape 5 : Mettre à jour la règle de correspondance

    Les stratégies marketing ont souvent des contraintes temporelles. Cette étape met à jour la règle pour exiger que les trois événements action = 0 se produisent dans un intervalle de 15 minutes.

    1. Définissez le paramètre usingEventTime sur true.

      1. Sur la page O&M > Deployments, localisez le déploiement cible et cliquez sur Cancel dans la colonne Actions.

      2. Dans Configuration > Entry Point Main Arguments , cliquez sur Edit, définissez le paramètre usingEventTime sur true, puis cliquez sur Save.

      3. Cliquez à nouveau sur Start pour redémarrer le job.

    2. Insérez la nouvelle règle.

      La description correspondante de l'API CEP est la suivante.

      Pattern<Event, Event> pattern =
        Pattern.<Event>begin("start", AfterMatchSkipStrategy.skipPastLastEvent())
                .where(new StartCondition("action == 0"))
                .timesOrMore(3,Time.minutes(15))
                .followedBy("end")
                .where(new EndCondition());
printTestPattern(pattern);

      Insérez la nouvelle règle dans la table rds_demo.

      # To avoid rule conflicts for this demo, delete the previous rule first.
DELETE FROM `rds_demo` WHERE `id` = 1;
# Insert the new rule: three consecutive `action = 0` events within 15 minutes, followed by an event where the action is not 1. The rule version is (1, 2).
INSERT INTO rds_demo (`id`,`version`,`pattern`,`function`) values('1',2,'{"name":"end","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":null,"nodes":[{"name":"end","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.EndCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"start","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["LOOPING"],"times":{"from":3,"to":3,"windowTime":{"unit":"MINUTES","size":15}},"untilCondition":null},"condition":{"expression":"action == 0","type":"AVIATOR"},"type":"ATOMIC"}],"edges":[{"source":"start","target":"end","type":"SKIP_TILL_NEXT"}],"window":null,"afterMatchStrategy":{"type":"SKIP_PAST_LAST_EVENT","patternName":null},"type":"COMPOSITE","version":1}','com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction');

    3. Sur la console Kafka, envoyez huit messages pour déclencher une correspondance.

      Voici huit exemples de messages.

      2,Tom,0,1,1739584800000   #10:00
2,Tom,0,1,1739585400000   #10:10
2,Tom,0,1,1739585700000   #10:15
2,Tom,0,1,1739586000000   #10:20
3,Ali,0,1,1739586600000   #10:30
3,Ali,0,1,1739588400000   #11:00
3,Ali,0,1,1739589000000   #11:10
3,Ali,0,1,1739590200000   #11:30

    4. Interrogez la table match_results en exécutant SELECT * FROM pour voir les résultats correspondant à la règle.

      La requête retourne deux enregistrements, avec les colonnes rule_id, rule_version, user_id, user_name et production_id. Le premier enregistrement concerne Ken (rule_version=1) et le second Tom (rule_version=2), tous deux associés à production_id=1.

      Les résultats montrent que seul le comportement de Tom correspond à la nouvelle règle, car les actions d'Ali se sont étalées sur plus de 15 minutes. Pour les promotions à durée limitée, cela permet d'envoyer des coupons aux utilisateurs qui consultent régulièrement un produit dans un laps de temps spécifique, afin d'encourager l'achat.