Big Data Techniques in PHP

Chunked Processing

Chunked processing is crucial when dealing with datasets that are too large to fit in memory. This technique involves processing data in smaller, manageable pieces.
Example of class ChunkedProcessor:

<?php

class ChunkedProcessor {
    private $chunkSize;
    private $maxMemoryUsage;
    private $logHandler;

    // Adjusted the constructor to avoid type errors
    public function __construct($chunkSize = 1000, $maxMemoryUsage = '256M', LogHandler $logHandler = null) {
        $this->chunkSize = $chunkSize;
        $this->maxMemoryUsage = $maxMemoryUsage;
        $this->logHandler = $logHandler ?? new LogHandler(); // Default if not provided
    }

    public function processLargeDataset($filename, callable $processor) {
        try {
            // Validate and set memory limit
            if (!ini_set('memory_limit', $this->maxMemoryUsage)) {
                throw new Exception("Failed to set memory limit.");
            }

            if (!file_exists($filename)) {
                throw new Exception("File not found: $filename");
            }

            $handle = fopen($filename, 'r');
            if ($handle === false) {
                throw new Exception("Failed to open file: $filename");
            }

            $stats = [
                'processed_rows' => 0,
                'failed_rows' => 0,
                'start_time' => microtime(true),
                'memory_peak' => 0
            ];

            // Process file in chunks
            while (!feof($handle)) {
                try {
                    $chunk = [];
                    $count = 0;

                    // Build chunk
                    while ($count < $this->chunkSize && ($line = fgets($handle)) !== false) {
                        $decodedLine = json_decode($line, true);
                        if (json_last_error() === JSON_ERROR_NONE) {
                            $chunk[] = $decodedLine;
                        } else {
                            $this->logHandler->warning("Failed to decode line: $line");
                            $stats['failed_rows']++;
                        }
                        $count++;
                    }

                    // Process current chunk
                    $processor($chunk);

                    // Update statistics
                    $stats['processed_rows'] += count($chunk);
                    $stats['memory_peak'] = max($stats['memory_peak'], memory_get_peak_usage(true));

                    // Log progress
                    $this->logProgress($stats);

                    // Clean up
                    unset($chunk);
                    if ($stats['processed_rows'] % ($this->chunkSize * 10) === 0) {
                        gc_collect_cycles();
                    }
                } catch (Exception $e) {
                    $stats['failed_rows'] += count($chunk);
                    $this->logHandler->error("Chunk processing failed: " . $e->getMessage());
                    continue;
                }
            }

            fclose($handle);
            return $this->generateReport($stats);
        } catch (Exception $e) {
            throw new Exception("Dataset processing failed: " . $e->getMessage());
        }
    }

    private function logProgress(array $stats) {
        $memoryUsage = memory_get_usage(true) / 1024 / 1024;
        $timeElapsed = microtime(true) - $stats['start_time'];
        $rowsPerSecond = $stats['processed_rows'] / $timeElapsed;

        $this->logHandler->info(sprintf(
            "Processed %d rows | Memory: %.2f MB | Speed: %.2f rows/sec",
            $stats['processed_rows'],
            $memoryUsage,
            $rowsPerSecond
        ));
    }

    private function generateReport(array $stats) {
        return [
            'total_processed' => $stats['processed_rows'],
            'total_failed' => $stats['failed_rows'],
            'memory_peak_mb' => $stats['memory_peak'] / 1024 / 1024,
            'time_taken_sec' => microtime(true) - $stats['start_time']
        ];
    }
}

// Example of use
class LogHandler {
    public function info($message) {
        echo "\nINFO: $message";
    }
    public function error($message) {
        echo "\nERROR: $message";
    }
    public function warning($message) {
        echo "\nWARNING: $message";
    }
}