万字长文-基于KaiwuDB在游乐场“刷卡+投币”双模消费系统中的落地实践

CREATE TABLE members (
    card_id VARCHAR(32) PRIMARY KEY,          -- 卡号/手环ID
    member_name VARCHAR(50),                    -- 会员姓名
    phone VARCHAR(20),                           -- 手机号
    card_type VARCHAR(20),                       -- 卡类型（普通卡/会员卡/月卡/年卡等）
    balance DECIMAL(10,2),                       -- 当前余额
    total_recharge DECIMAL(10,2),                 -- 累计充值金额
    total_bonus DECIMAL(10,2),                    -- 累计获赠金额
    create_date DATE,                             -- 开卡日期
    expire_date DATE,                             -- 有效期（针对月卡/年卡）
    statusVARCHAR(10),                           -- 状态（正常/挂失/冻结）
    INDEX idx_phone (phone),
    INDEX idx_card_type (card_type)
);

CREATE TABLE card_types (
    card_type_id INT PRIMARY KEY,
    type_name VARCHAR(50),                       -- 类型名称（如"会员卡"）
    discount_rate DECIMAL(3,2),                   -- 折扣率（0.8表示8折）
    need_password BOOLEAN,                        -- 消费是否需要密码
    daily_limit DECIMAL(10,2),                     -- 日消费限额
    per_trans_limit DECIMAL(10,2),                 -- 单笔消费限额
    bonus_rate DECIMAL(5,2),                       -- 充值赠送比例（如充100送20）
    description VARCHAR(200)
);

CREATE TABLE devices (
    device_id VARCHAR(32) PRIMARY KEY,           -- 设备ID
    device_name VARCHAR(100),                      -- 设备名称（如"旋转木马"）
    device_type VARCHAR(30),                       -- 设备类型（投币式/刷卡式/混合式）
    booth_id VARCHAR(20),                          -- 所属档口ID
    owner_type VARCHAR(10),                        -- 归属类型（自营/联营）
    split_ratio DECIMAL(4,2),                       -- 分账比例（如0.7表示场地方得70%）
    location VARCHAR(100),                          -- 位置描述
    price_per_play DECIMAL(6,2),                    -- 单次游玩价格
    coins_per_play INT,                             -- 所需投币数（投币机用）
    install_date DATE,                              -- 安装日期
    statusVARCHAR(10),                             -- 状态（正常/维护/停用）
    INDEX idx_booth (booth_id),
    INDEX idx_type (device_type)
);

use trace;

createtable transactions ( 
 transaction_time timestamptz(3) notnulldefault'now()'comment'交易发生时间（精度：秒）',
 amount intcomment'交易金额（单位：元），刷卡模式有效，投币模式为0',
 coins_used intcomment'投币数量（单位：个），投币模式有效，刷卡模式为0',
 balance_before intcomment'交易前会员卡余额（单位：元），刷卡模式有效',
 balance_after intcomment'交易后会员卡余额（单位：元），刷卡模式有效',
 game_duration intcomment'游戏时长（单位：秒），部分设备支持'
) 
tags(
 transaction_id varchar(64) notnullcomment'全局唯一交易ID，格式：YYYYMMDD+设备ID+序列号',
 device_id varchar(32) notnullcomment'设备ID，关联device_info表',
 card_id varchar(32) notnullcomment'会员卡号/手环ID，投币模式为NULL',
 transaction_type varchar(10) notnullcomment'交易类型：CARD-刷卡，COIN-投币，RECHARGE-充值，REFUND-退款',
 result_code intnotnullcomment'交易结果码：0-成功，1001-余额不足，1002-设备故障，1003-卡已挂失，1004-超时',
 sync_status varchar(10) notnullcomment'同步状态：online-实时上传，offline-脱机后补传'
) primary tags(transaction_id);

CREATE TABLE device_heartbeats (
    -- 时间列
    report_time TIMESTAMPCOMMENT'设备上报时间（精度：秒），通常每30秒-5分钟上报一次',
    
    -- 指标列（Fields）- 增量/瞬时值
    coin_box_level INTCOMMENT'投币箱满度（0-100%），80%以上建议清箱',
    temperature FLOATCOMMENT'设备温度（单位：摄氏度），超过50度报警',
    coins_increment INTCOMMENT'本周期内投币增量（单位：个），用于统计和计费',
    transactions_increment INTCOMMENT'本周期内交易次数增量',
    voltage FLOATCOMMENT'设备电压（单位：V），监控供电状态'
  )
    
    -- 标签列（Tags）- 用于过滤和分组
    TAGS(
        heartbeats_id VARCHAR(32) COMMENT'心跳时序表id',
        device_id VARCHAR(32) COMMENT'设备ID，关联device_info表',
        online_status BOOLEANCOMMENT'在线状态：true-在线，false-离线（超过3个周期未上报自动置为离线）',
        card_reader_status VARCHAR(20) COMMENT'读卡器状态：normal-正常，fault-故障，disconnected-断开',
        printer_paper BOOLEANCOMMENT'票据纸状态：true-有纸，false-缺纸',
        error_code INTCOMMENT'故障码：0-无故障，1-读卡器故障，2-投币器卡币，3-网络故障，4-温度过高',
        firmware_version VARCHAR(20) COMMENT'设备固件版本号，用于远程升级管理'
    )
primary tags(transaction_id);

use trace;

CREATETABLE offline_transactions (
    -- 时间列
    capture_time timestamptz(3) notnulldefault'now()'COMMENT'脱机交易发生时间（设备本地时间）',
    
    -- 指标列（Fields）
    transaction_count INTDEFAULT1COMMENT'本次记录包含的交易数量（通常是1，批量上传时可>1）',
    total_amount INTCOMMENT'交易总金额（单位：元）',
    data_size INTCOMMENT'交易数据大小（单位：KB）',
    upload_time TIMESTAMPCOMMENT'实际上传时间（网络恢复后）'
) 
    
    -- 标签列（Tags）- 用于过滤和分组
    TAGS(
     transaction_id varchar(64) notnullcomment'全局唯一交易ID，格式：YYYYMMDD+设备ID+序列号',
        batch_id VARCHAR(64) COMMENT'上传批次ID，格式：YYYYMMDD+设备ID+序列号',
        device_id VARCHAR(32) COMMENT'设备ID，关联device_info表',
        upload_status VARCHAR(10)  COMMENT'上传状态：pending-待上传，uploaded-已上传，failed-上传失败',
        
        sync_version INTCOMMENT'同步协议版本'
    )

 primary tags(transaction_id);

CREATE MATERIALIZEDVIEW booth_daily_revenue AS
SELECT
    d.booth_id,
    DATE(t.transaction_time) asdate,
    COUNT(*) as transaction_count,
    SUM(t.amount) as total_revenue,
    SUM(t.amount * d.split_ratio) as park_share,      -- 场地方分成
    SUM(t.amount * (1 - d.split_ratio)) as booth_share, -- 档口分成
    COUNT(DISTINCT t.card_id) as unique_users
FROM transactions t
JOIN devices d ON t.device_id = d.device_id
WHERE t.transaction_type = '刷卡'
GROUPBY d.booth_id, DATE(t.transaction_time);

# 伪代码：传统架构的复杂处理
def card_payment(card_id, device_id):
    # 1. 先从Redis查缓存余额（快）
    balance = redis.get(f"balance:{card_id}")
    
    # 2. 再到MySQL确认并扣款（事务）
    conn.begin()
    try:
        # 查询真实余额
        real_balance = conn.execute("SELECT balance FROM members WHERE card_id = ? FOR UPDATE", card_id)
        
        # 检查余额
        price = get_device_price(device_id)
        if real_balance < price:
            return"余额不足"
        
        # 扣款
        conn.execute("UPDATE members SET balance = balance - ? WHERE card_id = ?", price, card_id)
        
        # 记录交易
        conn.execute("INSERT INTO transactions ...")
        
        conn.commit()
        
        # 3. 更新Redis缓存
        redis.set(f"balance:{card_id}", real_balance - price)
        
        return"成功"
    except:
        conn.rollback()
        return"失败"

-- 1. 查询会员余额（带行锁，防止并发扣款）
SELECT balance 
FROM main.members 
WHERE card_id = 'CARD-628001'
FORUPDATE;
-- 假设结果：balance = 100

-- 2. 查询设备价格
SELECT price_per_play 
FROM main.devices 
WHERE device_id = 'device-001'ANDstatus = '正常';
-- 假设结果：price_per_play = 30

-- 3. 业务校验：余额 >= 价格（100 >= 30，通过）

-- 4. 插入交易记录（时序表）- 只记录交易
INSERTINTO"trace".transactions
(transaction_id, device_id, card_id, transaction_type, transaction_time,result_code, sync_status, amount, coins_used, balance_before, balance_after, game_duration) 
VALUES('20250901100000D001P00000001', 'device-001', 'CARD-628001', 'CARD', 'now()', 0, 'online', 20, null, 100, 80, 5);

-- 5. 更新会员余额（关系表）
UPDATE main.members 
SET balance = 80
WHERE card_id = 'CARD-628001';

-- 投币模式：只需记录投币事件，不涉及余额

-- 1. 查询设备所需投币数（可选，用于校验）
SELECT coins_per_play, device_name
FROM devices 
WHERE device_id = 'device-003';
-- 假设设备需要2个币玩一次

-- 2. 插入投币交易记录（时序表）
INSERTINTO"trace".transactions
(transaction_id, device_id, card_id, transaction_type, transaction_time,result_code, sync_status, amount, coins_used, balance_before, balance_after, game_duration) 
VALUES('20250901100000D001P00000201', 'device-002', null, 'COIN', 'now()', 0, 'online', null, 2, null, null, 10);

-- 投币模式不涉及会员余额，无需复杂事务

-- 分析刷卡用户和投币用户的游玩偏好，同时关联设备状态
SELECT
    MAX(d.device_name),
    d.device_type,
    t.transaction_type,
    COUNT(*) as play_count,
    AVG(t.game_duration) as avg_duration,
    AVG(h.temperature) as avg_device_temp,
    SUM(CASEWHEN h.error_code > 0THEN1ELSE0END) as error_count
FROM trace.transactions t
JOIN main.devices d ON t.device_id = d.device_id
LEFTJOIN trace.device_heartbeats h 
    ON t.device_id = h.device_id 
    AND h.report_time BETWEEN t.transaction_time - INTERVAL'1 MINUTE'
                          AND t.transaction_time + INTERVAL'1 MINUTE'
WHERE t.transaction_time > NOW() - INTERVAL'365 DAY'
GROUPBY d.device_type, t.transaction_type
ORDERBY d.device_type, t.transaction_type;

INSERT INTO device_heartbeats (report_time,coin_box_level,temperature,coins_increment,transactions_increment,voltage,heartbeat_id,device_id,online_status,card_reader_status,printer_paper,error_code,firmware_version) VALUES 
('2026-03-04 10:03:20.000',5,30,800,400,220,'ad12fh23940xeha21001','device-002',true,'normal',true,0,'V1.0.1')
;

-- 查询最近1小时有交易但超过10分钟没心跳的设备(跨模查询)
SELECTDISTINCT t.device_id, max(d.device_name), COUNT(*) as transaction_count
FROM trace.transactions t
LEFTJOIN trace.device_heartbeats h 
    ON t.device_id = h.device_id 
    AND h.report_time > NOW() - INTERVAL'10 MINUTE'
INNERJOIN main.devices d 
    ON d.device_id = t.device_id 
WHERE t.transaction_time > NOW() - INTERVAL'1 HOUR'
AND h.device_id ISNULL
GROUPBY t.device_id;

-- 查询在线但最近30分钟无交易的设备
SELECT
    h.device_id,
    h.report_time,
    h.temperature,
    h.coin_box_level,
    d.device_name
FROM device_heartbeats h
INNERJOIN main.devices d 
    ON d.device_id = h.device_id 
WHERE h.report_time > NOW() - INTERVAL'5 MINUTE'-- 5分钟内还在上报
AND h.online_status = true
ANDNOTEXISTS (
      SELECT1FROM transactions t
      WHERE t.device_id = h.device_id
        AND t.transaction_time > NOW() - INTERVAL'30 MINUTE'
  );

-- 查询投币箱满度>80%的设备，并查看最近1小时投币量
SELECT
    h.device_id,
    d.device_name,
    d.location,
    h.coin_box_level as 投币箱满度,
    h.report_time as last_heartbeat,
    COALESCE(SUM(t.coins_used), 0) as coins_last_hour,
    COALESCE(COUNT(t.transaction_id), 0) as transactions_last_hour
FROM device_heartbeats h
JOIN devices d ON h.device_id = d.device_id
LEFTJOIN transactions t 
    ON h.device_id = t.device_id 
    AND t.transaction_time > NOW() - INTERVAL'1 HOUR'
WHERE h.coin_box_level > 80
AND (h.device_id, h.report_time) IN (
      SELECT device_id, MAX(report_time)
      FROM device_heartbeats
      GROUPBY device_id
  )
GROUPBY h.device_id, d.device_name, d.location, h.coin_box_level, h.report_time
ORDERBY h.coin_box_level DESC;

SELECT 
    d.device_id,
    d.device_name,
    d.location,
    d.booth_id,
    -- 设备价值（基于历史收益）
    COALESCE(rev.yesterday_revenue, 0) as yesterday_revenue,
    -- 故障情况
    f.failure_count_7d,
    -- 当前设备状态
    h.coin_box_level,
    h.online_status,
    h.temperature,
    h.last_report_time,
    -- 计算综合优先级得分
    (COALESCE(rev.yesterday_revenue, 0) / 100) * 0.4 +
    (f.failure_count_7d) * 10 * 0.3 +
    (CASE
        WHEN h.coin_box_level > 80THEN50   -- 投币箱快满
        WHEN h.coin_box_level > 60THEN30
        ELSE0
    END) * 0.3as priority_score
FROM main.devices d
-- 关联最新设备状态（时序表）
LEFTJOIN (
    SELECT device_id, coin_box_level, online_status, temperature, 
           error_code, report_time as last_report_time
    FROM trace.device_heartbeats
    WHERE (device_id, report_time) IN (
        SELECT device_id, MAX(report_time)
        FROM trace.device_heartbeats
        GROUPBY device_id
    )
) h ON d.device_id = h.device_id
-- 关联昨日收益（时序聚合）
LEFTJOIN (
    SELECT device_id, SUM(amount) as yesterday_revenue
    FROM trace.transactions
    WHERE transaction_time = CURRENT_DATE - INTERVAL'1 DAY'
      AND transaction_type IN ('刷卡', '投币')
    GROUPBY device_id
) rev ON d.device_id = rev.device_id
-- 关联近期故障统计（时序聚合）
LEFTJOIN (
    SELECT device_id, 
           COUNT(*) as failure_count_7d
          
    FROM trace.device_heartbeats
    WHERE report_time >= NOW() - INTERVAL'7 DAY'
      AND error_code > 0
    GROUPBY device_id
) f ON d.device_id = f.device_id
WHERE d.status = '正常'
AND (
      h.coin_box_level > 60OR
      f.failure_count_7d > 0OR
      h.online_status = false
  )
ORDERBY priority_score DESC
LIMIT20;

-- 按小时统计各项目游玩人次
SELECT
    d.device_name,
    d.device_type,
    EXTRACT(HOURFROM t.transaction_time) as hour_of_day,
    COUNT(*) as play_count,
    SUM(t.amount) as revenue,
    -- 同比上周同一时段
    (COUNT(*) - AVG(COUNT(*)) OVER (
        PARTITIONBY d.device_id, EXTRACT(DOW FROM t.transaction_time), EXTRACT(HOURFROM t.transaction_time)
        ORDERBY t.transaction_time
        ROWSBETWEEN1PRECEDINGAND1PRECEDING
    )) as vs_last_week
FROM transactions t
JOIN devices d ON t.device_id = d.device_id
WHERE t.transaction_time >= NOW() - INTERVAL'7 DAY'
GROUPBY d.device_id, d.device_name, d.device_type, EXTRACT(HOURFROM t.transaction_time)
ORDERBY play_count DESC;

-- 项目热度排名（用于大屏展示）
SELECT 
    d.device_name,
    COUNT(*) as today_plays
FROM trace.transactions t
JOIN main.devices d ON t.device_id = d.device_id
GROUP BY d.device_name
order BY COUNT(*) DESC

<dir_path>, <level>

/data/kwdb/hot,0
/data/kwdb/warm,1
/data/kwdb/cold,2

SET CLUSTER SETTING ts.tier.duration = "<period>,<period>";

-- 实时总营收（今日）
SELECTSUM(amount) as today_revenue 
FROM trace.transactions 
WHERE transaction_time = CURRENT_DATE
AND result_code = 0;

-- 实时游客数（今日去重）
SELECTCOUNT(DISTINCT card_id) as today_visitors 
FROM trace.transactions 
WHERE transaction_time = CURRENT_DATE;

-- 设备在线率
SELECT
    ROUND(100.0 * SUM(CASEWHEN online_status = trueTHEN1ELSE0END) / COUNT(*), 2) as online_rate
FROM (
    SELECT device_id, online_status
    FROM trace.device_heartbeats
    WHERE (device_id, report_time) IN (
        SELECT device_id, MAX(report_time)
        FROM trace.device_heartbeats
        GROUPBY device_id
    )
) t;

-- 当前故障设备
SELECT
    d.device_name,
    d.location,
    h.error_code,
    h.report_time as last_heartbeat
FROM device_heartbeats h
JOIN main.devices d ON h.device_id = d.device_id
WHERE h.error_code > 0
AND (h.device_id, h.report_time) IN (
      SELECT device_id, MAX(report_time)
      FROM trace.device_heartbeats
      GROUPBY device_id
  )
ORDERBY h.report_time DESC;

-- 投币箱即将满的设备（>80%）
SELECT
    d.device_name,
    d.location,
    h.coin_box_level
FROM device_heartbeats h
JOIN main.devices d ON h.device_id = d.device_id
WHERE h.coin_box_level > 80
AND (h.device_id, h.report_time) IN (
      SELECT device_id, MAX(report_time)
      FROM trace.device_heartbeats
      GROUPBY device_id
  );

-- 今日热门项目TOP10
SELECT
    d.device_name,
    COUNT(*) as play_count,
    SUM(t.amount) as revenue
FROM trace.transactions t
JOIN main.devices d ON t.device_id = d.device_id
WHERE t.transaction_time = CURRENT_DATE
GROUPBY d.device_name
ORDERBY play_count DESC
LIMIT10;

-- 时段热度分析（过去7天）
SELECT
    EXTRACT(HOURFROM t.transaction_time) ashour,
    COUNT(*) as avg_plays
FROM trace.transactions t
WHERE t.transaction_time > NOW() - INTERVAL'7 DAY'
GROUPBYEXTRACT(HOURFROM t.transaction_time)
ORDERBYhour;

-- 各档口今日营收及分成
SELECT 
    d.booth_id,
    SUM(t.amount) as revenue,
    SUM(t.amount * d.split_ratio) as park_share,
    SUM(t.amount * (1 - d.split_ratio)) as booth_share
FROM trace.transactions t
JOIN main.devices d ON t.device_id = d.device_id
WHERE t.transaction_time = CURRENT_DATE
GROUP BY d.booth_id
ORDER BY revenue DESC;

-- 各区域实时热度
SELECT 
    d.location,
    COUNT(*) as recent_plays
FROM trace.transactions t
JOIN main.devices d ON t.device_id = d.device_id
WHERE t.transaction_time > NOW() - INTERVAL '30 MINUTE'
GROUP BY d.location;

服务名称: kwdb-prometheus
镜像: prom/prometheus:v2.53.0
端口: 9090

# 数据采集周期：10秒
# 数据保留时间：15天
# 支持热加载配置：--web.enable-lifecycle
# 提供管理API：--web.enable-admin-api

服务名称: kwdb-alertmanager
镜像: prom/alertmanager:v0.26.0
端口: 9093
路径前缀: /alertmanager/

# 告警分组策略
group_by: ['alertname', 'cluster']  # 按告警名和集群分组
group_wait: 10s      # 组内告警等待时间
group_interval: 10s  # 组内告警发送间隔
repeat_interval: 1h  # 重复告警间隔

服务名称: kwdb-grafana
镜像: grafana/grafana:10.2.2
端口: 3000
默认账号: admin/admin

# 自动添加Prometheus数据源
# 自动加载仪表盘配置
# 支持自定义仪表盘
# 实时数据展示

服务名称: kwdb-cadvisor
镜像: gcr.io/cadvisor/cadvisor:latest
端口: 8088
权限: privileged