Почему Flink 1.7 CEP SQL (MATCH_RECOGNIZE) не может работать - PullRequest
Новая
       9

Почему Flink 1.7 CEP SQL (MATCH_RECOGNIZE) не может работать

0 голосов
/ 25 января 2019 
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

Мой Env следующим образом:

env yaml SQLClient: 

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: необходимо указать либо время строки, либо время выполнения заказа в качестве первого.

1 Ответ

0 голосов
/ 26 января 2019

в качестве основного порядка сообщения об исключении должен быть указан в поле атрибут времени .Вы можете проверить, как назначить преобразование поля в атрибут времени в этом примере .

Важная часть: 

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