All Products
Search

This Product

    Realtime Compute for Apache Flink:Dynamic Flink CEP

    Document Center

    Realtime Compute for Apache Flink:Dynamic Flink CEP

    Last Updated:Jun 22, 2026

    Realtime Compute for Apache Flink mendukung tugas Flink CEP dengan aturan yang diperbarui secara dinamis dalam pekerjaan DataStream. Artikel ini menggunakan skenario pemasaran real-time untuk menunjukkan cara membangun pekerjaan Flink CEP yang memuat aturan secara dinamis guna memproses data dari topik Kafka hulu.

    Kasus penggunaan

    Berkat arsitektur terdistribusinya, latensi tingkat milidetik, dan kemampuan ekspresi aturan yang kuat, Flink CEP sangat cocok untuk berbagai aplikasi. Tiga skenario khas adalah:

    • Kontrol risiko real-time: Flink CEP dapat digunakan untuk mengidentifikasi pengguna berisiko. Misalnya, sistem dapat menganalisis log perilaku pelanggan untuk menandai pengguna yang melakukan lebih dari 10 transfer dengan total lebih dari 10.000 dalam waktu 5 menit.

    • Pemasaran real-time: Flink CEP dapat mengoptimalkan strategi pemasaran. Misalnya, dengan menganalisis log perilaku pengguna selama promosi e-commerce, Anda dapat mengidentifikasi pengguna yang menambahkan lebih dari tiga produk ke keranjang belanja mereka dalam waktu 10 menit tetapi tidak menyelesaikan pembelian, sehingga memungkinkan penyesuaian pemasaran yang ditargetkan. Flink CEP juga efektif dalam skenario anti-penipuan untuk pemasaran real-time.

    • Internet of Things (IoT): Flink CEP dapat mendeteksi kondisi abnormal dan mengirim peringatan. Misalnya, sistem dapat mengeluarkan peringatan risiko jika sepeda bersama meninggalkan area yang ditentukan dan tidak kembali dalam waktu 15 menit. Sistem ini juga dapat dikombinasikan dengan sensor IoT untuk mendeteksi anomali pada garis produksi. Misalnya, jika sensor suhu terus-menerus melaporkan suhu yang melebihi ambang batas yang ditetapkan selama tiga periode waktu berturut-turut, peringatan akan dipicu.

    Contoh langkah demi langkah

    Artikel ini menunjukkan cara menggunakan CEP dinamis untuk mengatasi skenario tersebut. Dalam contoh ini, log perilaku pelanggan disimpan di ApsaraMQ for Kafka. Pekerjaan Flink CEP mengonsumsi data ini sambil melakukan polling tabel aturan di database ApsaraDB RDS for MySQL. Sistem mengambil aturan terbaru yang ditambahkan oleh administrator kebijakan dan menggunakannya untuk mencocokkan event. Ketika terjadi kecocokan, pekerjaan mengirim peringatan atau menulis informasi terkait ke penyimpanan data lain. Gambar berikut menunjukkan keseluruhan pipa data.

    Langkah demi langkah ini pertama-tama menjalankan pekerjaan Flink CEP, lalu menyisipkan Aturan 1, yang mencocokkan urutan tiga event berturut-turut dengan action bernilai 0 diikuti oleh event dengan action bukan 1. Hal ini menandakan pengguna melihat produk tiga kali tanpa melakukan pembelian.

    Prasyarat

    Prosedur

    Artikel ini menjelaskan cara menulis dan memperbarui secara dinamis pekerjaan Flink CEP yang memantau serta mencatat pengguna yang log perilakunya sesuai dengan aturan tertentu.

    Langkah 1: Siapkan data uji

    Siapkan topik Kafka hulu

    1. Masuk ke konsol ApsaraMQ for Kafka.

    2. Buat topik bernama demo_topic untuk menyimpan log perilaku pengguna simulasi.

      Untuk informasi selengkapnya, lihat Langkah 1: Buat topik.

    Siapkan database RDS

    Di konsol Data Management (DMS), siapkan data uji untuk ApsaraDB RDS for MySQL.

    1. Masuk ke instans ApsaraDB RDS for MySQL dengan akun istimewa.

      Untuk informasi selengkapnya, lihat Masuk ke instans ApsaraDB RDS for MySQL menggunakan DMS.

    2. Buat tabel aturan rds_demo untuk menyimpan aturan pekerjaan Flink CEP. Buat tabel match_results untuk menyimpan data yang sesuai dengan aturan.

      Di jendela SQLConsole aktif, masukkan perintah berikut dan klik 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)
);

      Setiap baris dalam tabel aturan rds_demo merepresentasikan satu aturan. Tabel ini mencakup id dan version untuk membedakan antara aturan dan versinya, bidang pattern yang menjelaskan objek pattern API CEP, serta bidang function yang menjelaskan cara memproses urutan event yang sesuai dengan pola tersebut.

      Setiap baris dalam tabel match_results merepresentasikan kecocokan di mana perilaku pengguna untuk produk tertentu sesuai dengan aturan tertentu. Catatan ini dapat digunakan untuk merumuskan strategi penjualan yang sesuai, seperti mengirim kupon untuk produk terkait.

    Langkah 2: Konfigurasikan daftar putih IP

    Untuk mengizinkan pekerjaan Flink mengakses instans ApsaraDB RDS for MySQL, tambahkan Blok CIDR ruang kerja Realtime Compute for Apache Flink ke daftar putih alamat IP instans tersebut.

    1. Dapatkan Blok CIDR VPC ruang kerja Realtime Compute for Apache Flink.

      1. Masuk ke konsol Realtime Compute for Apache Flink.

      2. Di kolom Actions workspace target, pilih More > Workspace details.

      3. Di kotak dialog Workspace details, lihat CIDR block vSwitch Flink yang sepenuhnya dikelola.

    2. Tambahkan Blok CIDR Flink yang sepenuhnya dikelola ke daftar putih alamat IP instans ApsaraDB RDS for MySQL Anda.

      Untuk informasi selengkapnya, lihat Konfigurasikan daftar putih alamat IP. Di kotak dialog Modify whitelist, masukkan Blok CIDR Flink di kotak teks IP addresses in whitelist. Jika Anda memiliki beberapa Blok CIDR, pisahkan dengan koma. Lalu, klik OK.

    Langkah 3: Kembangkan dan jalankan pekerjaan CEP

    Catatan

    Semua kode untuk artikel ini tersedia di repositori GitHub kami. Untuk tujuan demonstrasi, kode contoh dalam artikel ini sedikit dimodifikasi pada cabang timeOrMoreAndWindow. Anda dapat mengunduh file lengkap ververica-cep-demo-master.zip sebagai referensi.

    1. Tambahkan flink-cep sebagai dependensi proyek dalam file Maven POM pekerjaan.

      Untuk informasi selengkapnya tentang menangani paket JAR terkait Flink lainnya dan menyelesaikan konflik, lihat Konfigurasikan dependensi lingkungan Flink.

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

    2. Kembangkan kode pekerjaan.

      1. Bangun Kafka Source.

        Untuk detail penulisan kode, lihat Kafka DataStream Connector.

      2. Bangun API CEP.dynamicPatterns().

        Untuk mendukung perubahan aturan dinamis dan pencocokan multi-aturan untuk CEP, Realtime Compute for Apache Flink mendefinisikan API CEP.dynamicPatterns(). API ini didefinisikan sebagai berikut.

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

        Tabel berikut menjelaskan parameter untuk API ini. Anda dapat memperbarui nilai parameter berdasarkan kasus penggunaan aktual Anda.

        Parameter

        Deskripsi

        DataStream<T> input

        Aliran event masukan.

        PatternProcessorDiscovererFactory<T> discovererFactory

        Pabrik yang membuat PatternProcessorDiscoverer. Discoverer ini mengambil aturan terbaru dan membangun instance PatternProcessor yang sesuai.

        TimeBehaviour timeBehaviour

        Menjelaskan cara pekerjaan Flink CEP menangani atribut waktu event. Nilai yang valid:

        • TimeBehaviour.ProcessingTime: Memproses event berdasarkan waktu pemrosesan.

        • TimeBehaviour.EventTime: Memproses event berdasarkan waktu event.

        TypeInformation<R> outTypeInfo

        Menjelaskan informasi tipe aliran keluaran.

        Untuk informasi selengkapnya tentang konsep umum Flink seperti DataStream, TimeBehaviour, dan TypeInformation, lihat DataStream API, Event Time dan Processing Time, dan TypeInformation.

        Antarmuka PatternProcessor merupakan komponen utama. PatternProcessor berisi Pattern spesifik yang menjelaskan cara mencocokkan event, serta PatternProcessFunction yang menjelaskan cara menangani kecocokan, seperti mengirim peringatan. Komponen ini juga mencakup id dan version untuk mengidentifikasi PatternProcessor. Untuk latar belakang lebih lanjut, lihat proposal.

        patternProcessorDiscovererFactory membuat discoverer untuk mengambil PatternProcessor terbaru. Kode contoh mencakup kelas abstrak yang menunjukkan cara melakukan polling berkala ke penyimpanan eksternal untuk instance PatternProcessor baru.

        public abstract class PeriodicPatternProcessorDiscoverer<T>
        implements PatternProcessorDiscoverer<T> {
    ...
    @Override
    public void discoverPatternProcessorUpdates(
            PatternProcessorManager<T> patternProcessorManager) {
        // Melakukan polling berkala untuk memperbarui processor pola.
        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 menyediakan implementasi JDBCPeriodicPatternProcessorDiscoverer untuk mengambil aturan terbaru dari database yang mendukung protokol JDBC, seperti ApsaraDB RDS for MySQL atau Hologres. Saat menggunakannya, Anda perlu menentukan parameter berikut.

        Parameter

        Deskripsi

        jdbcUrl

        URL JDBC database.

        jdbcDriver

        Nama kelas driver database.

        tableName

        Nama tabel database.

        initialPatternProcessors

        PatternProcessor default yang digunakan saat tabel aturan di database kosong.

        intervalMillis

        Interval polling database, dalam milidetik.

        Dalam kode Anda, Anda dapat menggunakannya sebagai berikut. Pekerjaan akan mencetak aturan yang cocok ke output Flink TaskManager.

        // 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 berdasarkan userId dan productionId.
        // Catatan: Hanya event dengan key yang sama yang akan diproses untuk melihat apakah ada kecocokan.
        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();
        // Kompilasi dan kirim pekerjaan
        env.execute("CEPDemo");
    }
}
        Catatan

        Untuk tujuan demonstrasi, kode demo melakukan keying aliran data masukan berdasarkan id dan product_id sebelum menghubungkannya ke CEP.dynamicPatterns(). Artinya, hanya event dengan id dan product_id yang sama yang dipertimbangkan untuk pencocokan aturan. Event dengan key berbeda tidak akan dicocokkan satu sama lain.

    3. Di konsol Realtime Compute for Apache Flink, unggah file JAR dan terapkan pekerjaan JAR. Untuk informasi selengkapnya, lihat Terapkan pekerjaan.

      Untuk membantu Anda memulai dengan cepat, Anda dapat mengunduh file uji cep-demo.jar. Tabel berikut menjelaskan parameter yang harus dikonfigurasi selama penerapan.

      Catatan

      Karena sumber Kafka hulu kosong dan tabel aturan database tidak berisi data, pekerjaan tidak akan menghasilkan output apa pun setelah dijalankan.

      Parameter

      Deskripsi

      Deployment mode

      Pilih Stream Mode.

      Deployment name

      Masukkan nama untuk pekerjaan JAR.

      Engine version

      Untuk informasi selengkapnya tentang versi engine, lihat Versi engine dan Kebijakan siklus hidup. Kami menyarankan menggunakan versi yang direkomendasikan atau stabil. Tag versi dijelaskan sebagai berikut:

      • Versi yang direkomendasikan: Versi minor terbaru dari versi utama saat ini.

      • Versi stabil: Versi minor terbaru dari versi utama yang masih dalam periode dukungan produk dan telah memperbaiki cacat historis.

      • Versi normal: Versi minor lain yang masih dalam periode dukungan produk.

      • Versi EOS: Versi yang telah melewati periode dukungan produk.

      JAR URL

      Unggah file JAR yang telah Anda kemas, atau unggah file JAR uji yang kami sediakan.

      Entry Point Class

      Masukkan com.alibaba.ververica.cep.demo.CepDemo.

      Entry Point Main Arguments

      Jika Anda menggunakan pekerjaan yang dikembangkan sendiri dan telah mengonfigurasi informasi penyimpanan hulu dan hilir, Anda dapat mengosongkan bidang ini. Namun, jika Anda menggunakan file JAR uji yang disediakan, Anda harus mengonfigurasi parameter ini. Kodenya sebagai berikut.

      --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

      Tabel berikut menjelaskan parameter tersebut.

      • kafkaBrokers: Alamat broker Kafka.

      • inputTopic: Nama topik Kafka.

      • inputTopicGroup: Kelompok konsumen Kafka.

      • jdbcUrl: URL JDBC database.

        Catatan

        Username dan password dalam URL JDBC untuk contoh ini harus berasal dari akun standar, dan password hanya boleh berisi huruf dan angka. Dalam skenario dunia nyata, Anda dapat menggunakan metode autentikasi berbeda dalam pekerjaan Anda sesuai kebutuhan.

      • tableName: Nama tabel target.

      • jdbcIntervalMs: Interval polling database.

      • usingEventTime: Menentukan apakah akan menggunakan waktu event untuk pemrosesan (true/false).

      Catatan

      • Anda harus mengganti nilai placeholder dengan informasi penyimpanan hulu dan hilir aktual Anda.

      • Hindari menggunakan password teks biasa di lingkungan produksi. Kami menyarankan menggunakan fitur manajemen variabel. Untuk informasi selengkapnya, lihat Manajemen variabel.

    4. Di halaman Deployment details, pada bagian Other configuration, tambahkan parameter runtime pekerjaan berikut.

      Dalam aplikasi praktis, file JAR flink-cep dimuat oleh system class loader, sedangkan kelas terkait aviator biasanya dikemas dalam JAR pengguna dan dimuat oleh user class loader. Dengan menggunakan dua konfigurasi di bawah ini, Anda dapat memastikan bahwa system class loader dapat mengakses kelas dalam JAR pengguna saat mencoba memuat kelas, sehingga menghindari kegagalan pemuatan kelas.

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

      Untuk informasi selengkapnya tentang mengonfigurasi parameter runtime, lihat Konfigurasi parameter runtime.

    5. Di halaman O&M > Deployments, temukan penerapan target dan klik Start di kolom Actions.

      Untuk informasi selengkapnya tentang mengonfigurasi parameter startup pekerjaan, lihat Jalankan pekerjaan.

    Langkah 4: Sisipkan aturan

    Dengan pekerjaan Flink CEP sedang berjalan, sisipkan Aturan 1: setelah tiga event berturut-turut dengan action bernilai 0, action event berikutnya bukan 1. Artinya, pengguna melihat produk tiga kali tanpa melakukan pembelian.

    1. Masuk ke konsol ApsaraDB RDS for MySQL.

    2. Sisipkan aturan pembaruan dinamis.

      Gabungkan string JSON dengan id, version, dan nama kelas fungsi, lalu sisipkan ke 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')
;

      Untuk meningkatkan kegunaan dan keterbacaan bidang pattern di database, Realtime Compute for Apache Flink mendefinisikan format aturan berbasis JSON. Untuk informasi selengkapnya, lihat Format JSON untuk aturan dalam CEP dinamis. Bidang pattern dalam pernyataan SQL di atas berisi string JSON serialisasi. String ini merepresentasikan pola yang mencocokkan urutan berikut: setelah tiga event berturut-turut dengan action bernilai 0, action event berikutnya bukan 1.

      Catatan

      Dalam kode EndCondition, kondisi yang didefinisikan adalah action != 1.

      • Deskripsi API CEP yang sesuai adalah sebagai berikut.

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

      • Anda dapat mengonversinya ke string JSON yang sesuai dengan menggunakan metode dalam CepJsonUtils.

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

      • String JSON yang sesuai adalah sebagai berikut.

        Contoh string JSON aturan CEP dinamis 

        {
  "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. Kirim pesan ke topik demo_topic menggunakan klien Kafka.

      Dalam demo ini, Anda juga dapat menggunakan halaman Start to send and consume message yang disediakan oleh ApsaraMQ for Kafka untuk mengirim pesan uji.

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

      Pilih metode pengiriman Console, masukkan 1 di bidang Message key, tempel data uji ke bidang Message content, atur Send to specified partition ke No, lalu kirim pesan. Halaman akan menampilkan notifikasi Message sent successfully.

      Tabel berikut menjelaskan bidang dalam demo_topic.

      Parameter

      Deskripsi

      id

      ID pengguna.

      username

      Username.

      action

      Aksi pengguna. Nilai yang valid:

      • 0: operasi melihat

      • 1: aksi pembelian

      product_id

      ID produk.

      event_time

      Waktu kejadian perilaku.

    4. Lihat aturan terbaru di log JobManager dan kecocokan di log TaskManager.

      • Di log JobManager, cari JDBCPeriodicPatternProcessorDiscoverer untuk melihat aturan terbaru.

        Untuk menemukan log, navigasi ke Logs > JobManager, lalu klik tab Logs. Masukkan kata kunci di kotak pencarian untuk menemukan entri log yang relevan. Pesan log PatternProcessors have been updated mengonfirmasi bahwa aturan berhasil diperbarui.

      • Di file log TaskManager yang diakhiri dengan .out, cari A match for Pattern of (id, version): (1, 1) untuk melihat hasil kecocokan.

        Di halaman detail penerapan, klik tab Logs, pilih Running task managers, lalu buka subtab Logs untuk TaskManager yang sesuai. Cari kata kunci di file flink.out untuk menemukan urutan event yang cocok.

    5. Kueri tabel match_results dengan menjalankan SELECT * FROM `match_results`; untuk melihat hasil yang sesuai dengan aturan.

      Kueri mengembalikan satu catatan dengan bidang rule_id, rule_version, user_id, user_name, dan production_id, dengan nilai masing-masing 1, 1, 1, Ken, dan 1.

    Langkah 5: Perbarui aturan pencocokan

    Strategi pemasaran sering kali memiliki batasan waktu. Langkah ini memperbarui aturan agar ketiga event action = 0 harus terjadi dalam interval 15 menit.

    1. Atur parameter usingEventTime ke true.

      1. Di halaman O&M > Deployments, temukan penerapan target dan klik Cancel di kolom Actions.

      2. Di Deployment Details > Entry Point Main Arguments , klik Edit, atur parameter usingEventTime ke true, lalu klik Save.

      3. Start pekerjaan lagi.

    2. Sisipkan aturan baru.

      Deskripsi API CEP yang sesuai adalah sebagai berikut.

      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);

      Sisipkan aturan baru ke tabel rds_demo.

      # Untuk menghindari konflik aturan dalam demo ini, hapus aturan sebelumnya terlebih dahulu.
DELETE FROM `rds_demo` WHERE `id` = 1;
# Sisipkan aturan baru: tiga event berturut-turut `action = 0` dalam waktu 15 menit, diikuti oleh event dengan action bukan 1. Versi aturan adalah (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. Di konsol Kafka, kirim delapan pesan untuk memicu kecocokan.

      Berikut adalah delapan pesan contoh.

      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. Kueri tabel match_results dengan menjalankan SELECT * FROM `match_results`; untuk melihat hasil yang sesuai dengan aturan.

      Kueri mengembalikan dua catatan, dengan kolom rule_id, rule_version, user_id, user_name, dan production_id. Catatan pertama untuk Ken (rule_version=1), dan yang kedua untuk Tom (rule_version=2), keduanya terkait dengan production_id=1.

      Hasil menunjukkan bahwa hanya perilaku Tom yang sesuai dengan aturan baru karena aksi Ali terjadi lebih dari 15 menit. Untuk promosi berbatas waktu, hal ini memungkinkan Anda mengirim kupon kepada pengguna yang berulang kali mengunjungi produk dalam jangka waktu tertentu, mendorong pembelian.