Flink sql 完成CEP操作地址

我会带着你远行 2023-07-24 14:40 4阅读 0赞

[https://help.aliyun.com/document\_detail/73845.html][https_help.aliyun.com_document_detail_73845.html]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

## 示例 ##

*  示例语法
    
        SELECT *
        FROM Ticker MATCH_RECOGNIZE (
        PARTITION BY symbol
        ORDER BY tstamp
        MEASURES STRT.tstamp AS start_tstamp,
        LAST(DOWN.tstamp) AS bottom_tstamp,
        LAST(UP.tstamp) AS end_tstamp
        ONE ROW PER MATCH
        AFTER MATCH SKIP TO NEXT ROW
        PATTERN (STRT DOWN+ UP+) WITHIN INTERVAL '10' SECOND
        DEFINE
        DOWN AS DOWN.price < PREV(DOWN.price),
        UP AS UP.price > PREV(UP.price)
        );
 *  测试数据
    
    <table>  &nbsp;  
     <thead> 
      <tr> 
       <th>timestamp（TIMESTAMP）</th> 
       <th>card_id(VARCHAR)</th> 
       <th>location（VARCHAR）</th> 
       <th>action（VARCHAR）</th> 
      </tr> 
     </thead> 
     <tbody> 
      <tr> 
       <td><code>2018-04-13 12:00:00</code></td> 
       <td>1</td> 
       <td>Beijing</td> 
       <td>Consumption</td> 
      </tr> 
      <tr> 
       <td><code>2018-04-13 12:05:00</code></td> 
       <td>1</td> 
       <td>Shanghai</td> 
       <td>Consumption</td> 
      </tr> 
      <tr> 
       <td><code>2018-04-13 12:10:00</code></td> 
       <td>1</td> 
       <td>Shenzhen</td> 
       <td>Consumption</td> 
      </tr> 
      <tr> 
       <td><code>2018-04-13 12:20:00</code></td> 
       <td>1</td> 
       <td>Beijing</td> 
       <td>Consumption</td> 
      </tr> 
     </tbody> 
    </table>
 *  测试案例语法
    
        CREATE TABLE datahub_stream (
            `timestamp`               TIMESTAMP,
            card_id                   VARCHAR,
            location                  VARCHAR,
            `action`                  VARCHAR,
            WATERMARK wf FOR `timestamp` AS withOffset(`timestamp`, 1000)
        ) WITH (
            type = 'datahub'
            ...
        );
        CREATE TABLE rds_out (
            start_timestamp               TIMESTAMP,
            end_timestamp                 TIMESTAMP,
            card_id                       VARCHAR,
            event                         VARCHAR
        ) WITH (
            type= 'rds'
            ...
        );
        
        --案例描述：当相同的card_id在十分钟内，在两个不同的location发生刷卡现象，就会触发报警机制，以便监测信用卡盗刷等现象。
        
        --定义计算逻辑。
        insert into rds_out
        select 
        `start_timestamp`, 
        `end_timestamp`, 
        card_id, `event`
        from datahub_stream
        MATCH_RECOGNIZE (
            PARTITION BY card_id   --按card_id分区，将相同卡号的数据分到同一个计算节点上。
            ORDER BY `timestamp`   --在窗口内，对事件时间进行排序。
            MEASURES               --定义如何根据匹配成功的输入事件构造输出事件。
                e2.`action` as `event`,                
                e1.`timestamp` as `start_timestamp`,   --第一次的事件时间为start_timestamp。
                LAST(e2.`timestamp`) as `end_timestamp`--最新的事件时间为end_timestamp。
            ONE ROW PER MATCH           --匹配成功输出一条。
            AFTER MATCH SKIP TO NEXT ROW--匹配后跳转到下一行。
            PATTERN (e1 e2+) WITHIN INTERVAL '10' MINUTE  -- 定义两个事件，e1和e2。
            DEFINE                     --定义在PATTERN中出现的patternVariable的具体含义。
                e1 as e1.action = 'Consumption',    --事件一的action标记为Consumption。
                e2 as e2.action = 'Consumption' and e2.location <> e1.location --事件二的action标记为Consumption，且事件一和事件二的location不一致。
        );
 *  测试结果
    
    <table>  &nbsp;  
     <thead> 
      <tr> 
       <th>start_timestamp（TIMESTAMP）</th> 
       <th>end_timestamp（TIMESTAMP）</th> 
       <th>card_id（VARCHAR）</th> 
       <th>event（VARCHAR）</th> 
      </tr> 
     </thead> 
     <tbody> 
      <tr> 
       <td><code>2018-04-13 12:00:00.0</code></td> 
       <td><code>2018-04-13 12:05:00.0</code></td> 
       <td>1</td> 
       <td>Consumption</td> 
      </tr> 
      <tr> 
       <td><code>2018-04-13 12:05:00.0</code></td> 
       <td><code>2018-04-13 12:10:00.0</code></td> 
       <td>1</td> 
       <td>Consumption</td> 
      </tr> 
     </tbody> 
    </table>

[https_help.aliyun.com_document_detail_73845.html]: https://help.aliyun.com/document_detail/73845.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20230528/6e14137bd51a44a5b99e7c66cb8e1343.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20230528/58b813b7e0d044a9b305a4173fe08c40.png