視頻直播：實時數(shù)據(jù)可視化分析

#!/usr/bin/python3# 首次使用該腳本，需 "pip3 install kafka" 安裝kafka模塊import jsonimport randomimport timefrom kafka import KafkaProducer
TIME_FORMAT = "%Y-%m-%d %H:%M:%S"PROVINCES = ["北京", "廣東", "山東", "江蘇", "河南", "上海", "河北", "浙江", "香港",             "陜西", "湖南", "重慶", "福建", "天津", "云南", "四川", "廣西", "安徽",             "海南", "江西", "湖北", "山西", "遼寧", "臺灣", "黑龍江", "內(nèi)蒙古",             "澳門", "貴州", "甘肅", "青海", "新疆", "西藏", "吉林", "寧夏"]
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'))
# 模擬幾天前，幾小時前的數(shù)據(jù)pre_day_count = 0pre_hour_count = 0hour_unit = 3600day_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)

# 進(jìn)入HBase命令[root@172~]# hbase shell# 建表語句create ‘dim_hbase’, ‘cf’

CREATE DATABASE livedb;     --創(chuàng)建數(shù)據(jù)庫列表

{'user_id': 3165, 'ip': '123.0.0.105', 'room_id': 20210813, 'arrive_time': '2021-08-16 09:48:01', 'create_time': '2021-08-16 09:48:01', 'leave_time': '2021-08-16 09:48:01', 'region': 1122, 'grade': 1, 'province': '浙江'}

{     'user_id': 3994     , 'gift_type': 3     , 'gift_total_amount': 28     , 'room_id': 20210813     , 'ip': '123.0.0.105'     , 'create_time': '2021-08-16 09:46:51'     , 'update_time': '2021-08-16 09:46:51'}

CREATE TABLE `live_streaming_log_source ` (     `user_id`       BIGINT,     `ip`            VARCHAR,     `room_id`       BIGINT,     `arrive_time`   TIMESTAMP,     `leave_time`    TIMESTAMP,     `create_time`   TIMESTAMP,     `region_code`   INT,     `grade`         INT,     `province`      VARCHAR ) WITH (     'connector' = 'kafka',     'topic' = 'live_streaming_log',     'scan.startup.mode' = 'earliest-offset',     'properties.bootstrap.servers' = '172.28.28.13:9092',     '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`        BIGINT,     `arrive_time`      TIMESTAMP,     `leave_time`      TIMESTAMP,     `create_time`     TIMESTAMP,     `region_code`     INT,     `grade`           INT,     `province`        VARCHAR,     primary key(`user_id`, `ip`,`room_id`,`arrive_time`) not enforced) WITH (    'connector' = 'jdbc',         'url' ='jdbc:mysql://172.28.28.227:3306/livedb?rewriteBatchedStatements=true&serverTimezon=Asia/Shanghai',    'table-name' = 'live_streaming_log',    'username' = 'root',    'password' = 'xxxxx',    'sink.buffer-flush.max-rows' = '5000',    'sink.buffer-flush.interval' = '2s',    'sink.max-retries' = '3');

INSERT INTO `live_streaming_log_sink`SELECT `*` FROM `live_streaming_log_source`;

CREATE TABLE ` live_gift_total_source` (    `user_id`           VARCHAR,    `gift_type`          VARCHAR,    `gift_total_amount`   BIGINT,    `ip`                VARCHAR,    `create_time`        VARCHAR) WITH (     'connector' = 'kafka',     'topic' = 'live_gift_total',     'scan.startup.mode' = 'earliest-offset',     'properties.bootstrap.servers' = '172.28.28.13:9092',     'properties.group.id' = 'joylyu-consumer-1',     'format' = 'json',     'json.ignore-parse-errors' = 'false',     'json.fail-on-missing-field' = 'false' );

CREATE TABLE `live_gift_total_sink` (`gift_type` VARCHAR,`gift_total_amount` BIGINT,primary key(`user_id`, `gift_type`) not enforced) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://172.28.28.227:3306/livedb?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai','table-name' = 'live_gift_total','username' = 'root','password' = 'xxxxx','sink.buffer-flush.max-rows' = '5000','sink.buffer-flush.interval' = '2s','sink.max-retries' = '3');

INSERT INTO `live_gift_total_sink`SELECT `gift_type`,   SUM(`gift_total_amount`) as `gift_total_amount_all`FROM `live_gift_total_source`GROUP BY `gift_type`;

CREATE TABLE `live_gift_total_source` (`user_id`            VARCHAR,`gift_type`          VARCHAR,`gift_total_amount`   BIGINT,`ip`                   VARCHAR,`create_time`     VARCHAR,proc_time AS PROCTIME()) WITH ( 'connector' = 'kafka', 'topic' = 'live_gift_total', 'scan.startup.mode' = 'earliest-offset', 'properties.bootstrap.servers' = '172.28.28.13:9092', 'properties.group.id' = 'joylyu-consumer-1', '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' = '用戶自己的hbase服務(wù)器zookeeper地址');

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

INSERT INTO `module_gift_total_sink`SELECT`b`.`cf`.`module_id`,SUM(`a`.`gift_total_amount`) AS `module_gift_total_amount`FROM `live_gift_total_source` AS `a`LEFT JOIN `dim_hbase` AS `b` for SYSTEM_TIME as of `a`.`proc_time` ON `a`.`room_id` = `b`.`rowkey`GROUP BY `b`.`cf`.`module_id`;