Ejemplo de procesamiento por lotes con PHP

El procesamiento por lotes (batch processing) en PHP es la ejecución de tareas que implican una gran cantidad de datos u operaciones de forma secuencial y automatizada (actualizar bases de datos, importar grandes cantidades de datos o ejecutar cálculos complejos), generalmente sin intervención humana directa.

En este post implementaremos un ejemplo para importar millones de registros que importados de la forma tradicional nos traerian problemas de memoria o haría que el script se cuelgue si el volumen de datos es extremadamente grande, por lo que la mejor forma es procesar los datos en pequeños bloques (chunks), por ejemplo, en lugar de cargar 1 millón de registros a la vez, se cargan 1000 registros, se procesan, se guardan, y luego se cargan los siguientes 1000, y así sucesivamente hasta concluir con todos los registros.

Se importará de un archivo CSV de 1.030.291 registros de estudiantes, los campos son «name» y «email».

Y nuestro script «students.php» es:

<?php

// --- Configuración ---
const DB_HOST = 'localhost';
const DB_NAME = '__TU_BASE_DE_DATOS__';
const DB_USER = '__TU_USUARIO__';
const DB_PASS = '__TU_CONTRASEÑA__';
const CSV_FILE = 'students.csv';
const CHUNK_SIZE = 1000; // Número de registros a procesar por lote
const CSV_DELIMITER = ';'; // Delimitador del CSV

echo "Iniciando procesamiento por lotes de importación de registros desde CSV.\n";
echo "Archivo CSV: " . CSV_FILE . "\n";
echo "Tamaño del lote (chunk): " . CHUNK_SIZE . " registros.\n";
echo "--------------------------------------------------------\n";

// --- Conexión a la Base de Datos ---
try {
    $pdo = new PDO(
        "mysql:host=" . DB_HOST . ";dbname=" . DB_NAME . ";charset=utf8mb4",
        DB_USER,
        DB_PASS,
        [
            PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, // Lanzar excepciones en caso de error
            PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC, // Obtener resultados como arrays asociativos
        ]
    );
    echo "Conexión a la base de datos '" . DB_NAME . "' establecida con éxito.\n";
} catch (PDOException $e) {
    echo "ERROR: Fallo en la conexión a la base de datos: " . $e->getMessage() . "\n";
    exit(1);
}

// --- Preparar la sentencia SQL para inserción (uso de INSERT IGNORE para idempotencia) ---
$stmt = $pdo->prepare("INSERT IGNORE INTO students (name, email) VALUES (:name, :email)");

// --- Procesamiento del archivo CSV por lotes ---
$processedRecords = 0;
$chunkCount = 0;
$handle = null; // Para el manejador del archivo CSV

try {
    if (!file_exists(CSV_FILE)) {
        throw new Exception("ERROR: El archivo CSV '" . CSV_FILE . "' no se encontró.\n");
    }

    $handle = fopen(CSV_FILE, 'r');
    if ($handle === false) {
        throw new Exception("ERROR: No se pudo abrir el archivo CSV '" . CSV_FILE . "'.\n");
    }

    // Saltar la fila de encabezados si existe
    $header = fgetcsv($handle, 0, CSV_DELIMITER);
    if ($header === false) {
        throw new Exception("ERROR: El archivo CSV está vacío o no tiene encabezados.\n");
    }
    echo "Encabezados del CSV: " . implode(", ", $header) . "\n";

    $chunk = []; // Array para almacenar los registros del lote actual

    while (($row = fgetcsv($handle, 0, CSV_DELIMITER)) !== false) {        

        // Asignar los campos por nombre (asumiendo que 'name' es la primera y 'email' la segunda)
        $data = [
            'name' => trim($row[0]),
            'email' => trim($row[1]),
        ];

        // Validar que los datos no estén vacíos (básica)
        if (empty($data['name']) || empty($data['email'])) {
            echo "ADVERTENCIA: Campos 'name' o 'email' vacíos, saltando fila: " . implode(CSV_DELIMITER, $row) . "\n";
            continue;
        }

        $chunk[] = $data;

        // Si el chunk alcanza el tamaño definido, procesarlo
        if (count($chunk) >= CHUNK_SIZE) {
            $chunkCount++;
            echo "--- Procesando Lote #" . $chunkCount . " (" . count($chunk) . " registros) ---\n";
            $pdo->beginTransaction(); 
            try {
                foreach ($chunk as $record) {
                    $stmt->execute([
                        ':name' => $record['name'],
                        ':email' => $record['email'],
                    ]);
                    $processedRecords++;
                }
                $pdo->commit(); 
                echo "Lote #" . $chunkCount . " procesado con éxito. Registros procesados: " . $processedRecords . "\n";
            } catch (PDOException $e) { // Deshacer la transacción en caso de error
                $pdo->rollBack(); 
                echo "ERROR en el procesamiento del Lote #" . $chunkCount . ": " . $e->getMessage() . "\n";
            }
            $chunk = []; // Resetear el chunk para el siguiente lote
        }
    }

    // Se procesa los registros restantes que no forman un chunk completo
    if (!empty($chunk)) {
        $chunkCount++;
        echo "--- Procesando Lote Final #" . $chunkCount . " (" . count($chunk) . " registros restantes) ---\n";
        $pdo->beginTransaction();
        try {
            foreach ($chunk as $record) {
                $stmt->execute([
                    ':name' => $record['name'],
                    ':email' => $record['email'],
                ]);
                $processedRecords++;
            }
            $pdo->commit();
            echo "Lote final #" . $chunkCount . " procesado con éxito. Total registros procesados: " . $processedRecords . "\n";
        } catch (PDOException $e) {
            $pdo->rollBack();
            echo "ERROR en el procesamiento del Lote Final #" . $chunkCount . ": " . $e->getMessage() . "\n";
        }
    }

    echo "\n--------------------------------------------------------\n";
    echo "Procesamiento por lotes completado.\n";
    echo "Total de registros procesados con éxito: " . $processedRecords . "\n";

} catch (Exception $e) {
    echo "ERROR CRÍTICO: " . $e->getMessage() . "\n";
    echo "El procesamiento ha fallado.\n";
    exit(1);
} finally {
    if ($handle) {
        fclose($handle); 
    }    
    $pdo = null;
}

?>

Tanto el script students.php como el archivo students.csv deben estar en el mismo directorio.

Debemos asegurarnos que el servidor tenga suficientes recursos (CPU, RAM) para manejar las tareas por lotes, especialmente si se ejecutan en paralelo o durante horas pico. En nuestro caso, una prueba local, podemos modificar esos valores en apache.

Abrimos una consola y ejecutamos:

php students.php

esperamos unos segundos, la tarea puede tardar dependiendo de los recursos de nuestro equipo.

En el ejemplo del post, si se encuentra un registro vacio, lo ignora y continua con el resto de los registros. Si se produce un error al importar un lote, se notifica por pantalla y el lote completo no se registra, se continua con el resto de los datos, sin embargo en producción se desea contar con todos los datos, es asi que lo ideal sería registrar los mismo en un log para que una vez terminado el script, poder analizar e importar los registros fallidos.

enjoy!!!