• 本地数据中心（IDC）的自建 Kafka 集群

• 私有网络 VPC

• 专线接入/云联网/VPN连接/对等连接

• 流计算 Oceanus (Flink)

• 云数据库 Redis

流计算 Oceanus 是大数据产品生态体系的实时化分析利器，是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点的企业级实时大数据分析平台。流计算 Oceanus 以实现企业数据价值最大化为目标，加速企业实时化数字化的建设进程。

2.4 配置自建 Kafka 集群

2.4.1 修改自建 Kafka 集群配置

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

! 若在云上使用到自建的zookeeper地址，也需要将zk配置中的hostname修改IP地址形式。

2.4.2 模拟发送数据到topic

./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"}

2.5 打通自建 IDC 集群到腾讯云网络通信

• 专线接入

  https://cloud.tencent.com/document/product/216

  适用于本地数据中心 IDC 与腾讯云网络打通。

• 云联网

  https://cloud.tencent.com/document/product/877

  适用于本地数据中心 IDC 与腾讯云网络打通，也可用于云上不同地域间私有网络 VPC 打通。

• VPN连接

  https://cloud.tencent.com/document/product/554

  适用于本地数据中心 IDC 与腾讯云网络打通。

• 对等连接+ NAT网关

  对等连接：https://cloud.tencent.com/document/product/553

  NAT网关：https://cloud.tencent.com/document/product/552

  适合云上不同地域间私有网络 VPC 打通，不适合本地 IDC 到腾讯云网络。

3.1 业务目标

• 网站的独立访客数量 UV。Oceanus 处理后在 Redis 中通过 set 类型存储独立访客数量，同时也达到了对同一访客的数据去重的目的。

• 网站商品页面的点击量 PV。Oceanus 处理后在 Redis 中使用 list 类型存储页面点击量。

• 转化率（转化率 = 成交次数 / 点击量）。Oceanus 处理后在 Redis 中用 String 存储即可。

3.2 源数据格式

# 浏览记录
{
 "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"
}

3.3 编写 Flink SQL 作业

1、定义 Source

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
);

2、定义 Sink

-- UV sink
CREATE 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 sink
CREATE 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'  
);

-- 转化率 sink
CREATE 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'  
);

3、业务逻辑

-- 加工得到 UV 指标，统计所有时间内的 UV
INSERT INTO output_uv
SELECT
 'userids' AS `userids`,
CAST(user_id AS string) AS user_id
FROM input_web_record ;

-- 加工并得到 PV 指标，统计每 10 分钟内的 PV
INSERT INTO output_pv
SELECT
 'pagevisits' AS pagevisits,
CAST(product_id AS string) AS product_id,
SUM(product_id) AS hour_count
FROM 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 a
GROUP BY  
HOP(times, INTERVAL '5' MINUTE, INTERVAL '10' MINUTE),
product_id;

3.4 结果验证

流计算 Oceanus 限量秒杀专享活动火爆进行中↓↓

腾讯云大数据

长按二维码
关注我们