flink proctime error

Thu, 21 May 2020 06:08:11 -0700 

error:Window aggregate can only be defined over a time attribute column, but 
TIMESTAMP(3) encountered.
??????sink_table 
????watermark????????????????????source????????????sink??????????????????group 
by??????????????error??
CREATE TABLE source_table (
        sip VARCHAR,
        proctime as proctime()
) WITH (
        'connector.type' = 'kafka',
        'connector.version' = 'universal',
        'connector.startup-mode' = 'latest-offset',
        'connector.topic' = 'skyeye-tcpflow',
        'connector.properties.group.id' = 'testGroup',
        'connector.properties.zookeeper.connect' = 'x.x.x.x:2181',
        'connector.properties.bootstrap.servers' = 'x.x.x.x:9092',
        'update-mode' = 'append',
        'format.type' = 'json',
        'format.derive-schema' = 'true'
);


CREATE TABLE sink_table (
        ip VARCHAR,
        proctime timestamp(3)


) WITH (
        'connector.type' = 'kafka',
        'connector.version' = 'universal',
        'connector.startup-mode' = 'latest-offset',
        'connector.topic' = 'ip_agg',
        'connector.properties.zookeeper.connect' = 'x.x.x.x:2181',
        'connector.properties.bootstrap.servers' = 'x.x.x.x:9092',
        'update-mode' = 'append',
        'format.type' = 'json',
        'format.derive-schema' = 'true'
);

insert into sink_kafka select sip,proctime from source_kafka;



select TUMBLE_START(proctime, INTERVAL '10' MINUTE),count(1) from sink_table 
group by TUMBLE(proctime, INTERVAL '10' MINUTE);

回复