需求

對平均 1200w 數據量的數據表進行優化
數據表中有 2016年，2017 年，2018 年，2019 年數據
只查詢最近半年的數據
後台增加歷史數據查詢功能
盡量減少代碼改動

數據表

積分日誌表 tb_user_points_log
虛擬充值表 tb_order_recharge
虛擬充值執行表 tb_order_recharge_do

注意：

先備份數據，在備份的數據表的基礎上進行分表，不直接操作原始表!

步驟

將源數據表備份一份，依次將對應年份的數據歸檔，每成功歸檔一次，就將備份數據表中對應數據刪除（目的減少查詢數據量），最後根據備份表最小 ID，刪除源數據表小於 ID 的所有數據。

該步驟可以直接通過 SQL 執行，也可通過腳本執行。

腳本執行

刪除源數據表數據操作，建議通過手動執行 SQL完成，其他操作通過腳本執行

以積分日誌表 tb_user_points_log 為例

方式一、手動執行SQL

備份 tb_user_points_log 得到 tb_user_points_copy

2016年數據歸檔

將數據表 tb_user_points_copy 2016 年的數據歸檔存入 2016 年數據表 tb_user_points_log_2016

CREATE TABLE tb_user_points_log_2016 LIKE tb_user_points_log_copy;
INSERT INTO tb_user_points_log_2016 SELECT * FROM tb_user_points_log_copy WHERE add_time < 1483200000;

對比數量

SELECT COUNT(id) FROM tb_user_points_log_2016;
SELECT COUNT(id) FROM tb_user_points_log_copy WHERE add_time < 1483200000;

一致則刪除 tb_user_points_copy 的 2016 年數據
```
DELETE FROM tb_user_points_log_copy WHERE add_time < 1483200000;
```
2017年數據歸檔

將數據表 tb_user_points_copy 2017 年的數據歸檔存入 2017 年數據表 tb_user_points_log_2017

CREATE TABLE tb_user_points_log_2017 LIKE tb_user_points_log_copy;
INSERT INTO tb_user_points_log_2017 SELECT * FROM tb_user_points_log_copy WHERE add_time < 1514736000;

對比數量

SELECT COUNT(id) FROM tb_user_points_log_2017;
SELECT COUNT(id) FROM tb_user_points_log_copy WHERE add_time < 1514736000;

一致則刪除 tb_user_points_copy 的 2017 年數據
```
DELETE FROM tb_user_points_log_copy WHERE add_time < 1514736000;
```
2018年數據歸檔

將數據表 tb_user_points_copy 2018 年的數據歸檔存入 2018 年數據表 tb_user_points_log_2018

CREATE TABLE tb_user_points_log_2018 LIKE tb_user_points_log_copy;
INSERT INTO tb_user_points_log_2018 SELECT * FROM tb_user_points_log_copy WHERE add_time < 1546272000;

對比數量

SELECT COUNT(id) FROM tb_user_points_log_2018;
SELECT COUNT(id) FROM tb_user_points_log_copy WHERE add_time < 1546272000;

一致則刪除 tb_user_points_copy 的 2018 年數據
```
DELETE FROM tb_user_points_copy WHERE add_time < 1546272000;
```
2019年數據歸檔

現在是 11 月，將 5 月之前的數據歸檔

CREATE TABLE tb_user_points_log_2019 LIKE tb_user_points_log_copy;
INSERT INTO tb_user_points_log_2019 SELECT * FROM tb_user_points_log_copy WHERE add_time < 1556640000;

對比數量

SELECT COUNT(id) FROM tb_user_points_log_2019;
SELECT COUNT(id) FROM tb_user_points_log_copy WHERE add_time < 1556640000;

一致則刪除 tb_user_points_copy 的 2019 年 5 月之前的數據
```
DELETE FROM tb_user_points_log_copy WHERE add_time < 1556640000;
```
刪除原始數據

根據最小 tb_user_points_copy 的最小 ID，刪除原始表小於 ID 的所有數據

DELETE FROM tb_user_points_log WHERE id < (SELECT id FROM tb_user_points_log_copy ORDER BY id asc LIMIT 1);

刪除臨時表
```
DELETE FROM tb_user_points_log_copy;
```
數據表分表完成！
增量歸檔

每日凌晨，執行腳本將最近半年之前的數據歸檔

方式二、腳本執行

<?php
/**
 * Description: 將6個月前數據歸檔
 */

namespace wladmin\cmd;


use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;

class DataArchiving extends Command
{
    protected function configure()
    {
        $this->setName('DataArchiving')->setDescription('將6個月前數據歸檔');
    }

    /**
     * 將6個月前數據歸檔
     * php think DataArchiving
     * @param Input $input
     * @param Output $output
     *
     * @return int|void|null
     */
    protected function execute(Input $input, Output $output)
    {
        try {
            $this->archiveData('tb_user_points_log', 'id', 'add_time');
            $this->archiveData('tb_order_recharge', 'or_id', 'create_time');
            $this->archiveData('tb_order_recharge_do', 'ord_id', 'create_time');
            echo '歸檔完成';
        } catch (\Exception $e) {
            mylog($e->getMessage(),'歸檔發生錯誤:'.PHP_EOL);
        }
    }
  
         /**
     * 歸檔數據表
     * @param string $sourceTable 源數據表名
     * @param string $primaryKey 主鍵名
     * @param string $timeKey 時間鍵名
     *
     * @author Dong.cx 2019-11-18 18:05
     * @version V4.0.1
     */
    private function archiveData($sourceTable, $primaryKey, $timeKey)
    {
        try {
            date_default_timezone_set('PRC');
            // 1.複製源數據表
            $copyTable = $sourceTable . '_copy';
            $isExist = $this->tableExist($copyTable, $sourceTable);
            if (!$isExist) {
                echo "開始複製源數據表{$copyTable}" . PHP_EOL;
                $archivingTimeLine = time();
                $sql = "INSERT IGNORE INTO {$copyTable} SELECT * FROM {$sourceTable} WHERE {$timeKey} < {$archivingTimeLine}";
                Db::execute($sql);
                echo "複製源數據表{$copyTable}完成" . PHP_EOL;
            }
            echo "{$copyTable} 開始歸檔" . PHP_EOL;
            // 歸檔
            $this->archive(2016, $sourceTable, $primaryKey, $timeKey);
            $this->archive(2017, $sourceTable, $primaryKey, $timeKey);
            $this->archive(2018, $sourceTable, $primaryKey, $timeKey);
            $this->archive(2019, $sourceTable, $primaryKey, $timeKey);
            echo "{$copyTable} 歸檔完成";

        } catch (\Exception $e) {
            echo '歸檔發生錯誤:' . $e->getMessage() .PHP_EOL;
        }
    }

    /**
     * 歸檔操作
     * @param int $year 年份
     * @param string $sourceTable 源數據表名
     * @param string $primaryKey 主鍵名
     * @param string $timeKey 時間鍵名
     *
     * @return bool
     * @throws \Exception
     * @author Dong.cx 2019-11-18 18:12
     * @version V4.0.1
     */
    private function archive($year, $sourceTable, $primaryKey, $timeKey)
    {
        try {
            $copyTable = $sourceTable . '_copy';
            echo "{$copyTable} 開始歸檔{$year}年數據--->" . PHP_EOL;
            if ($year == date('Y')) {
                // 注意現在是 11月份,可以簡單這樣寫,如果是小於6月,則要相應修改
                $archivingTimeLine = strtotime('-6 month', strtotime('today'));
            } else {
                $archivingTimeLine = mktime(0,0,0,1,1,$year+1);
            }

            $sql = "SELECT COUNT({$primaryKey}) as num FROM {$copyTable} WHERE {$timeKey} < {$archivingTimeLine}";
            $res = Db::query($sql);
            if (!$res || !$res[0]['num']) {
                echo "{$copyTable} {$year}年數據歸檔完成,未查詢到需要歸檔的數據" . PHP_EOL;
                return true;
            }

            // 需歸檔數量
            $targetNum = $res[0]['num'];
            // 歸檔表名
            $tableArchivingName = $sourceTable . '_' . $year;
            $this->tableExist($tableArchivingName, $sourceTable);

            // 分批歸檔
            $this->archivingBatch($tableArchivingName, $copyTable, $primaryKey,$timeKey, $archivingTimeLine, $year, $targetNum);

            return true;
        } catch (\Exception $e) {
            throw $e;
        }
    }

    /**
     * 分批歸檔
     * @param string $tableArchivingName 歸檔表名稱
     * @param string $copyTable 複製表名
     * @param string $primaryKey 主鍵名
     * @param string $timeKey 時間鍵
     * @param int $archivingTimeLine 歸檔時間線
     * @param string $year 歸檔年
     * @param int $targetNum 需歸檔的數據量
     *
     * @throws \Exception
     * @author Dong.cx 2019-11-19 13:10
     * @version V4.0.1
     */
    private function archivingBatch($tableArchivingName, $copyTable, $primaryKey,$timeKey, $archivingTimeLine, $year, $targetNum)
    {
        // 歸檔表起始ID
        $res = Db::query("SELECT {$primaryKey} FROM {$tableArchivingName} ORDER BY {$primaryKey} DESC LIMIT 1");
        $startID = $res ? $res[0][$primaryKey] : 0;

        $totalDelNum = 0;
        $batchNum = 10000;
        $taskNum = ceil($targetNum/$batchNum);
        $minID = Db::query("SELECT {$primaryKey} FROM {$copyTable} ORDER BY {$primaryKey} ASC LIMIT 1");
        if (!$minID) throw new \Exception('$minID為空!');
        $minID = $minID[0][$primaryKey];
        $maxID = Db::query("SELECT {$primaryKey} FROM {$copyTable} WHERE {$timeKey} < {$archivingTimeLine} ORDER BY {$primaryKey} DESC LIMIT 1");
        if (!$maxID) throw new \Exception('$max 為空!');
        $maxID = $maxID ? $maxID[0][$primaryKey] : 0;

        for ($i = 1; $i <= $taskNum; $i++) {
            if ($i == $taskNum) {
                // 歸檔
                $sql = "INSERT IGNORE INTO {$tableArchivingName} SELECT * FROM {$copyTable} WHERE {$primaryKey} <= {$maxID} AND {$timeKey} < {$archivingTimeLine}";
                Db::execute($sql);
                // 刪除
                $sql = "DELETE FROM {$copyTable} WHERE {$primaryKey} <= {$maxID} AND {$timeKey} < {$archivingTimeLine}";
                $totalDelNum += Db::execute($sql);
            } else {
                $end = $minID + $i * $batchNum;
                // 歸檔
                $sql = "INSERT IGNORE INTO {$tableArchivingName} SELECT * FROM {$copyTable} WHERE {$primaryKey} <= {$end} AND {$timeKey} < {$archivingTimeLine}";
                Db::execute($sql);
                // 刪除
                $sql = "DELETE FROM {$copyTable} WHERE {$primaryKey} <= {$end} AND {$timeKey} < {$archivingTimeLine}";
                $totalDelNum += Db::execute($sql);
            }
        }
        // 成功歸檔數據量
        $num = Db::query("SELECT COUNT({$primaryKey}) as num FROM {$tableArchivingName} WHERE {$primaryKey} > {$startID}")[0]['num'];
        if ($targetNum != $num) throw new \Exception("歸檔數據不一致,過期數據量{$targetNum},歸檔量{$num},刪除量{$totalDelNum}");
        if ($num != $totalDelNum) throw new \Exception("刪除數據不一致,歸檔量{$num},刪除量{$totalDelNum}");

        echo "{$copyTable} {$year}年數據歸檔完成,過期數據量{$targetNum},歸檔量{$num},刪除量{$totalDelNum}" . PHP_EOL;
        
        // 刪除源數據表數據
        //echo "開始刪除源數據表 {$sourceTable}已歸檔數據" . PHP_EOL；    
        //$num = Db::execute("DELETE FROM {$sourceTable} WHERE {$primaryKey} < (SELECT id FROM {$copyTable} ORDER BY {$primaryKey} asc LIMIT 1)");
       //echo "源數據表 {$sourceTable}已歸檔數據刪除完成,刪除數據量{$num}" . PHP_EOL； 
      
        //echo "開始刪除臨時表 {$copyTable}" . PHP_EOL；    
        // 刪除臨時表
        //Db::execute("DELETE FROM {$copyTable}");
        //echo "臨時表{$copyTable}刪除完成" . PHP_EOL;
    }

最後由於是要刪除源數據表，屬於敏感操作，(腳本最後註釋部分) 建議再複查一次數據歸檔正確性，確認無誤后，手動執行 SQL操作。

DELETE FROM {$sourceTable} WHERE {$primaryKey} < (SELECT {$primaryKey} FROM {$copyTable} ORDER BY {$primaryKey} asc LIMIT 1;
DELETE FROM {$copyTable};

增量歸檔腳本

<?php
/**
 * Description: 將6個月前數據歸檔
 */

namespace wladmin\cmd;


use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\Db;

class DataArchiving extends Command
{
    protected function configure()
    {
        $this->setName('DataArchiving')->setDescription('將6個月前數據歸檔');
    }

    /**
     * 將6個月前數據歸檔
     * php think DataArchiving
     * @param Input $input
     * @param Output $output
     *
     * @return int|void|null
     */
    protected function execute(Input $input, Output $output)
    {
        try {
            $this->archiveDataEveryDay('tb_user_points_log', 'id', 'add_time');
            $this->archiveDataEveryDay('tb_order_recharge', 'or_id', 'create_time');
            $this->archiveDataEveryDay('tb_order_recharge_do', 'ord_id', 'create_time');
            echo '歸檔完成';
        } catch (\Exception $e) {
            mylog($e->getMessage(),'歸檔發生錯誤:'.PHP_EOL);
        }
    }

    /**
     * 歸檔數據
     * @param string $sourceTable 源數據表名
     * @param string $primaryKey 源數據表主鍵名
     * @param string $timeKey 時間控制鍵名
     *
     * @return bool
     * @throws \Exception
     * @author Dong.cx 2019-11-15 18:36
     * @version V4.0.1
     */
    private function archiveDataEveryDay($sourceTable, $primaryKey, $timeKey)
    {
        try {
            //mylog("{$sourceTable} 開始歸檔".PHP_EOL);
            // 歸檔時間線
            $archivingTimeLine = strtotime('-6 month', strtotime('today'));
            // 歸檔表的年份
            $year = date('Y', $archivingTimeLine);
            // 歸檔表名
            $tableArchivingName = $sourceTable . '_' . $year;

            // 需要歸檔的數據量
            $sql = "SELECT COUNT({$primaryKey}) as num FROM {$sourceTable} WHERE {$timeKey} < {$archivingTimeLine}";
            $res = Db::query($sql);
            // 沒有需要歸檔的,直接返回
            if (!$res) {
                mylog("{$sourceTable} 歸檔完成,未查詢到需要歸檔的數據");
                return true;
            }
            $count = $res[0]['num'];

            // 檢測數據表是否存在,不存在則創建
            $this->tableExist($tableArchivingName, $sourceTable);
            $sql = "INSERT IGNORE INTO {$tableArchivingName} SELECT * FROM {$sourceTable} WHERE {$timeKey} < {$archivingTimeLine}";

            // 1.開始歸檔
            // 歸檔表起始ID
            $res = Db::query("SELECT {$primaryKey} FROM {$tableArchivingName} ORDER BY {$primaryKey} DESC LIMIT 1");
            $startID = $res ? $res[0][$primaryKey] : 0;
            Db::execute($sql);
            // 成功歸檔數據量
            $num = Db::query("SELECT COUNT({$primaryKey}) as num FROM {$tableArchivingName} WHERE {$primaryKey} > {$startID}")[0]['num'];
            if ($count != $num) throw new \Exception("歸檔數據不一致,過期數據量{$count},歸檔量{$num}");
            $lastID = Db::query("SELECT {$primaryKey} FROM {$tableArchivingName} ORDER BY {$primaryKey} DESC LIMIT 1")[0][$primaryKey];

            // 2.刪除源數據
            $sql = "DELETE FROM {$sourceTable} WHERE {$primaryKey} <= {$lastID}  AND {$timeKey} < {$archivingTimeLine}";
            $delNum = Db::execute($sql);
            if ($delNum != $count) throw new \Exception("刪除數據不一致,過期數據量{$count},刪除量{$delNum}");
            //mylog("{$sourceTable} 歸檔完成,過期數據量{$count},歸檔量{$count},刪除量{$delNum}" . PHP_EOL);
            return true;
        } catch (\Exception $e) {
            Db::rollback();
            throw $e;
        }
    }

    /**
     * 檢測數據表是否存在,不存在則創建
     * @param $table
     * @param $likeTable
     */
    private function tableExist($table, $likeTable)
    {
        $sql = "SHOW TABLES LIKE '{$table}'";
        $isExist = Db::query($sql);

        if (!$isExist) {
            Db::execute("CREATE TABLE {$table} LIKE {$likeTable}");
        }
    }
}

歷史數據查詢

在數據訪問層中根據需要查詢時間動態修改數據表名即可

這裏使用的是 thinkphp Query 類中的 setTable() 和 getTable()

 if (isset($params['history']) && !empty($params['history'])) {
            $this->model()->setTable($this->model()->getTable().'_'.$params['history']);
        }

遇到的問題

開發中，曾嘗試使用事務控制，數據量太多會導致提交過慢，因此使用邏輯控制
DB 一次性執行100多w刪除操作后，發現程序不繼續向下執行，未找到原因，因此將數據分批進行處理，但是分批可能存在問題，因為主鍵可能不是連續的，如果間隔不大的話，影響不大。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※帶您來了解什麼是 USB CONNECTOR ?

※自行創業缺乏曝光? 下一步”網站設計“幫您第一時間規劃公司的門面形象

※如何讓商品強力曝光呢? 網頁設計公司幫您建置最吸引人的網站，提高曝光率!!

※綠能、環保無空污,成為電動車最新代名詞，目前市場使用率逐漸普及化

※廣告預算用在刀口上，網站設計公司幫您達到更多曝光效益

※試算大陸海運運費!

需求

數據表

步驟

方式一、手動執行SQL

2016年數據歸檔

2017年數據歸檔

2018年數據歸檔

2019年數據歸檔

刪除原始數據

方式二、腳本執行

增量歸檔腳本

歷史數據查詢

遇到的問題