【实时决策的脉搏】——构建基于MySQL的企业级数据编织平台，赋能智能商业决策

    -- 创建审计日志表
    CREATE TABLE IF NOT EXISTS audit_log (
    id INT AUTO_INCREMENT PRIMARY KEY,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    user_name VARCHAR(255) NOT NULL,
    action_type ENUM('SELECT', 'INSERT', 'UPDATE', 'DELETE') NOT NULL,
    affected_table VARCHAR(255),
    affected_rows INT,
    sql_statement TEXT,
    INDEX idx_user_action (user_name, action_type)
    );

    DELIMITER
$$
    
    -- 定义通用触发器模板
    CREATE TRIGGER trg_after_insert_general
    AFTER INSERT ON your_protected_table FOR EACH ROW
    BEGIN
    INSERT INTO audit_log (user_name, action_type, affected_table, affected_rows, sql_statement)
    VALUES (USER(), 'INSERT', 'your_protected_table', 1, CONCAT('INSERT INTO ', 'your_protected_table', ' VALUES(...)'));
    END$$
    
    CREATE TRIGGER trg_after_update_general
    AFTER UPDATE ON your_protected_table FOR EACH ROW
    BEGIN
    INSERT INTO audit_log (user_name, action_type, affected_table, affected_rows, sql_statement)
    VALUES (USER(), 'UPDATE', 'your_protected_table', 1, CONCAT('UPDATE ', 'your_protected_table', ' SET ... WHERE ...'));
    END$$
    
    CREATE TRIGGER trg_after_delete_general
    AFTER DELETE ON your_protected_table FOR EACH ROW
    BEGIN
    INSERT INTO audit_log (user_name, action_type, affected_table, affected_rows, sql_statement)
    VALUES (USER(), 'DELETE', 'your_protected_table', 1, CONCAT('DELETE FROM ', 'your_protected_table', ' WHERE ...'));
    END$$
    
    DELIMITER ;

    version: '2'
    services:
      connect:
    image: debezium/connect:1.6
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: broker:9092
      CONNECT_GROUP_ID: connector-group
      CONNECT_CONFIG_STORAGE_TOPIC: my-connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: my-connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: my-connect-status
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_INTERNAL_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_INTERNAL_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_PLUGIN_PATH: /kafka/connect/debezium-mysql-connector
    depends_on:
      - zookeeper
      - broker
      - mysql
      mysql:
    image: mysql:8.0
    ports:
      - "3306:3306"
    environment:
      MYSQL_ROOT_PASSWORD: password
      MYSQL_DATABASE: testdb
      MYSQL_USER: user
      MYSQL_PASSWORD: password
    volumes:
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
    command: --server-id=1 --log-bin=mysql-bin.log --binlog-format=ROW --gtid-mode=ON --enforce-gtid-consistency

    -- 创建测试数据库及表结构
    CREATE DATABASE IF NOT EXISTS testdb;
    USE testdb;
    CREATE TABLE IF NOT EXISTS orders (
    id INT AUTO_INCREMENT PRIMARY KEY,
    customer_id VARCHAR(50),
    order_date DATE,
    total_amount DECIMAL(10, 2)
    );
    
    -- 启用GTID复制模式
    SET GLOBAL gtid_mode = ON;
    SET GLOBAL enforce_gtid_consistency = ENFORCING;
    
    -- 设置二进制日志格式为行级（ROW）
    SET GLOBAL binlog_format = 'ROW';
    
    -- 确保binlog已启用
    SHOW VARIABLES LIKE 'log_bin';

    {
      "name": "mysql-connector",
      // 其他配置项...
    }

    docker pull prom/prometheus:latest
    docker run -d --name prometheus -p 9090:9090 -v $(pwd)/prometheus.yml:/etc/prometheus/prometheus.yml prom/prometheus:latest

    scrape_configs:
    - job_name: 'mysql'
      static_configs:
      - targets: ['localhost:9104'] # MySQL Exporter监听地址

    docker pull prom/mysqld-exporter
    docker run -d --name mysqld-exporter \
    -e DATA_SOURCE_NAME="user:password@(localhost:3306)/" \
    -p 9104:9104 \
    prom/mysqld-exporter

    docker pull grafana/grafana:latest
    docker run -d --name grafana -p 3000:3000 grafana/grafana:latest

    -- 获取InnoDB行级锁等待情况
    SELECT r.trx_id waiting_trx_id, r.trx_mysql_thread_id waiting_thread,
    r.trx_query waiting_query, b.trx_id blocking_trx_id,
    b.trx_mysql_thread_id blocking_thread, b.trx_query blocking_query
    FROM information_schema.innodb_lock_waits w
    JOIN information_schema.innodb_trx b ON b.trx_id = w.blocking_trx_id
    JOIN information_schema.innodb_trx r ON r.trx_id = w.requesting_trx_id;
    
    -- 统计各表的缓存命中率
    SELECT table_name, SUM(CASE WHEN index_name IS NULL THEN rows_examined ELSE 0 END) / SUM(rows_examined) AS cache_hit_ratio
    FROM performance_schema.table_io_waits_summary_by_index_usage
    GROUP BY table_name;