flinksql进行相关计算模型

ゝ一纸荒年。 2022-12-15 03:23 49阅读 0赞

1，flinksql如下：

CREATE TABLE SourTable (
                          carid VARCHAR,
                          utime BIGINT,
                          accountId VARCHAR,
                          uuid VARCHAR,
                          upBytes BIGINT,
                          downBytes BIGINT,
                          usedBytes BIGINT,
                          bssid VARCHAR,
                          mac VARCHAR,
                          ip VARCHAR,
                          status INT,
                          uphone VARCHAR,
                          options VARCHAR
                         ) WITH (
                          'connector.type' = 'kafka',
                          'connector.property-version' = '1',
    		      'connector.version' = '0.10',	
                          'connector.topic' = 'flink_old_open_wifi_dwd',
                          'update-mode' = 'append',
                          'connector.properties.0.key' = 'bootstrap.servers',
                          'connector.properties.0.value' = '10.2.2.8:9092',
                          'connector.properties.1.key' = 'group.id',
                          'connector.properties.1.value' = 'flink-liqiang-flow',
                          'connector.properties.2.key' = 'key.deserializer',
                          'connector.properties.2.value' = 'org.apache.kafka.common.serialization.StringSerializer',
                          'connector.properties.3.key' = 'value.deserializer',
                          'connector.properties.3.value' = 'org.apache.kafka.common.serialization.StringSerializer',
                          'connector.startup-mode' = 'latest-offset',
                          'connector.sink-partitioner' = 'round-robin',
                          'format.type' = 'json',
                          'format.property-version' = '1',
                          'format.derive-schema' = 'true'
                        );
     CREATE TABLE SinkTable (
    		      biztime BIGINT,
                          bizDate VARCHAR,
                          mac VARCHAR,
                          phone VARCHAR
                         ) WITH (
                          'connector.type' = 'kafka-gt',
                          'connector.property-version' = '1',
                          'connector.topic' = 'flink_first_flow_drf_dws',
                          'update-mode' = 'append',
                          'connector.properties.0.key' = 'bootstrap.servers',
                          'connector.properties.0.value' = '10.2.2.8:9092',
                          'connector.properties.1.key' = 'group.id',
                          'connector.properties.1.value' = 'flink-liqiang-flow',
                          'connector.properties.2.key' = 'key.deserializer',
                          'connector.properties.2.value' = 'org.apache.kafka.common.serialization.StringSerializer',
                          'connector.properties.3.key' = 'value.deserializer',
                          'connector.properties.3.value' = 'org.apache.kafka.common.serialization.StringSerializer',
                          'connector.startup-mode' = 'latest-offset',
                          'connector.sink-partitioner' = 'round-robin',
                          'format.type' = 'json',
                          'format.property-version' = '1',
                          'format.derive-schema' = 'true'
                        );
    
    #主要对日期做处理，主要对时间进行 FROM_UNIXTIME 处理
    insert into SinkTable
    select min(utime) as biztime,FROM_UNIXTIME(utime/1000,'yyyy-MM-dd') as bizDate,mac,max(uphone) as phone  from SourTable
    where FROM_UNIXTIME(utime/1000,'yyyy-MM-dd')  = DATE_FORMAT(LOCALTIMESTAMP,'yyyy-MM-dd')  group by FROM_UNIXTIME(utime/1000,'yyyy-MM-dd'),mac having count(mac) = 1;