为什么 flink 1.7 CEP SQL(MATCH_RECOGNIZE) 不能工作

Question

Flink SQL> describe Ticker;
root
 |-- symbol: String
 |-- rowtime: Timestamp
 |-- price: Long


Flink SQL> SELECT *
> FROM Ticker
> MATCH_RECOGNIZE (
>     PARTITION BY symbol
>     ORDER BY rowtime
>     MEASURES
>         START_ROW.rowtime AS start_rowtime,
>         LAST(PRICE_DOWN.rowtime) AS bottom_rowtime,
>         LAST(PRICE_UP.rowtime) AS end_rowtime
>     ONE ROW PER MATCH
>     AFTER MATCH SKIP TO LAST PRICE_UP
>     PATTERN (START_ROW PRICE_DOWN+ PRICE_UP)
>     DEFINE
>         PRICE_DOWN AS
>             (LAST(PRICE_DOWN.price, 1) IS NULL AND PRICE_DOWN.price < START_ROW.price) OR
>                 PRICE_DOWN.price < LAST(PRICE_DOWN.price, 1),
>         PRICE_UP AS
>             PRICE_UP.price > LAST(PRICE_DOWN.price, 1)
>     ) MR;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: You must specify either rowtime or proctime for order by as the first one.

页面https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/table/streaming/match_recognize.html

我的环境如下：

SQLClient 的 env yaml：

tables:
  - name: Ticker
    type: source-table
    update-mode: append
    connector:
      type: filesystem
      path: "/home/leisore/flink/flink-1.7.1/table.csv"
    format:
      type: csv
      fields:
        - name: symbol
          type: STRING
        - name: rowtime
          type: TIMESTAMP
        - name: price
          type: LONG
      line-delimiter: "\n"
      comment-prefix: "#"
    schema:
        - name: symbol
          type: STRING
        - name: rowtime
          type: TIMESTAMP
        - name: price
          type: LONG

table.csv

ACME,2019-01-1,12
ACME,2019-01-2,17
ACME,2019-01-3,19
ACME,2019-01-4,21
ACME,2019-01-5,25
ACME,2019-01-6,12
ACME,2019-01-7,15
ACME,2019-01-8,20
ACME,2019-01-9,24
ACME,2019-01-10,25
ACME,2019-01-11,19
ACME,2019-01-12,15
ACME,2019-01-13,25
ACME,2019-01-14,25
ACME,2019-01-15,14
ACME,2019-01-16,12
ACME,2019-01-17,14
ACME,2019-01-18,24
ACME,2019-01-19,23
ACME,2019-01-20,22

Run SQLClient:
============================
./sql-client.sh embedded -e ../sqlenv.yaml 
>>Flink SQL> SELECT *
> FROM Ticker
> MATCH_RECOGNIZE (
>     PARTITION BY symbol
>     ORDER BY rowtime
>     MEASURES
>         START_ROW.rowtime AS start_rowtime,
>         LAST(PRICE_DOWN.rowtime) AS bottom_rowtime,
>         LAST(PRICE_UP.rowtime) AS end_rowtime
>     ONE ROW PER MATCH
>     AFTER MATCH SKIP TO LAST PRICE_UP
>     PATTERN (START_ROW PRICE_DOWN+ PRICE_UP)
>     DEFINE
>         PRICE_DOWN AS
>             (LAST(PRICE_DOWN.price, 1) IS NULL AND PRICE_DOWN.price < START_ROW.price) OR
>                 PRICE_DOWN.price < LAST(PRICE_DOWN.price, 1),
>         PRICE_UP AS
>             PRICE_UP.price > LAST(PRICE_DOWN.price, 1)
>     ) MR;
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.table.api.ValidationException: You must specify either rowtime or proctime for order by as the first one.

总是输出： [错误] 无法执行 SQL 语句。原因： org.apache.flink.table.api.ValidationException: 您必须将 order by 的 rowtime 或 proctime 指定为第一个。

Answer 1

因为必须在time-attribute 字段中指定异常消息的主要顺序。您可以在 example 中查看如何将字段转换为时间属性。

重要的是：

  - name: rowTime
    type: TIMESTAMP
    rowtime:
      timestamps:
        type: "from-field"
        from: "rideTime"
      watermarks:
        type: "periodic-bounded"
        delay: "60000"