Импорт большого XML файла в базу данных

Импорт большого XML файла в базу данных: Production решение

Архитектура решения

Для импорта 50 ГБ XML файла необходимо использовать потоковую обработку с chunk-based разбиением, чтобы избежать загрузки всего файла в память. Рекомендуемый подход использует Laravel Queue для асинхронной обработки и обеспечения отказоустойчивости.

Выбор библиотеки

XMLReader vs SimpleXML:

XMLReader (рекомендуется):

• Работает с файлами потоком, экономит память
• O(1) память независимо от размера файла
• Позволяет обрабатывать элементы по одному
• Идеален для больших файлов

SimpleXML:

• Загружает всю структуру в память
• O(n) память для файла размера n
• Удобен для маленьких файлов
• Непригоден для 50 ГБ

Решение 1: XMLReader со streaming обработкой

// app/Services/XmlImportService.php
namespace App\Services;

use App\Jobs\ProcessXmlChunk;
use App\Models\ImportLog;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;

class XmlImportService
{
    private const CHUNK_SIZE = 1000;
    private const CACHE_PREFIX = 'xml_import_';
    
    /**
     * Начинает импорт большого XML файла
     */
    public function import(string $filePath, string $importId): void
    {
        // Проверяем существование файла
        if (!file_exists($filePath)) {
            throw new \RuntimeException("XML file not found: {$filePath}");
        }
        
        // Инициализируем логирование
        $importLog = ImportLog::create([
            'id' => $importId,
            'file_path' => $filePath,
            'status' => 'processing',
            'total_records' => 0,
            'processed_records' => 0,
            'failed_records' => 0,
        ]);
        
        // Сохраняем состояние
        Cache::put(
            self::CACHE_PREFIX . $importId,
            [
                'processed' => 0,
                'failed' => 0,
                'last_position' => 0,
            ],
            60 * 60 // 1 час TTL
        );
        
        // Запускаем потоковую обработку
        $this->processXmlStream($filePath, $importId);
    }
    
    /**
     * Обрабатывает XML файл потоком
     */
    private function processXmlStream(string $filePath, string $importId): void
    {
        $reader = new \XMLReader();
        
        // Открываем файл для потоковой обработки
        if (!$reader->open($filePath)) {
            throw new \RuntimeException("Cannot open XML file: {$filePath}");
        }
        
        $chunk = [];
        $position = 0;
        $lastSavedPosition = $this->getLastSavedPosition($importId);
        
        // Ищем элементы для импорта
        while ($reader->read()) {
            // Пропускаем элементы до последней сохраненной позиции
            if ($position < $lastSavedPosition) {
                $position++;
                continue;
            }
            
            // Обрабатываем элементы типа "item" (адаптируй под свой XML)
            if ($reader->nodeType === \XMLReader::ELEMENT && $reader->name === 'item') {
                try {
                    // Разбираем элемент
                    $dom = new \DOMDocument();
                    $node = $reader->expand();
                    $domElement = $dom->importNode($node, true);
                    $dom->appendChild($domElement);
                    
                    // Преобразуем в массив
                    $data = $this->parseXmlElement($dom);
                    
                    // Валидируем
                    if ($this->validateData($data)) {
                        $chunk[] = $data;
                    } else {
                        $this->recordFailure($importId, $position, 'Validation failed');
                    }
                    
                } catch (\Exception $e) {
                    $this->recordFailure($importId, $position, $e->getMessage());
                }
                
                // Отправляем chunk в очередь при достижении размера
                if (count($chunk) >= self::CHUNK_SIZE) {
                    ProcessXmlChunk::dispatch($chunk, $importId, $position);
                    $chunk = [];
                }
            }
            
            $position++;
            
            // Логируем прогресс каждые 10000 элементов
            if ($position % 10000 === 0) {
                $this->logProgress($importId, $position);
            }
        }
        
        // Обрабатываем оставшиеся элементы
        if (!empty($chunk)) {
            ProcessXmlChunk::dispatch($chunk, $importId, $position);
        }
        
        $reader->close();
        
        // Обновляем статус
        ImportLog::find($importId)->update(['status' => 'completed']);
    }
    
    /**
     * Разбирает XML элемент в массив
     */
    private function parseXmlElement(\DOMDocument $dom): array
    {
        $xpath = new \DOMXPath($dom);
        
        return [
            'name' => $xpath->query('//name')[0]?->nodeValue,
            'email' => $xpath->query('//email')[0]?->nodeValue,
            'phone' => $xpath->query('//phone')[0]?->nodeValue,
            'address' => $xpath->query('//address')[0]?->nodeValue,
            'data' => $xpath->query('//data')[0]?->nodeValue,
        ];
    }
    
    /**
     * Валидирует данные перед импортом
     */
    private function validateData(array $data): bool
    {
        return !empty($data['name']) && !empty($data['email']);
    }
    
    /**
     * Записывает ошибку
     */
    private function recordFailure(string $importId, int $position, string $error): void
    {
        $state = Cache::get(self::CACHE_PREFIX . $importId);
        $state['failed']++;
        Cache::put(self::CACHE_PREFIX . $importId, $state);
        
        DB::table('import_errors')->insert([
            'import_id' => $importId,
            'position' => $position,
            'error' => $error,
            'created_at' => now(),
        ]);
    }
    
    /**
     * Логирует прогресс импорта
     */
    private function logProgress(string $importId, int $processed): void
    {
        ImportLog::find($importId)->update([
            'processed_records' => $processed,
        ]);
    }
    
    /**
     * Получает последнюю сохраненную позицию
     */
    private function getLastSavedPosition(string $importId): int
    {
        $state = Cache::get(self::CACHE_PREFIX . $importId);
        return $state['last_position'] ?? 0;
    }
}

Решение 2: Laravel Queue Job

// app/Jobs/ProcessXmlChunk.php
namespace App\Jobs;

use App\Models\ImportLog;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;

class ProcessXmlChunk implements ShouldQueue
{
    use Queueable, SerializesModels;
    
    private const CACHE_PREFIX = 'xml_import_';
    public $timeout = 300; // 5 минут на один chunk
    public $tries = 3; // 3 попытки
    
    public function __construct(
        private array $chunk,
        private string $importId,
        private int $position
    ) {}
    
    /**
     * Обрабатывает chunk данных
     */
    public function handle(): void
    {
        try {
            // Используем транзакцию для атомарности
            DB::transaction(function () {
                foreach ($this->chunk as $item) {
                    // Проверяем дубликаты
                    $exists = DB::table('users')
                        ->where('email', $item['email'])
                        ->exists();
                    
                    if (!$exists) {
                        DB::table('users')->insert([
                            'name' => $item['name'],
                            'email' => $item['email'],
                            'phone' => $item['phone'],
                            'address' => $item['address'],
                            'external_data' => json_encode($item['data'] ?? null),
                            'created_at' => now(),
                            'updated_at' => now(),
                        ]);
                    }
                }
            });
            
            // Обновляем прогресс
            $this->updateProgress(count($this->chunk));
            
        } catch (\Exception $e) {
            $this->fail($e);
        }
    }
    
    /**
     * Обновляет прогресс в кэше
     */
    private function updateProgress(int $count): void
    {
        $state = Cache::get(self::CACHE_PREFIX . $this->importId);
        $state['processed'] += $count;
        $state['last_position'] = $this->position;
        Cache::put(self::CACHE_PREFIX . $this->importId, $state);
        
        // Обновляем лог
        ImportLog::find($this->importId)->increment('processed_records', $count);
    }
    
    /**
     * Обработка ошибки job'а
     */
    public function failed(\Throwable $exception): void
    {
        ImportLog::find($this->importId)->update([
            'status' => 'failed',
            'error' => $exception->getMessage(),
        ]);
    }
}

Решение 3: Использование raw SQL для оптимизации

// Вместо INSERT по одному, используем COPY или BULK INSERT
class XmlBulkImporter
{
    /**
     * Импортирует chunk через CSV в PostgreSQL
     */
    public function importViaCSV(array $chunk, string $importId): void
    {
        // Создаем временный CSV файл
        $csvPath = storage_path("imports/{$importId}_chunk.csv");
        $handle = fopen($csvPath, 'w');
        
        foreach ($chunk as $item) {
            fputcsv($handle, [
                $item['name'],
                $item['email'],
                $item['phone'],
                $item['address'],
                json_encode($item['data'] ?? null),
                now()->toDateTimeString(),
            ]);
        }
        fclose($handle);
        
        // PostgreSQL COPY (в 10 раз быстрее, чем INSERT)
        DB::statement("
            COPY users (name, email, phone, address, external_data, created_at)
            FROM '" . $csvPath . "'
            WITH (FORMAT csv)
        ");
        
        // Удаляем временный файл
        unlink($csvPath);
    }
    
    /**
     * Импортирует через MySQL LOAD DATA
     */
    public function importViaLoadData(array $chunk, string $importId): void
    {
        $csvPath = storage_path("imports/{$importId}_chunk.csv");
        
        // Подготавливаем данные
        $fp = fopen($csvPath, 'w');
        foreach ($chunk as $item) {
            fputcsv($fp, [
                $item['name'],
                $item['email'],
                $item['phone'],
                $item['address'],
            ]);
        }
        fclose($fp);
        
        // MySQL LOAD DATA (в 20 раз быстрее обычного INSERT)
        DB::statement("
            LOAD DATA INFILE '" . $csvPath . "'
            INTO TABLE users
            FIELDS TERMINATED BY ','
            (name, email, phone, address)
        ");
        
        unlink($csvPath);
    }
}

Миграция БД для отслеживания

// database/migrations/create_import_logs_table.php
Schema::create('import_logs', function (Blueprint $table) {
    $table->uuid('id')->primary();
    $table->string('file_path');
    $table->enum('status', ['pending', 'processing', 'completed', 'failed']);
    $table->unsignedBigInteger('total_records')->default(0);
    $table->unsignedBigInteger('processed_records')->default(0);
    $table->unsignedBigInteger('failed_records')->default(0);
    $table->text('error')->nullable();
    $table->timestamp('started_at')->useCurrent();
    $table->timestamp('completed_at')->nullable();
    $table->timestamps();
    $table->index('status');
});

Schema::create('import_errors', function (Blueprint $table) {
    $table->uuid('id')->primary();
    $table->uuid('import_id');
    $table->unsignedBigInteger('position');
    $table->text('error');
    $table->timestamps();
    $table->foreign('import_id')->references('id')->on('import_logs');
    $table->index(['import_id', 'position']);
});

Использование

// Запуск импорта
$service = app(XmlImportService::class);
$importId = \Illuminate\Support\Str::uuid();
$service->import('/path/to/huge-50gb-file.xml', $importId);

// Мониторинг прогресса
$log = ImportLog::find($importId);
echo "Processed: {$log->processed_records} / {$log->total_records}";

// Обработка ошибок
$errors = DB::table('import_errors')
    ->where('import_id', $importId)
    ->get();

Оптимизация для 50 ГБ файла

Best practices:

• Используй XMLReader, не SimpleXML
• Обрабатывай по 1000-5000 элементов за раз
• Используй BULK INSERT (COPY/LOAD DATA) вместо обычного INSERT
• Отключи индексы перед импортом, включи после
• Используй асинхронные Job'ы для параллельной обработки
• Логируй позицию каждые 10000 элементов для восстановления после сбоя
• Используй подключение с пулингом (PgBouncer для PostgreSQL)

Заключение

Предложенное решение обеспечивает импорт 50 ГБ XML файла за счет потоковой обработки, асинхронных Job'ов и оптимизированных SQL операций, поддерживает восстановление после сбоев и логирование прогресса.