视频直播解决方案之实时 BI 分析

#!/usr/bin/python3
# 首次使用该脚本，需 "pip3 install kafka" 安装kafka模块
import json
import random
import time
from kafka import KafkaProducer

TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
PROVINCES = ["北京", "广东", "山东", "江苏", "河南", "上海", "河北", "浙江", "香港",
          "陕西", "湖南", "重庆", "福建", "天津", "云南", "四川", "广西", "安徽",
          "海南", "江西", "湖北", "山西", "辽宁", "台湾", "黑龙江", "内蒙古",
          "澳门", "贵州", "甘肃", "青海", "新疆", "西藏", "吉林", "宁夏"]

broker_lists = ['172.28.28.13:9092']
topic_live_gift_total = 'live_gift_total'
topic_live_streaming_log = 'live_streaming_log'

producer = KafkaProducer(bootstrap_servers=broker_lists,
                      value_serializer=lambda m: json.dumps(m).encode('ascii'))

# 模拟几天前，几小时前的数据
pre_day_count = 0
pre_hour_count = 0
hour_unit = 3600
day_unit = 3600 * 24

def generate_data_live_gift_total():
 # construct time
 update_time = time.time() - day_unit * pre_day_count
 update_time_str = time.strftime(TIME_FORMAT, time.localtime(update_time))
 create_time = update_time - hour_unit * pre_hour_count
 create_time_str = time.strftime(TIME_FORMAT, time.localtime(create_time))
 results = []

 for _ in range(0, 10):
     user_id = random.randint(2000, 4000)
     random_gift_type = random.randint(1, 10)
     random_gift_total = random.randint(1, 100)
     msg_kv = {"user_id": user_id, "gift_type": random_gift_type,
               "gift_total_amount": random_gift_total,
               "create_time": create_time_str, "update_time": update_time_str}
     results.append(msg_kv)
 return results


def generate_live_streaming_log():
 # construct time
 update_time = time.time() - day_unit * pre_day_count
 leave_time_str = time.strftime(TIME_FORMAT, time.localtime(update_time))
 create_time = update_time - hour_unit * pre_hour_count
 create_time_str = time.strftime(TIME_FORMAT, time.localtime(create_time))
 results = []

 for _ in range(0, 10):
     user_id = random.randint(2000, 4000)
     random_province = random.randint(0, len(PROVINCES) - 1)
     province_name = PROVINCES[random_province]
     grade = random.randint(1, 5)
     msg_kv = {"user_id": user_id, "ip": "123.0.0." + str(user_id % 255),
               "room_id": 20210813, "arrive_time": create_time_str,
               "create_time": create_time_str, "leave_time": leave_time_str,
               "region": 1122, "grade": (user_id % 5 + 1), "province": province_name}
     results.append(msg_kv)
 return results


def send_data(topic, msgs):
 count = 0

 # produce asynchronously
 for msg in msgs:
     import time
     time.sleep(1)
     count += 1
     producer.send(topic, msg)
     print(" send %d data...\\n %s" % (count, msg))

 producer.flush()


if __name__ == '__main__':
 count = 1
 while True:
     time.sleep(60)
 #for _ in range(count):
     msg_live_stream_logs = generate_live_streaming_log()
     send_data(topic_live_streaming_log, msg_live_stream_logs)

     msg_topic_live_gift_totals = generate_data_live_gift_total()
     send_data(topic_live_gift_total, msg_topic_live_gift_totals)

# 进入HBase命令
[root@172~]# hbase shell
# 建表语句
create 'dim_hbase', 'cf'

CREATE DATABASE livedb;     --创建数据库列表

CREATE TABLE `live_streaming_log_source` (
  `user_id`       BIGINT,
  `ip`            VARCHAR,
  `room_id`       BIGINT, 
  `arrive_time`   VARCHAR,
  `leave_time`    VARCHAR,
  `create_time`   VARCHAR,
  `region_code`   INT,
  `grade`         INT,
  `province`      VARCHAR
 ) WITH (
  'connector' = 'kafka', 
  'topic' = 'live_streaming_log', 
  'scan.startup.mode' = 'earliest-offset', 
  'properties.bootstrap.servers' = 'xx.xx.xx.xx:xxxx',  
  'properties.group.id' = 'joylyu-consumer-2', 
  'format' = 'json',
  'json.ignore-parse-errors' = 'false', 
  'json.fail-on-missing-field' = 'false'
 );

CREATE TABLE `live_streaming_log_sink` (
  `user_id`       BIGINT,
  `ip`            VARCHAR,
  `room_id`       VARCHAR, 
  `arrive_time`   TIMESTAMP,
  `leave_time`    TIMESTAMP,
  `create_time`   TIMESTAMP,
  `region_code`   VARCHAR,
  `grade`         INT,
  `province`      VARCHAR,
  primary key(`user_id`, `ip`,`room_id`,`arrive_time`) not enforced
) WITH (
 'connector' = 'jdbc',
 'url' = 'jdbc:mysql://xx.xx.xx.xx:xxxx/livedb?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',          -- 请替换为您的实际 MySQL 连接参数
 'table-name' = 'live_streaming_log',    -- 需要写入的数据表
 'username' = 'root',                    -- 数据库访问的用户名（需要提供 INSERT 权限）
 'password' = 'xxxxxxxxx',               -- 数据库访问的密码
 'sink.buffer-flush.max-rows' = '5000', 
 'sink.buffer-flush.interval' = '2s',
 'sink.max-retries' = '3'
);

INSERT INTO `live_streaming_log_sink`
SELECT 
CASE WHEN `user_id` IS NULL THEN 0000 ELSE `user_id` END AS `user_id`
, `ip`
, CAST(`room_id`     AS VARCHAR)   AS `room_id`
, CAST(`arrive_time` AS TIMESTAMP) AS `arrive_time`
, CAST(`leave_time`  AS TIMESTAMP) AS `leave_time`
, CAST(`create_time` AS TIMESTAMP) AS `create_time`
, CAST(`region_code` AS VARCHAR)   AS `region_code`
, `grade`
, `province`
FROM `live_streaming_log_source`;

 CREATE TABLE `live_gift_total_source` (
 `user_id` BIGINT,
 `gift_type` INT,
 `gift_total_amount` BIGINT,
 `create_time` VARCHAR,
 `update_time` VARCHAR
 ) WITH (
  'connector' = 'kafka',
  'topic' = 'live_gift_total',  -- 替换为您要消费的 Topic
  'scan.startup.mode' = 'earliest-offset',
  'properties.bootstrap.servers' = 'xx.xx.xx.xx:xxxx',  -- 替换为您的 Kafka 连接地址
  'properties.group.id' = 'demo3Group2',  -- 必选参数, 一定要指定 Group ID
  'format' = 'json',
  'json.ignore-parse-errors' = 'false',
  'json.fail-on-missing-field' = 'false'
 );

CREATE TABLE `live_gift_total_amount_sink ` (
 `gift_type`         VARCHAR,
 `gift_total_amount` BIGINT,
 primary key(`gift_type`) not enforced
) WITH (
 'connector' = 'jdbc',
 'url' = 'jdbc:mysql://xx.xx.xx.xx:xxxx/livedb?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',
 'table-name' = 'live_gift_total_amount',   -- 需要写入的数据表
 'username' = 'root',           -- 数据库访问的用户名（需要提供 INSERT 权限）
 'password' = 'xxxxxxxxxxxxx',  -- 数据库访问的密码
 'sink.buffer-flush.max-rows' = '5000',
 'sink.buffer-flush.interval' = '2s',
 'sink.max-retries' = '3'
);

INSERT INTO `live_gift_total_amount_sink`
SELECT 
CAST(`gift_type` AS VARCHAR) AS `gift_type`
, SUM(`gift_total_amount`) AS `gift_total_amount_all`
FROM `live_gift_total_source`
GROUP BY CAST(`gift_type` AS VARCHAR);

CREATE TABLE `live_streaming_log_source` (
  `user_id`       BIGINT,
  `ip`            VARCHAR,
  `room_id`       BIGINT, 
  `arrive_time`   VARCHAR,
  `leave_time`    VARCHAR,
  `create_time`   VARCHAR,
  `region_code`   INT,
  `grade`         INT,
  `province`      VARCHAR,
  `proc_time`  AS PROCTIME()
 ) WITH (
  'connector' = 'kafka',
  'topic' = 'live_streaming_log',                       -- 替换为您要消费的 Topic
  'scan.startup.mode' = 'earliest-offset',
  'properties.bootstrap.servers' = 'xx.xx.xx.xx:xxxx',  -- 替换为您的 Kafka 连接地址
  'properties.group.id' = 'demo3Group3',                -- 必选参数, 一定要指定 Group ID
  'format' = 'json',
  'json.ignore-parse-errors' = 'false',
  'json.fail-on-missing-field' = 'false'
 );

CREATE TABLE `dim_hbase` (
 `rowkey` STRING,
 `cf` ROW <`module_id` STRING>,
 PRIMARY KEY (`rowkey`) NOT ENFORCED
) WITH (
 'connector' = 'hbase-1.4',
 'table-name' = 'dim_hbase',
 'zookeeper.quorum' = 'xx.xx.xx.xx:8121,xx.xx.xx.xx:8121,xx.xx.xx.xx:8121'
);

CREATE TABLE `live_module_number_count_sink` (
 `module_id` BIGINT,
 `module_number_count` BIGINT,
 primary key(`module_id`) not enforced
) WITH (
 'connector' = 'jdbc',
 'url' = 'jdbc:mysql://xx.xx.xx.xx:xxxx/livedb?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai',
 'table-name' = 'live_module_number_count',
 'username' = 'root',
 'password' = 'xxxxxxxxxxx',
 'sink.buffer-flush.max-rows' = '5000',
 'sink.buffer-flush.interval' = '2s',
 'sink.max-retries' = '3'
);

INSERT INTO `live_module_number_count_sink`
SELECT
CAST(dim_hbase.cf.module_id AS BIGINT) AS module_id,
COUNT(live_streaming_log_source.`user_id`) AS module_number_count
FROM `live_streaming_log_source`
JOIN `dim_hbase` for SYSTEM_TIME as of live_streaming_log_source.proc_time
 ON CAST(live_streaming_log_source.room_id AS STRING) = dim_hbase.rowkey
GROUP BY CAST(dim_hbase.cf.module_id AS BIGINT);