PHP大数据处理技巧：高效处理CSV 使用PHP处理百万级数据导入导出

file_get_contents()

file()

fopen()

fgetcsv()

function readCsvRows(string $filePath): \Generator
{
    if (!file_exists($filePath) || !is_readable($filePath)) {
        throw new \RuntimeException("文件不存在或不可读: {$filePath}");
    }

    if (($handle = fopen($filePath, 'r')) !== false) {
        // 跳过CSV头部（如果存在）
        // fgetcsv($handle);
        while (($data = fgetcsv($handle)) !== false) {
            yield $data; // 每次迭代返回一行数据，而不是全部加载
        }
        fclose($handle);
    } else {
        throw new \RuntimeException("无法打开文件: {$filePath}");
    }
}

INSERT INTO ... VALUES (), (), ...

// 假设这是你的数据库连接 $pdo
// $pdo->beginTransaction();

$batchSize = 2000; // 每批处理的行数
$rowsToInsert = [];
$counter = 0;

foreach (readCsvRows('your_large_file.csv') as $rowData) {
    // 假设你的CSV数据和数据库表结构匹配，或者需要一些转换
    $rowsToInsert[] = [
        'column1' => $rowData[0],
        'column2' => $rowData[1],
        // ...
    ];
    $counter++;

    if ($counter % $batchSize === 0) {
        // 执行批处理插入
        insertBatchIntoDatabase($pdo, $rowsToInsert);
        $rowsToInsert = []; // 清空，准备下一批
        // 可选：在这里提交一次事务，或者在循环结束后一次性提交
        // $pdo->commit();
        // $pdo->beginTransaction();
    }
}

// 处理剩余不足一批的数据
if (!empty($rowsToInsert)) {
    insertBatchIntoDatabase($pdo, $rowsToInsert);
}

// $pdo->commit(); // 最终提交事务

function insertBatchIntoDatabase(\PDO $pdo, array $batchData): void
{
    if (empty($batchData)) {
        return;
    }

    $placeholders = [];
    $values = [];
    $columns = implode(', ', array_keys($batchData[0])); // 假设所有行的键都相同

    foreach ($batchData as $row) {
        $rowPlaceholders = [];
        foreach ($row as $value) {
            $rowPlaceholders[] = '?';
            $values[] = $value;
        }
        $placeholders[] = '(' . implode(', ', $rowPlaceholders) . ')';
    }

    $sql = "INSERT INTO your_table ({$columns}) VALUES " . implode(', ', $placeholders);
    $stmt = $pdo->prepare($sql);
    $stmt->execute($values);
}

fputcsv()

function exportLargeCsv(string $filePath, \PDO $pdo): void
{
    if (($handle = fopen($filePath, 'w')) === false) {
        throw new \RuntimeException("无法创建或写入文件: {$filePath}");
    }

    // 写入CSV头部
    fputcsv($handle, ['Header1', 'Header2', 'Header3']);

    // 假设你的数据表很大，需要分批查询
    $offset = 0;
    $limit = 5000;
    while (true) {
        $stmt = $pdo->prepare("SELECT col1, col2, col3 FROM your_large_table LIMIT :limit OFFSET :offset");
        $stmt->bindValue(':limit', $limit, \PDO::PARAM_INT);
        $stmt->bindValue(':offset', $offset, \PDO::PARAM_INT);
        $stmt->execute();

        $hasRows = false;
        while ($row = $stmt->fetch(\PDO::FETCH_ASSOC)) {
            fputcsv($handle, array_values($row)); // 写入一行
            $hasRows = true;
        }

        if (!$hasRows) {
            break; // 没有更多数据了
        }
        $offset += $limit;
    }

    fclose($handle);
}

file_get_contents()

file_get_contents()

memory_limit

file()

yield

foreach

yield

readCsvRows

Generator

foreach

fgetcsv

yield

INSERT

INSERT

INSERT INTO your_table (col1, col2) VALUES (v1, v2), (v3, v4), ...