如何同步SQL数据库与德州混合饮料报告API数据并解决重复问题

为满足数据分析需求，我将德州混合饮料报告API（约300万条数据）存储到SQL数据库，避免频繁调用API产生高额成本。目前数据库数据滞后约一个月，需要实现与API数据的同步——该API持续新增数据且部分字段值会更新。

原本计划在网页加载时运行脚本：检查最新月份数据是否在库中，若不存在则获取整月数据进行增改，直至找到匹配数据。但脚本运行后数据库出现重复数据，附上PHP脚本寻求帮助与优化建议。

• 唯一键缺失：ON DUPLICATE KEY UPDATE 依赖表的唯一键（或主键）触发更新逻辑，但当前脚本未给表设置明确的唯一约束，导致即使存在相同记录，也只会执行插入而非更新。
• 存在性检查逻辑缺陷：检查记录是否存在时，将beer_receipts和total_receipts作为判断条件，但API中这些字段是可能更新的。如果旧记录的金额字段已变化，会误判为“不存在”，进而重复插入。
• 日期处理漏洞：循环中$last_date按天递减，但API返回的是月度数据（obligation_end_date_yyyymmdd是每月最后一天），按天递减会重复处理同一个月的多次请求，增加重复插入风险。

1. 数据库表添加唯一约束 #

在mixed_bev_data表中创建复合唯一键，确保同一地点、同一月份的记录唯一：

ALTER TABLE mixed_bev_data 
ADD UNIQUE KEY unique_location_month (location_name, location_address, record_end_date);

这是让ON DUPLICATE KEY UPDATE生效的核心前提。

2. 修正存在性检查逻辑 #

移除存在性检查中对金额字段的依赖，只通过地点+日期判断记录是否存在——金额是可能更新的字段，不适合作为存在性判断依据：

// 原存在性检查SQL
$sql = "SELECT COUNT(*) FROM mixed_bev_data WHERE location_address='" . $location_address . "' AND record_end_date='" . $record_end_date . "' AND location_name='" . $location_name . "' AND beer_receipts=" . $row['beer_receipts'] . " AND total_receipts=" . $row['total_receipts'];

// 修改后
$sql = "SELECT COUNT(*) FROM mixed_bev_data WHERE location_address='" . $location_address . "' AND record_end_date='" . $record_end_date . "' AND location_name='" . $location_name . "'";

3. 优化日期循环逻辑 #

由于API返回的是月度数据，循环应按月份递减而非按天，避免重复处理同一月份：

// 原日期递减逻辑
$last_date = date('Y-m-d', strtotime($record_end_date . ' - 1 day'));

// 修改后：直接跳转到上月最后一天
$last_date = date('Y-m-t', strtotime($record_end_date . ' -1 month'));

4. 批量插入优化 #

单次插入单条记录效率极低，尤其是处理月度大量数据时，改用批量插入减少数据库交互次数：

// 替换原foreach中的单条插入逻辑
$values = [];
foreach ($data as $row) {
    // 保留原字段转义处理逻辑
    $taxpayer_name = mysqli_real_escape_string($conn, $row['taxpayer_name']);
    $location_name = mysqli_real_escape_string($conn, $row['location_name']);
    $location_address = mysqli_real_escape_string($conn, $row['location_address']);
    $location_city = mysqli_real_escape_string($conn, $row['location_city']);
    $location_state = mysqli_real_escape_string($conn, $row['location_state']);
    $location_zip = mysqli_real_escape_string($conn, $row['location_zip']);
    $record_end_date = date('Y-m-d', strtotime($row['obligation_end_date_yyyymmdd']));
    $beer_receipts = intval($row['beer_receipts']);
    $total_receipts = intval($row['total_receipts']);
    
    // 组装批量插入值
    $values[] = "('$taxpayer_name', '$location_name', '$location_address', '$location_city', '$location_state', '$location_zip', '$record_end_date', $beer_receipts, $total_receipts)";
}

// 执行批量插入
if (!empty($values)) {
    $sql = "INSERT INTO mixed_bev_data (taxpayer_name, location_name, location_address, location_city, location_state, location_zip, record_end_date, beer_receipts, total_receipts)
    VALUES " . implode(',', $values) . "
    ON DUPLICATE KEY UPDATE beer_receipts = VALUES(beer_receipts), total_receipts = VALUES(total_receipts), time = CURRENT_TIMESTAMP();";
    
    if (!mysqli_query($conn, $sql)) {
        echo "Error: " . $sql . "<br>" . mysqli_error($conn);
    }
}

5. 脚本执行时机优化 #

不要在网页加载时运行同步脚本，300万条数据的同步会导致页面超时。改用定时任务（Cron） 定期执行，比如每天凌晨执行一次，避免影响用户体验。

function update_mixed_bev($conn)
{
    $last_date = date('Y-m-t'); // 初始化为当月最后一天

    $count = 0;
    while ($count == 0) {
        // 获取指定日期之前的最新一条记录
        $url = 'https://data.texas.gov/resource/naix-2893.json?$limit=1&$where=obligation_end_date_yyyymmdd%20<=%20%27' . $last_date . '%27&$order=obligation_end_date_yyyymmdd%20DESC';
        $json = file_get_contents($url);
        $data = json_decode($json, true);

        if (empty($data)) {
            break; // 无数据时退出循环
        }

        $row = $data[0];

        // 检查该地点+日期的记录是否存在（移除金额字段判断）
        $location_address = mysqli_real_escape_string($conn, $row['location_address']);
        $location_name = mysqli_real_escape_string($conn, $row['location_name']);
        $record_end_date = date('Y-m-d', strtotime($row['obligation_end_date_yyyymmdd']));
        $sql = "SELECT COUNT(*) FROM mixed_bev_data WHERE location_address='" . $location_address . "' AND record_end_date='" . $record_end_date . "' AND location_name='" . $location_name . "'";

        $result = mysqli_query($conn, $sql);
        $count = mysqli_fetch_array($result)[0];

        if ($count == 0) {
            // 获取该月份的所有数据
            $url = 'https://data.texas.gov/resource/naix-2893.json?$where=obligation_end_date_yyyymmdd%20=%20%27' . $record_end_date . '%27&$order=obligation_end_date_yyyymmdd%20DESC';
            $json = file_get_contents($url);
            $month_data = json_decode($json, true);

            if (!empty($month_data)) {
                $values = [];
                foreach ($month_data as $row) {
                    $taxpayer_name = mysqli_real_escape_string($conn, $row['taxpayer_name']);
                    $location_name = mysqli_real_escape_string($conn, $row['location_name']);
                    $location_address = mysqli_real_escape_string($conn, $row['location_address']);
                    $location_city = mysqli_real_escape_string($conn, $row['location_city']);
                    $location_state = mysqli_real_escape_string($conn, $row['location_state']);
                    $location_zip = mysqli_real_escape_string($conn, $row['location_zip']);
                    $record_end_date = date('Y-m-d', strtotime($row['obligation_end_date_yyyymmdd']));
                    $beer_receipts = intval($row['beer_receipts']);
                    $total_receipts = intval($row['total_receipts']);

                    $values[] = "('$taxpayer_name', '$location_name', '$location_address', '$location_city', '$location_state', '$location_zip', '$record_end_date', $beer_receipts, $total_receipts)";
                }

                // 批量插入并处理更新
                $sql = "INSERT INTO mixed_bev_data (taxpayer_name, location_name, location_address, location_city, location_state, location_zip, record_end_date, beer_receipts, total_receipts)
                VALUES " . implode(',', $values) . "
                ON DUPLICATE KEY UPDATE beer_receipts = VALUES(beer_receipts), total_receipts = VALUES(total_receipts), time = CURRENT_TIMESTAMP();";

                if (!mysqli_query($conn, $sql)) {
                    echo "Error: " . $sql . "<br>" . mysqli_error($conn);
                }
            }
        }

        // 跳转到上月最后一天，避免按天循环
        $last_date = date('Y-m-t', strtotime($record_end_date . ' -1 month'));
    }
}

内容的提问来源于stack exchange，提问作者Eric Love