如何为PHP项目实现数据归并？

本文目录导读：

1. 合并数组数据（最基础）
2. 数据库表归并（常见场景）
3. 文件数据归并（CSV / JSON）
4. 主键冲突处理（UPSERT 归并）
5. 不同结构的数据归并（对象/数组转换）
6. 性能优化建议
7. 其他：使用队列系统

为PHP项目实现数据归并（Merge），通常指的是将多个数据源（如数据库、API、数组、文件）中的数据进行整合、去重、合并，最终形成一个统一的数据集,具体实现方式取决于你的数据来源和业务需求。

以下是几种常见场景的实现方案：

合并数组数据（最基础）

如果数据来自不同数组（例如从多个API拉取的结果），可以使用 PHP 内置函数。

<?php
$data1 = [
    ['id' => 1, 'name' => 'Alice'],
    ['id' => 2, 'name' => 'Bob'],
];
$data2 = [
    ['id' => 3, 'name' => 'Charlie'],
    ['id' => 1, 'name' => 'Alice'], // 重复数据
];
// 1. 索引数组合并（允许重复）
$merged = array_merge($data1, $data2);
// 2. 关联数组合并（后覆盖前）
$merged = $data1 + $data2;
// 3. 按指定key去重合并（推荐用于数据库行）
$all = array_merge($data1, $data2);
$unique = [];
foreach ($all as $item) {
    $unique[$item['id']] = $item; // 以ID为键，后出现的会覆盖前面的
}
$result = array_values($unique); // 重新索引
print_r($result);
?>

注意：如果数据量较大，避免在循环内使用 in_array() 或 array_unique()，这会导致 O(n²) 复杂度。

数据库表归并（常见场景）

当数据来自多个数据库表或库时，推荐在数据库层面解决（性能最佳）。

方案A：SQL UNION ALL + 去重

SELECT id, name, email FROM users_2023
UNION
SELECT id, name, email FROM users_2024;

• UNION 自动去重（但会排序，可能慢）。
• UNION ALL 保留所有行，更快,适合后续程序去重。

方案B：使用临时表 + INSERT … ON DUPLICATE KEY UPDATE

-- 创建临时合并表
CREATE TEMPORARY TABLE temp_merged LIKE users;
-- 插入第一张表
INSERT INTO temp_merged SELECT * FROM users_2023;
-- 插入第二张表，主键冲突时更新（去重+合并）
INSERT INTO temp_merged 
SELECT * FROM users_2024 
ON DUPLICATE KEY UPDATE 
    name = VALUES(name), 
    email = VALUES(email);

方案C：PHP 分批处理（适合超大表）

<?php
$pdo = new PDO('mysql:host=...', 'user', 'pass');
$pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
// 每次处理10000条
$limit = 10000;
$offset = 0;
while (true) {
    $sql = "INSERT INTO target_table (id, name, email)
            SELECT id, name, email FROM source_table
            WHERE id NOT IN (SELECT id FROM target_table)
            LIMIT $limit OFFSET $offset";
    $affected = $pdo->exec($sql);
    if ($affected == 0) break;
    $offset += $limit;
    usleep(100000); // 短暂休眠，避免数据库过载
}
?>

文件数据归并（CSV / JSON）

CSV 文件归并（去重）

<?php
function mergeCsvFiles($files, $outputFile, $keyField = 'id') {
    $handle = fopen($outputFile, 'w');
    $headersWritten = false;
    $seen = [];
    foreach ($files as $file) {
        $fh = fopen($file, 'r');
        $headers = fgetcsv($fh); // 假设有表头
        if (!$headersWritten) {
            fputcsv($handle, $headers);
            $headersWritten = true;
        }
        while ($row = fgetcsv($fh)) {
            $record = array_combine($headers, $row); // 转为关联数组
            $key = $record[$keyField];
            if (!isset($seen[$key])) {
                $seen[$key] = true;
                fputcsv($handle, $row);
            }
        }
        fclose($fh);
    }
    fclose($handle);
}
?>

主键冲突处理（UPSERT 归并）

当合并数据到已有表时,需要处理主键冲突：

• 忽略冲突：INSERT IGNORE 或 ON CONFLICT ... DO NOTHING
• 覆盖更新：REPLACE INTO（MySQL）或 INSERT … ON DUPLICATE KEY UPDATE
• 聚合合并：例如合并用户积分,求和而不是覆盖。

// 示例：合并用户积分（求和）
$stmt = $pdo->prepare(
    "INSERT INTO user_points (user_id, points, total_orders)
     VALUES (:uid, :points, 1)
     ON DUPLICATE KEY UPDATE 
         points = points + VALUES(points),
         total_orders = total_orders + 1"
);
foreach ($newData as $row) {
    $stmt->execute([':uid' => $row['uid'], ':points' => $row['points']]);
}

不同结构的数据归并（对象/数组转换）

当数据来源结构不同（如一个来源有phone，另一个有email）：

<?php
function normalizeAndMerge($source1, $source2) {
    $merged = [];
    foreach ($source1 as $item) {
        $merged[$item['id']] = [
            'id'    => $item['id'],
            'name'  => $item['name'] ?? '',
            'phone' => $item['phone'] ?? '',
            'email' => '',
        ];
    }
    foreach ($source2 as $item) {
        if (isset($merged[$item['id']])) {
            $merged[$item['id']]['email'] = $item['email'];
        } else {
            $merged[$item['id']] = [
                'id'    => $item['id'],
                'name'  => '',
                'phone' => '',
                'email' => $item['email'],
            ];
        }
    }
    return array_values($merged);
}
?>

性能优化建议

数据量	推荐方案	说明
< 10,000 行	内存数组操作	简单快速
10K – 100W 行	SQL UNION / 临时表	数据库层面处理
> 100W 行	分批处理 + 索引	避免一次性加载，使用 LIMIT/OFFSET
跨系统	文件交换（CSV/Parquet）	先导出再导入，避免网络瓶颈

• 索引：对归并用的 key 字段（如 id、UUID）建立索引。
• 内存：处理大数组时启用 memory_limit 并考虑使用生成器(Generator)。
• 事务：归并多条数据时使用数据库事务,保证一致性。

其他：使用队列系统

对于实时或高频率的归并需求（如日志聚合）,可以用消息队列：

// 伪代码流程
// 1. 各个服务将数据发送到 Redis List / RabbitMQ
// 2. 消费者进程从队列拉取数据
// 3. 使用 ON DUPLICATE KEY UPDATE 合并到汇总表
// 4. 定期清理过期数据

• 小数据：直接 PHP 数组 + array_merge / array_values / 自定义去重。
• 数据库数据：利用 SQL 的 UNION/INSERT ... ON DUPLICATE KEY,性能最好。
• 大数据/跨系统：分批处理或文件交换,避免内存溢出。
• 实时归并：使用消息队列 + 批量Upsert。

根据你的具体场景（数据规模、来源、实时性要求）,选择合适的方法即可。