Flink Sql match_recognize returns different time (not as input event time)

Question

I have set of streaming data which has timestamp in it , after flink sql processing the return result shows different time which doesn't match with incoming (event time ) timestamp of data. Here is my data with code where last two rows represent output from flink sql which shows different time. DO we need to add any specific code to get correct time?

3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:00:20.0,Entered,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:00:20.0,Entered,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:05:10.0,Conveyer,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:05:10.0,Conveyer,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:10:13.0,Waiting,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:10:13.0,Waiting,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:30:09.0,Security Scan,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:30:09.0,Security Scan,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:50:54.0,Waiting,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 06:50:54.0,Waiting,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 07:00:34.0,Boarding,Airport Recipe,2020-07-07 12:32:10.0) 3> (1001,j001,Goa Vacation,ctx[],action[],NULL,2020-06-05 07:00:34.0,Boarding,Airport Recipe,2020-07-07 12:32:10.0) 4> (1001,2020-06-05 00:30:49.999,2020-06-05 00:35:49.999,Entered,Conveyer) 4> (1001,2020-06-05 00:31:39.999,2020-06-05 00:35:49.999,Entered,Conveyer)

Table matchResult = bsTableEnv
                .sqlQuery("Select Id, StartTime, EndTime, EventName, EventB from eb_user_journey "
                        + "MATCH_RECOGNIZE" + "( PARTITION BY user_id " + "ORDER BY event_time " + "MEASURES "
                        + "A.user_id AS Id, " + "FIRST(A.event_time) AS StartTime, "
                        + "FIRST(B.event_time) AS EndTime, A.event_name AS EventName, B.event_name AS EventB"
                        + " ONE ROW PER MATCH " + " AFTER MATCH SKIP TO NEXT ROW " + "  PATTERN (A+ B) " + " DEFINE"
                        + " A as A.event_name = 'Entered' , B as B.event_name = 'Conveyer')");
        matchResult.printSchema();

        TupleTypeInfo<Tuple5<String, Timestamp, Timestamp, String, String>> tupleType = new TupleTypeInfo<>(
                Types.STRING(), Types.SQL_TIMESTAMP(), Types.SQL_TIMESTAMP(), Types.STRING(), Types.STRING());

        DataStream<Tuple5<String, Timestamp, Timestamp, String, String>> cassandraDSTP = bsTableEnv
                .toAppendStream(matchResult, tupleType); `

I suspect timezone conversion between the timezone for India and UTC is part of the explanation, but that doesn't appear to be the only reason why the timestamps are being transformed. — David Anderson
Thanks @DavidAnderson for reply , I have verified the timezone it is converting India timezone to UTC . Can I know if their is any way to change this ? — IraSantuManu lifeblog

David Anderson David Anderson · Accepted Answer · 2020-07-07T11:48:54

Looks like something is converting the timestamps to UTC. You should be able to use one of these temporal functions in your select statement to convert back, perhaps LOCALTIMESTAMP or CONVERT_TZ.

Flink Sql match_recognize returns different time (not as input event time)

1 Answers