指标统计：基于流计算Oceanus(Flink) 实现实时UVPV统计

# 0.10.X及以后版本advertised.listeners=PLAINTEXT://10.1.0.10:9092# 0.10.X之前版本advertised.host.name=PLAINTEXT://10.1.0.10:9092

./bin/kafka-console-producer.sh --broker-list 10.1.0.10:9092 --topic uvpv-demo>{"record_type":0, "user_id": 2, "client_ip": "100.0.0.2", "product_id": 101, "create_time": "2021-09-08 16:20:00"}>{"record_type":0, "user_id": 3, "client_ip": "100.0.0.3", "product_id": 101, "create_time": "2021-09-08 16:20:00"}>{"record_type":1, "user_id": 2, "client_ip": "100.0.0.1", "product_id": 101, "create_time": "2021-09-08 16:20:00"}

# 浏览记录{ "record_type":0,  # 0 表示浏览记录 "user_id": 6,  "client_ip": "100.0.0.6",  "product_id": 101,  "create_time": "2021-09-06 16:00:00"}
# 购买记录{ "record_type":1, # 1 表示购买记录 "user_id": 6,  "client_ip": "100.0.0.8",  "product_id": 101,  "create_time": "2021-09-08 18:00:00"}

CREATE TABLE `input_web_record` (  `record_type` INT,  `user_id` INT,  `client_ip` VARCHAR,  `product_id` INT,  `create_time` TIMESTAMP,  `times` AS create_time,  WATERMARK FOR times AS times - INTERVAL '10' MINUTE ) WITH (    'connector' = 'kafka',   -- 可选 'kafka','kafka-0.11'. 注意选择对应的内置 Connector    'topic' = 'uvpv-demo',      'scan.startup.mode' = 'earliest-offset',     --'properties.bootstrap.servers' = '82.157.27.147:9092',     'properties.bootstrap.servers' = '10.1.0.10:9092',      'properties.group.id' = 'WebRecordGroup',  -- 必选参数, 一定要指定 Group ID    'format' = 'json',    'json.ignore-parse-errors' = 'true',     -- 忽略 JSON 结构解析异常    'json.fail-on-missing-field' = 'false'   -- 如果设置为 true, 则遇到缺失字段会报错 设置为 false 则缺失字段设置为 null);

-- UV sinkCREATE TABLE `output_uv` (  `userids`   STRING,`user_id` STRING) WITH ( 'connector' = 'redis',           'command' = 'sadd',              -- 使用集合保存uv（支持命令：set、lpush、sadd、hset、zadd） 'nodes' = '192.28.28.217:6379',  -- redis连接地址,集群模式多个节点使用'',''分隔。 -- 'additional-key' = '<key>',   -- 用于指定hset和zadd的key。hset、zadd必须设置。 'password' = 'yourpassword'   );
-- PV sinkCREATE TABLE `output_pv` (  `pagevisits`   STRING,`product_id` STRING,`hour_count` BIGINT) WITH ( 'connector' = 'redis',           'command' = 'lpush',              -- 使用列表保存pv（支持命令：set、lpush、sadd、hset、zadd） 'nodes' = '192.28.28.217:6379',   -- redis连接地址,集群模式多个节点使用'',''分隔。 -- 'additional-key' = '<key>',   -- 用于指定hset和zadd的key。hset、zadd必须设置。 'password' = 'yourpassword'   );
-- 转化率 sinkCREATE TABLE `output_conversion_rate` (  `conversion_rate`   STRING,`rate` STRING) WITH ( 'connector' = 'redis',         'command' = 'set',              -- 使用列表保存pv（支持命令：set、lpush、sadd、hset、zadd） 'nodes' = '192.28.28.217:6379', -- redis连接地址,集群模式多个节点使用'',''分隔。 -- 'additional-key' = '<key>', -- 用于指定hset和zadd的key。hset、zadd必须设置。 'password' = 'yourpassword'   );

-- 加工得到 UV 指标，统计所有时间内的 UVINSERT INTO output_uv SELECT  'userids' AS `userids`,CAST(user_id AS string) AS user_id FROM input_web_record ;
-- 加工并得到 PV 指标，统计每 10 分钟内的 PVINSERT INTO output_pv SELECT  'pagevisits' AS pagevisits, CAST(product_id AS string) AS product_id, SUM(product_id) AS hour_countFROM input_web_record WHERE record_type = 0 GROUP BY HOP(times, INTERVAL '5' MINUTE, INTERVAL '10' MINUTE), product_id, user_id;
-- 加工并得到转化率指标，统计每 10 分钟内的转化率INSERT INTO output_conversion_rate SELECT  'conversion_rate' AS conversion_rate, CAST( (((SELECT COUNT(1) FROM input_web_record WHERE record_type=0)*1.0)/SUM(a.product_id)) as string) FROM (SELECT * FROM input_web_record where record_type = 1) AS aGROUP BY  HOP(times, INTERVAL '5' MINUTE, INTERVAL '10' MINUTE), product_id;