MySQL玩转数据可视化之从SQL查询到动态图表的完整实战

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  业务mysql   │────▶│  etl工具    │────▶│  分析mysql   │────▶│   bi工具    │
│  (oltp)     │     │ (airflow/   │     │  (olap)     │     │ (metabase/  │
│             │     │  canal/     │     │             │     │  superset)  │
│  实时交易    │     │  datax)     │     │  预聚合数据  │     │             │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘

-- 事实表：订单事实
create table fact_orders (
    order_id bigint primary key,
    order_date_key int not null,      -- 外键关联日期维表
    customer_key int not null,        -- 外键关联客户维表
    product_key int not null,         -- 外键关联产品维表
    region_key int not null,          -- 外键关联地区维表
    
    -- 度量值（可加）
    quantity int not null,
    unit_price decimal(10,2) not null,
    discount_amount decimal(10,2) default 0,
    shipping_fee decimal(10,2) default 0,
    
    -- 派生度量
    gross_amount decimal(10,2) as (quantity * unit_price) stored,
    net_amount decimal(10,2) as (quantity * unit_price - discount_amount) stored,
    
    -- 元数据
    created_at timestamp default current_timestamp,
    
    index idx_date (order_date_key),
    index idx_customer (customer_key),
    index idx_product (product_key)
) engine=innodb;

-- 维度表：日期维度（预生成10年数据）
create table dim_date (
    date_key int primary key,         -- 格式：yyyymmdd
    full_date date not null,
    year smallint not null,
    quarter tinyint not null,
    month tinyint not null,
    day tinyint not null,
    week_of_year tinyint not null,
    day_of_week tinyint not null,     -- 1=周一, 7=周日
    is_weekend boolean generated always as (day_of_week in (6,7)) stored,
    is_holiday boolean default false, -- 需手动维护节假日
    
    index idx_year_month (year, month)
) engine=innodb;

-- 维度表：客户维度（scd type 2缓慢变化维）
create table dim_customer (
    customer_key int auto_increment primary key,
    customer_id varchar(50) not null, -- 业务主键
    customer_name varchar(100) not null,
    customer_level enum('普通','银卡','金卡','钻石') default '普通',
    region varchar(50),
    
    -- scd type 2 字段
    effective_date date not null,
    expiry_date date default '9999-12-31',
    is_current boolean default true,
    
    index idx_customer_id (customer_id, is_current)
) engine=innodb;

-- 存储过程：每日增量同步
delimiter //

create procedure sp_sync_orders(in sync_date date)
begin
    declare exit handler for sqlexception
    begin
        rollback;
        insert into etl_log (table_name, sync_date, status, error_msg) 
        values ('fact_orders', sync_date, 'failed', 'transaction rolled back');
    end;
    
    start transaction;
    
    -- 删除已存在的分区数据（支持重跑）
    delete from fact_orders 
    where order_date_key = date_format(sync_date, '%y%m%d');
    
    -- 增量插入
    insert into fact_orders (
        order_id, order_date_key, customer_key, product_key, region_key,
        quantity, unit_price, discount_amount, shipping_fee
    )
    select 
        o.order_id,
        date_format(o.created_at, '%y%m%d'),
        dc.customer_key,
        dp.product_key,
        dr.region_key,
        o.quantity,
        o.unit_price,
        o.discount_amount,
        o.shipping_fee
    from oltp.orders o
    join dim_customer dc on o.customer_id = dc.customer_id and dc.is_current = true
    join dim_product dp on o.product_id = dp.product_id and dp.is_current = true
    join dim_region dr on o.region_code = dr.region_code
    where date(o.created_at) = sync_date;
    
    commit;
    
    insert into etl_log (table_name, sync_date, status, rows_affected) 
    values ('fact_orders', sync_date, 'success', row_count());
end //

delimiter ;

-- 定时任务（event scheduler）
create event evt_daily_sync
on schedule every 1 day starts '2024-01-01 02:00:00'
do call sp_sync_orders(curdate() - interval 1 day);

-- 日销售趋势（支持同比环比）
select 
    d.full_date,
    sum(f.net_amount) as daily_sales,
    count(distinct f.order_id) as order_count,
    avg(f.net_amount) as avg_order_value,
    
    -- 同比（去年同期）
    lag(sum(f.net_amount), 365) over (order by d.full_date) as sales_yoy,
    
    -- 环比（上周同日）
    lag(sum(f.net_amount), 7) over (order by d.full_date) as sales_wow,
    
    -- 7日移动平均
    avg(sum(f.net_amount)) over (
        order by d.full_date 
        rows between 6 preceding and current row
    ) as ma7_sales
    
from fact_orders f
join dim_date d on f.order_date_key = d.date_key
where d.full_date between '2023-01-01' and '2024-01-01'
group by d.full_date, d.date_key
order by d.full_date;

-- 关键索引
create index idx_fact_orders_date_amount on fact_orders(order_date_key, net_amount);

-- 用户行为漏斗：访问->加购->下单->支付
with funnel_stages as (
    select 
        user_id,
        session_date,
        max(case when event_type = 'page_view' then 1 else 0 end) as has_view,
        max(case when event_type = 'add_cart' then 1 else 0 end) as has_cart,
        max(case when event_type = 'create_order' then 1 else 0 end) as has_order,
        max(case when event_type = 'pay_success' then 1 else 0 end) as has_pay
    from user_events
    where session_date between @start_date and @end_date
    group by user_id, session_date
)
select 
    '访问' as stage,
    count(*) as user_count,
    100.0 as conversion_rate
from funnel_stages where has_view = 1

union all

select 
    '加购',
    count(*),
    count(*) * 100.0 / (select count(*) from funnel_stages where has_view = 1)
from funnel_stages where has_cart = 1

union all

select 
    '下单',
    count(*),
    count(*) * 100.0 / (select count(*) from funnel_stages where has_view = 1)
from funnel_stages where has_order = 1

union all

select 
    '支付',
    count(*),
    count(*) * 100.0 / (select count(*) from funnel_stages where has_view = 1)
from funnel_stages where has_pay = 1;

-- 日粒度预聚合表
create table agg_daily_sales as
select 
    order_date_key,
    region_key,
    product_category_key,
    count(*) as order_count,
    sum(quantity) as total_quantity,
    sum(gross_amount) as gross_sales,
    sum(net_amount) as net_sales,
    sum(discount_amount) as total_discount,
    count(distinct customer_key) as unique_customers
from fact_orders
group by order_date_key, region_key, product_category_key;

-- 实时刷新（mysql 8.0.13+支持原子性ddl）
create or replace table agg_daily_sales as ...;

-- 或者使用触发器保持同步（适合近实时场景）
delimiter //
create trigger trg_orders_agg_insert 
after insert on fact_orders
for each row
begin
    insert into agg_daily_sales (
        order_date_key, region_key, product_category_key,
        order_count, total_quantity, gross_sales, net_sales, total_discount, unique_customers
    ) values (
        new.order_date_key, new.region_key, new.product_category_key,
        1, new.quantity, new.gross_amount, new.net_amount, new.discount_amount, 1
    )
    on duplicate key update
        order_count = order_count + 1,
        total_quantity = total_quantity + new.quantity,
        gross_sales = gross_sales + new.gross_amount,
        net_sales = net_sales + new.net_amount,
        total_discount = total_discount + new.discount_amount,
        unique_customers = unique_customers + if(
            (select count(*) from fact_orders 
             where order_date_key=new.order_date_key 
             and customer_key=new.customer_key) = 1, 1, 0
        );
end //
delimiter ;

# docker-compose.yml
version: '3'
services:
  metabase:
    image: metabase/metabase:latest
    ports:
      - "3000:3000"
    environment:
      - mb_db_type=mysql
      - mb_db_dbname=metabase
      - mb_db_port=3306
      - mb_db_user=metabase
      - mb_db_pass=secret
      - mb_db_host=mysql-analytics
    volumes:
      - metabase-data:/metabase-data
  mysql-analytics:
    image: mysql:8.0
    environment:
      - mysql_root_password=root
      - mysql_database=metabase
    volumes:
      - mysql-data:/var/lib/mysql
      - ./analytics_dump.sql:/docker-entrypoint-initdb.d/init.sql
volumes:
  metabase-data:
  mysql-data:

-- 保存为metabase question，选择"line"图表类型
-- 变量语法支持动态筛选
select 
    {{date_column}} as date,
    region,
    sum(net_amount) as sales
from fact_orders
where {{date_column}} between {{start_date}} and {{end_date}}
  [[and region = {{selected_region}}]]
group by 1, 2
order by 1;

// metabase嵌入式仪表盘配置示例
// 在前端应用中集成iframe
const metabaseurl = "http://localhost:3000";
const dashboardid = 123;
const token = generatesignedtoken({  // 使用metabase嵌入sdk生成
  resource: { dashboard: dashboardid },
  params: {
    "region": "华东",      // 预筛选参数
    "start_date": "2024-01-01"
  },
  exp: math.round(date.now() / 1000) + (10 * 60)  // 10分钟过期
});
const iframeurl = `${metabaseurl}/embed/dashboard/${token}#bordered=true&titled=true`;
// 嵌入到react/vue组件中
<iframe src={iframeurl} width="100%" height="800" frameborder="0"></iframe>

# superset_config.py
# 配置mysql作为元数据库和查询引擎
sqlalchemy_database_uri = 'mysql+mysqlconnector://superset:password@localhost/superset_metadata'
# 添加mysql数据源
from superset.connectors.sqla.models import sqlatable
from superset import db
# 通过api或ui添加数据库
database = database(
    database_name='analytics_warehouse',
    sqlalchemy_uri='mysql+mysqlconnector://readonly:password@analytics-host:3306/analytics_db',
    extra=json.dumps({
        "metadata_params": {},
        "engine_params": {
            "connect_args": {
                "ssl_disabled": true,
                "autocommit": true
            }
        },
        "metadata_cache_timeout": {
            "schema_cache_timeout": 300,
            "table_cache_timeout": 600
        }
    })
)
db.session.add(database)
db.session.commit()

// 开发superset自定义图表：桑基图（用户流转分析）
// plugin-chart-sankey/src/sankeychart.tsx
import react from 'react';
import { sankey } from '@ant-design/charts';
import { chartprops } from '@superset-ui/core';
interface sankeydata {
  source: string;
  target: string;
  value: number;
}
export default function sankeychart(props: chartprops<sankeydata[]>) {
  const { data, width, height } = props;
  const config = {
    data: data.map(d => ({ source: d.source, target: d.target, value: d.value })),
    sourcefield: 'source',
    targetfield: 'target',
    weightfield: 'value',
    nodewidthratio: 0.02,
    nodepaddingratio: 0.03,
    width,
    height,
    tooltip: {
      formatter: (datum: any) => ({
        name: `${datum.source} → ${datum.target}`,
        value: datum.value,
      }),
    },
  };
  return <sankey {...config} />;
}
// 对应sql查询
/*
select 
    '首页' as source,
    case 
        when page = 'product_list' then '商品列表'
        when page = 'search_result' then '搜索结果'
        else '其他'
    end as target,
    count(*) as value
from user_behavior
where event_date = '2024-01-01'
group by 1, 2;
*/

# grafana-datasources.yml
apiversion: 1
datasources:
  - name: mysql-analytics
    type: mysql
    url: analytics-db:3306
    database: analytics_db
    user: grafana_reader
    securejsondata:
      password: ${mysql_password}
    jsondata:
      maxopenconns: 100
      maxidleconns: 100
      connmaxlifetime: 14400

-- grafana query a：实时销售额（5分钟粒度）
select
    unix_timestamp(date_format(created_at, '%y-%m-%d %h:%i:00')) as time_sec,
    sum(net_amount) as value,
    '销售额' as metric
from fact_orders
where created_at >= date_sub(now(), interval 6 hour)
group by unix_timestamp(date_format(created_at, '%y-%m-%d %h:%i:00'))
order by time_sec;

-- grafana query b：实时订单量
select
    unix_timestamp(date_format(created_at, '%y-%m-%d %h:%i:00')) as time_sec,
    count(*) as value,
    '订单量' as metric
from fact_orders
where created_at >= date_sub(now(), interval 6 hour)
group by unix_timestamp(date_format(created_at, '%y-%m-%d %h:%i:00'))
order by time_sec;

# grafana-alerts.yml
apiversion: 1
groups:
  - orgid: 1
    name: sales_alerts
    folder: business metrics
    interval: 60s
    rules:
      - uid: sales_drop_alert
        title: 销售额骤降告警
        condition: c
        data:
          - refid: a
            relativetimerange:
              from: 300
              to: 0
            datasourceuid: mysql-analytics
            model:
              format: time_series
              rawsql: |
                select 
                    now() as time,
                    sum(net_amount) as current_sales
                from fact_orders
                where created_at >= date_sub(now(), interval 5 minute)
          - refid: b
            relativetimerange:
              from: 600
              to: 300
            datasourceuid: mysql-analytics
            model:
              format: time_series
              rawsql: |
                select 
                    now() as time,
                    sum(net_amount) as previous_sales
                from fact_orders
                where created_at between date_sub(now(), interval 10 minute) 
                                    and date_sub(now(), interval 5 minute)
          - refid: c
            datasourceuid: __expr__
            model:
              type: threshold
              expression: 'a / b < 0.7'  # 当前销售额低于上期70%触发
        nodatastate: nodata
        execerrstate: error
        for: 5m
        annotations:
          summary: "销售额较5分钟前下降超过30%"
          description: "当前销售额: {{ $values.a }}, 上期: {{ $values.b }}"
        labels:
          severity: critical

# python + flask-socketio + mysql实现实时数据推送
# 替代方案：使用apache kafka + flink + mysql cdc
from flask import flask
from flask_socketio import socketio, emit
import pymysql
import threading
import time
app = flask(__name__)
socketio = socketio(app, cors_allowed_origins="*")
class mysqlwatcher:
    def __init__(self):
        self.conn = pymysql.connect(
            host='localhost',
            user='realtime',
            password='secret',
            database='analytics_db',
            cursorclass=pymysql.cursors.dictcursor
        )
        self.last_id = 0
    def watch_orders(self):
        """监控新订单并推送"""
        while true:
            with self.conn.cursor() as cursor:
                sql = """
                    select order_id, net_amount, created_at 
                    from fact_orders 
                    where order_id > %s 
                    order by order_id 
                    limit 100
                """
                cursor.execute(sql, (self.last_id,))
                new_orders = cursor.fetchall()
                if new_orders:
                    self.last_id = new_orders[-1]['order_id']
                    # 计算实时指标
                    total_sales = sum(o['net_amount'] for o in new_orders)
                    # 推送到前端
                    socketio.emit('sales_update', {
                        'new_orders': len(new_orders),
                        'sales_amount': float(total_sales),
                        'timestamp': time.time()
                    }, broadcast=true)
            time.sleep(1)  # 每秒轮询（生产环境应使用binlog监听）
watcher = mysqlwatcher()
@socketio.on('connect')
def handle_connect():
    print('client connected')
    emit('init_data', {'status': 'connected'})
if __name__ == '__main__':
    # 启动监控线程
    t = threading.thread(target=watcher.watch_orders)
    t.daemon = true
    t.start()
    socketio.run(app, host='0.0.0.0', port=5000)

<!-- 实时销售监控页面 -->
<!doctype html>
<html>
<head>
    <title>实时销售监控</title>
    <script src="https://cdn.socket.io/4.5.4/socket.io.min.js"></script>
    <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script>
</head>
<body>
    <div id="main" style="width: 100%; height: 600px;"></div>
    <script>
        const socket = io('http://localhost:5000');
        const chart = echarts.init(document.getelementbyid('main'));
        // 初始化空数据
        const data = [];
        const now = new date();
        for (let i = 0; i < 60; i++) {
            data.push({
                name: new date(now - (60 - i) * 1000).tostring(),
                value: [new date(now - (60 - i) * 1000), 0]
            });
        }
        const option = {
            title: { text: '实时销售额（元/秒）' },
            tooltip: { trigger: 'axis' },
            xaxis: { type: 'time', splitline: { show: false } },
            yaxis: { type: 'value', splitline: { show: true } },
            series: [{
                name: '销售额',
                type: 'line',
                smooth: true,
                data: data,
                areastyle: {
                    color: new echarts.graphic.lineargradient(0, 0, 0, 1, [
                        { offset: 0, color: 'rgb(255, 158, 68)' },
                        { offset: 1, color: 'rgb(255, 70, 131)' }
                    ])
                }
            }]
        };
        chart.setoption(option);
        // 接收实时数据
        socket.on('sales_update', function(msg) {
            const now = new date();
            data.shift();
            data.push({
                name: now.tostring(),
                value: [now, msg.sales_amount]
            });
            chart.setoption({ series: [{ data: data }] });
        });
    </script>
</body>
</html>

-- 按时间范围分区（mysql 8.0）
create table fact_orders_partitioned (
    order_id bigint,
    order_date_key int,
    -- ... 其他字段
    primary key (order_id, order_date_key)  -- 分区键必须包含在主键中
) partition by range (order_date_key) (
    partition p202301 values less than (20230200),
    partition p202302 values less than (20230300),
    partition p202303 values less than (20230400),
    -- ...
    partition p_future values less than maxvalue
);

-- 查询优化器自动分区裁剪
explain partitions 
select * from fact_orders_partitioned 
where order_date_key between 20230101 and 20230131;
-- 结果：只扫描p202301分区

-- 对于纯分析场景，使用clickhouse作为mysql的从库
-- mysql主库 -> canal -> kafka -> clickhouse

-- 在clickhouse中创建mysql引擎表（实时查询mysql数据）
create table mysql_orders (
    order_id uint64,
    order_date date,
    net_amount decimal(10,2)
) engine = mysql('mysql-host:3306', 'analytics_db', 'fact_orders', 'readonly', 'password');

-- 本地物化视图加速查询
create materialized view mv_daily_sales
engine = summingmergetree()
order by (order_date)
as select 
    order_date,
    sum(net_amount) as total_sales,
    count() as order_count
from mysql_orders
group by order_date;

┌─────────────────────────────────────────────────────────────┐
│                        前端展示层                            │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│  │ 实时大屏  │  │ metabase │  │ superset │  │ grafana  │   │
│  │ (websocket)│  │ (自助bi) │  │ (高级可视化)│  │ (监控告警)│   │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘   │
└───────┼────────────┼────────────┼────────────┼───────────┘
        │            │            │            │
        └────────────┴────────────┴────────────┘
                          │
                    ┌─────┴─────┐
                    │  api网关   │  ← 统一查询接口/权限控制
                    │ (graphql) │
                    └─────┬─────┘
                          │
        ┌─────────────────┼─────────────────┐
        │                 │                 │
   ┌────┴────┐      ┌────┴────┐      ┌────┴────┐
   │ mysql   │      │clickhouse│      │  redis  │
   │ (热数据) │      │ (冷分析) │      │ (缓存)  │
   │ 7天数据 │      │ 历史数据 │      │ 实时指标 │
   └────┬────┘      └────┬────┘      └────┬────┘
        │                 │                 │
        └─────────────────┼─────────────────┘
                          │
                    ┌─────┴─────┐
                    │  etl管道   │
                    │(airflow+  │
                    │ canal+     │
                    │ flink)     │
                    └─────┬─────┘
                          │
        ┌─────────────────┼─────────────────┐
        │                 │                 │
   ┌────┴────┐      ┌────┴────┐      ┌────┴────┐
   │ 业务mysql│      │ 日志系统 │      │ 第三方api│
   │ (订单/用户)│     │ (kafka) │      │ (广告/支付)│
   └─────────┘      └─────────┘      └─────────┘