[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: how to use the TUMBLE(time_attr, interval) window function in Flink SQL


The problem is that sTime is not a Time Attribute[1], which has to be aligned with watermarks mechanism. Right now you cannot create a time attribute from within TableFunction, as far as I know.

What you could do is to do the splitting logic in DataStream API and register a proper table with implemented watermarks in TableEnvironment. Then you can apply the windowing on a table prepared that way.




On 23/10/2018 06:45, maidangdang44 wrote:
below is one line of my source, the body containes the user logs:
    body: [
        "user1,url1,2018-10-23 00:00:00;user2,url2,2018-10-23 00:01:00;user3,url3,2018-10-23 00:02:00"

I user LATERAL TABLE and a User-Defined TableFunction flatmap the source to a new table log, and I want to group by the time and username, here is my code:

public class BodySplitFun extends TableFunction<Tuple3<String, String, Long>> {
private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        public void eval(Row bodyRow) {
            String body = bodyRow.getField(0).toString();
            String[] lines = body.split(";");
               for (String line : lines) {
        String user = line.split(",")[0];
        String url = "">
        String sTime = line.split(",")[2];
                        collect(new Tuple3<>(user, url, sdf.parse(sTime).getTime());

tblEnv.registerFunction("bodySplit", new BodySplitFun());
                     mySource LEFT JOIN LATERAL TABLE(bodySplit(body)) as T(username, url, sTime) ON TRUE
            group by 
               TUMBLE(log.sTime, INTERVAL '1' MINUTE), log.username");

when I run my program, I got these error message:
Caused by: org.apache.calcite.sql.validate.SqlValidatorException: Cannot apply 'TUMBLE' to arguments of type 'TUMBLE(<BIGINT>, <INTERVAL DAY>)'. Supported form(s): 'TUMBLE(<DATETIME>, <DATETIME_INTERVAL>)'
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(
        at java.lang.reflect.Constructor.newInstance(
        at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(
        at org.apache.calcite.runtime.Resources$ExInst.ex(
        ... 49 more

how can I group by the sTime in the table log?

Attachment: signature.asc
Description: OpenPGP digital signature