基于php-resque解析验证大文件数据并导入MySQL的技术问询

1. Resque Job Class Foundation #

First, let's build a robust Resque Job class that handles file streaming, validation, batch inserts, and progress tracking. This avoids loading the entire large file into memory and ensures we keep the ItemsFile table updated with real-time status:

class ProcessLargeFileJob
{
    public function perform()
    {
        // Pull the target file ID from Resque job arguments
        $fileId = $this->args['file_id'];
        
        // Fetch the file record and guard against duplicate processing
        $itemsFile = ItemsFile::find($fileId);
        if (!$itemsFile || $itemsFile->processed) {
            throw new Exception("File either doesn't exist or is already processed");
        }

        // Initialize counters and batch settings
        $validCount = 0;
        $invalidCount = 0;
        $processedCount = 0;
        $batchSize = 1000; // Tune this based on your server memory limits
        $validItems = [];

        // Open the file in stream mode to avoid memory overload
        $handle = fopen($itemsFile->filepath, 'r');
        if (!$handle) {
            throw new Exception("Failed to access file at: {$itemsFile->filepath}");
        }

        try {
            // Skip header line if your file includes one (uncomment if needed)
            // fgets($handle);

            while (($line = fgets($handle)) !== false) {
                $line = trim($line);
                if (empty($line)) continue;

                // Parse the line into structured data (adjust for your file format)
                $parsedData = $this->parseLine($line);
                
                // Validate against your business rules
                if ($this->validateLine($parsedData)) {
                    $validItems[] = [
                        'uid' => $parsedData['uid'],
                        'item' => $parsedData['item'],
                        'file_id' => $fileId,
                        'created_at' => date('Y-m-d H:i:s')
                    ];
                    $validCount++;

                    // Insert batch once we hit the batch size
                    if (count($validItems) >= $batchSize) {
                        $this->insertBatch($validItems);
                        $validItems = [];
                    }
                } else {
                    $invalidCount++;
                }

                $processedCount++;

                // Update progress every 500 lines to minimize DB hits
                if ($processedCount % 500 === 0) {
                    $itemsFile->processed_items = $processedCount;
                    $itemsFile->save();
                }
            }

            // Insert any remaining valid items in the final batch
            if (!empty($validItems)) {
                $this->insertBatch($validItems);
            }

            // Finalize the file record with complete stats
            $itemsFile->valid_items = $validCount;
            $itemsFile->invalid_items = $invalidCount;
            $itemsFile->processed_items = $processedCount;
            $itemsFile->processed = true;
            $itemsFile->save();

        } catch (Exception $e) {
            // Mark the file as failed and log the error (add an `error_message` column to ItemsFile if needed)
            $itemsFile->processed = false;
            $itemsFile->error_message = $e->getMessage();
            $itemsFile->save();
            throw $e; // Let Resque handle failure logging and retries
        } finally {
            fclose($handle);
        }
    }

    /**
     * Parse a single line into an array (customize for CSV/TSV/other formats)
     */
    private function parseLine(string $line): array
    {
        // Example for CSV files:
        $parts = str_getcsv($line);
        return [
            'uid' => $parts[0] ?? null,
            'item' => $parts[1] ?? null
        ];

        // For TSV, replace with:
        // return explode("\t", $line);
    }

    /**
     * Validate parsed data against your rules (expand as needed)
     */
    private function validateLine(array $data): bool
    {
        // Check required fields exist
        if (empty($data['uid']) || empty($data['item'])) {
            return false;
        }

        // Validate UID format (e.g., numeric, UUID)
        if (!is_numeric($data['uid'])) {
            return false;
        }

        // Add more rules: item length constraints, allowed values, etc.
        return true;
    }

    /**
     * Insert a batch of valid items into the Item table efficiently
     */
    private function insertBatch(array $items): void
    {
        if (empty($items)) return;

        // Build bulk insert query (use your DB abstraction layer if available)
        $columns = implode(', ', array_keys($items[0]));
        $valueStrings = [];

        foreach ($items as $item) {
            $escapedValues = array_map(function($val) {
                return is_string($val) ? "'" . mysqli_real_escape_string($val) . "'" : $val;
            }, $item);
            $valueStrings[] = '(' . implode(', ', $escapedValues) . ')';
        }

        $query = "INSERT INTO Item ($columns) VALUES " . implode(', ', $valueStrings);
        
        // Execute with your DB connection (example using PDO)
        $pdo = new PDO('mysql:host=localhost;dbname=your_database', 'user', 'password');
        $pdo->exec($query);
    }
}

2. Key Optimizations & Best Practices #

• Streamed File Reading: Using fgets() ensures we only load one line at a time, avoiding out-of-memory errors with massive files.
• Batched Inserts: Grouping inserts into batches of 1000-5000 records cuts down on database round-trips, drastically speeding up imports.
• Progress Tracking: Updating processed_items periodically (not every line) balances real-time visibility with database performance.
• File Locking: Add a locked_at column to ItemsFile to prevent multiple workers from processing the same file. Lock the file when starting, release on completion/failure.
• Failure Resilience: The try/catch block ensures we always update the ItemsFile record, even if the job crashes, making retries or debugging easy.

3. Enqueuing the Job #

To kick off processing for a file, enqueue the job with the target file ID:

Resque::enqueue('file_processing', 'ProcessLargeFileJob', ['file_id' => $yourFileId]);

Replace 'file_processing' with your preferred Resque queue name.

4. Optional Enhancements #

• Invalid Item Logging: Add an InvalidItem table to store invalid line content and error messages, allowing you to debug and reprocess bad entries later.
• User-Facing Progress: Build a simple API endpoint to fetch ItemsFile stats, then use polling or websockets to display progress to users in real time.
• Retry Logic: Use Resque's built-in retry functionality or a dedicated retry queue to automatically reprocess files that failed due to transient issues (e.g., database timeouts).

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