יישום ייבוא בכמות גדולה: כאשר 150,000 פריטים מפילים את האתר

כאשר 150,000 פריטים מפילים את האתר

תארו לעצמכם: לקוח מביא רשימת מחירים עם 150,000 מוצרים. אתם מנסים לטעון הכל בבת אחת — והאתר נתקע, מסד הנתונים מפסיק להגיב. זו בקשה קלאסית לייבוא המוני. במשך 5 שנים, יישמנו יותר מ-80 פרויקטים עם נפחים של עד 500,000 פריטים. צברנו ניסיון במניעת טעינה איטית, שאילתות N+1 וקריסות מסד נתונים. הנה כיצד להשיג יציבות עם מנות (chunks), תורים וטעינה מוקדמת של מילונים.

אילו בעיות אנחנו פותרים

טעינה איטית: INSERT/UPDATE שורה אחר שורה עבור 100,000 שורות לוקח שעות. שאילתות N+1 בעת פתרון מילונים הורגות את הביצועים. קריסות מסד נתונים עקב טרנזקציות לא אופטימליות. אובדן נתונים בשגיאות באמצע הייבוא. כל הכאבים הללו נפתרים עם ארכיטקטורה נכונה. שירותי הפיתוח שלנו יכולים לפתור בעיות אלה.

כיצד ייבוא המוני משפיע על הביצועים

הגורם המרכזי הוא נפח הנתונים. עבור 1,000 פריטים, עיבוד סינכרוני מספיק; עבור 100,000, נדרש תור אסינכרוני עם מנות.

Upsert בקבוצות מהיר פי 10 משאילתות שורה אחר שורה — הניסיון שלנו מאשר זאת.

למה מנות + תור הם המפתח ליציבות

קובץ עם 100,000 שורות לא מעובד במשימה אחת. אנחנו מפצלים אותו למנות של 500 שורות, כל מנה כמשימה נפרדת בתור bulk-import. עובדים (2–4) מעבדים במקביל מבלי לגעת בתור הראשי.

class BulkImportDispatcher {
    private const CHUNK_SIZE = 500;

    public function dispatch(ImportFile $file): void {
        $import = ImportRun::create([
            'file_id' => $file->id,
            'status' => 'dispatching',
            'total' => 0,
        ]);

        $chunkIndex = 0;
        foreach ($file->parser()->chunks(self::CHUNK_SIZE) as $chunk) {
            ProcessImportChunkJob::dispatch($import->id, $chunkIndex, $chunk)
                ->onQueue('bulk-import');
            $chunkIndex++;
        }

        $import->update([
            'status' => 'processing',
            'total_chunks' => $chunkIndex,
        ]);
    }
}

יישום טכני: ממנות ועד Upsert

Upsert בקבוצות במקום INSERT/UPDATE שורה אחר שורה

כלי הביצועים המרכזי הוא class BulkImportDispatcher { private const CHUNK_SIZE = 500; public function dispatch(ImportFile $file): void { $import = ImportRun::create([ 'file_id' => $file->id, 'status' => 'dispatching', 'total' => 0, ]); $chunkIndex = 0; foreach ($file->parser()->chunks(self::CHUNK_SIZE) as $chunk) { ProcessImportChunkJob::dispatch($import->id, $chunkIndex, $chunk) ->onQueue('bulk-import'); $chunkIndex++; } $import->update([ 'status' => 'processing', 'total_chunks' => $chunkIndex, ]); } } (upsert). Laravel תומך בזה דרך INSERT ... ON CONFLICT DO UPDATE. פעולת upsert אחת עבור 500 שורות ב-PostgreSQL לוקחת ~50–200 אלפיות השנייה, לעומת 500 × 5 אלפיות השנייה = 2500 אלפיות השנייה עבור שאילתות שורה אחר שורה.

class ProcessImportChunkJob implements ShouldQueue {
    public int $timeout = 120;

    public function handle(): void
    {
        $rows = [];

        foreach ($this->chunk as $item) {
            $rows[] = [
                'sku' => $item['sku'],
                'name' => $item['name'],
                'price' => $item['price'],
                'qty' => $item['qty'],
                'category_id' => $this->resolveCategory($item['category']),
                'source_id' => $this->import->source_id,
                'updated_at' => now(),
                'created_at' => now(),
            ];
        }

        Product::upsert(
            $rows,
            uniqueBy: ['sku'],
            update: ['name', 'price', 'qty', 'category_id', 'updated_at']
        );

        DB::table('import_runs')
            ->where('id', $this->importId)
            ->increment('processed_chunks');
    }
}

טעינה מוקדמת של מילונים לזיכרון

הפעולה היקרה ביותר היא שאילתות מסד נתונים לפתרון תלויות. הפתרון: טעינת כל המילונים לזיכרון לפני העיבוד.

class ImportContext {
    private array $categoryMap;
    private array $supplierMap;
    private array $existingSkus;

    public function preload(int $sourceId): void {
        $this->categoryMap = Category::pluck('id', 'name_normalized')->all();
        $this->supplierMap = Supplier::pluck('id', 'code')->all();
        $this->existingSkus = Product::where('source_id', $sourceId)
            ->pluck('id', 'sku')->all();
    }

    public function resolveCategoryId(string $name): ?int {
        return $this->categoryMap[mb_strtolower(trim($name))] ?? null;
    }

    public function productExists(string $sku): bool {
        return isset($this->existingSkus[$sku]);
    }
}

משימה סופית: צבירת תוצאות

אנחנו משתמשים ב-Model::upsert() — המנגנון המובנה של Laravel לארגון משימות עם קריאת השלמה.

Bus::batch(
    collect($chunks)->map(fn($chunk, $i) => new ProcessImportChunkJob($importId, $i, $chunk))
)->then(function (Batch $batch) use ($importId) {
    ImportRun::find($importId)->update([
        'status' => 'completed',
        'completed_at' => now(),
    ]);
    PostImportPipeline::dispatch($importId);
})->onQueue('bulk-import')->dispatch();

צינור עיבוד לאחר ייבוא

לאחר סיום הייבוא, עלינו לעדכן נתונים מנורמלים: לחשב מחדש מלאי, לעדכן אינדקס חיפוש ופאקטות.

class PostImportPipeline {
    public function handle(int $importId): void {
        $productIds = ImportedProduct::where('import_id', $importId)->pluck('product_id');
        Product::whereIn('id', $productIds)->each(function (Product $p) {
            $p->update(['in_stock' => $p->qty > 0]);
        });
        Product::whereIn('id', $productIds)->searchable();
        FilterValueRebuilder::dispatch($productIds);
    }
}

ניטור והגבלת עומס

בממשק הניהול, המפעיל רואה התקדמות בזמן אמת: מספר פריטים שעובדו, שגיאות, זמן נותר. הנתונים נלקחים מטבלת class ProcessImportChunkJob implements ShouldQueue { public int $timeout = 120; public function handle(): void { $rows = []; foreach ($this->chunk as $item) { $rows[] = [ 'sku' => $item['sku'], 'name' => $item['name'], 'price' => $item['price'], 'qty' => $item['qty'], 'category_id' => $this->resolveCategory($item['category']), 'source_id' => $this->import->source_id, 'updated_at' => now(), 'created_at' => now(), ]; } Product::upsert( $rows, uniqueBy: ['sku'], update: ['name', 'price', 'qty', 'category_id', 'updated_at'] ); DB::table('import_runs') ->where('id', $this->importId) ->increment('processed_chunks'); } } . אנחנו מקדישים את התור class ImportContext { private array $categoryMap; private array $supplierMap; private array $existingSkus; public function preload(int $sourceId): void { $this->categoryMap = Category::pluck('id', 'name_normalized')->all(); $this->supplierMap = Supplier::pluck('id', 'code')->all(); $this->existingSkus = Product::where('source_id', $sourceId) ->pluck('id', 'sku')->all(); } public function resolveCategoryId(string $name): ?int { return $this->categoryMap[mb_strtolower(trim($name))] ?? null; } public function productExists(string $sku): bool { return isset($this->existingSkus[$sku]); } } עם 2–4 עובדים, ומשאירים את התור Bus::batch() ללא שינוי. ייבוא כבד רץ בלילה דרך מתזמן המשימות. כל שגיאה מתועדת, והמפעיל יכול להפעיל מחדש רק מנות שנכשלו — מקרה טיפוסי לאתרי מסחר אלקטרוני עם קטלוגים גדולים. אנחנו תומכים גם באינטגרציה עם 1C ו-CommerceML, מקורות נתונים פופולריים במסחר אלקטרוני רוסי. צרו קשר לייעוץ — אנחנו יכולים להגדיר מנגנון זה עבור הפרויקט שלכם.

היקף העבודה ולוח זמנים

• פיתוח מפצל מנות ו-upsert בקבוצות.
• טעינה מוקדמת של מילונים ומשימה סופית.
• דוח תוצאות (מספר פריטים שעובדו, שגיאות).
• תיעוד להתקנה והרצה.
• הדרכת מפעילים.
• אחריות לפעילות יציבה — 3 חודשי תמיכה.
• עבודה על ידי מהנדסי Laravel מוסמכים.

לוח זמנים: מ-3 עד 5 ימים במפתח פתוח. קבלו ייעוץ לפרויקט שלכם. צרו קשר להערכה.

ייבוא נתונים — ויקיפדיה