יישום עיבוד קבצים בקבוצות על השרת

אנו נתקלים לעתים קרובות במצבים שבהם לקוח צריך לייבא 50,000 שורות מקובץ CSV. הטמעת עיבוד הקבצים האצוותי שלנו מנצלת עבודות תור של Laravel לעיבוד נתונים מקבילי, ומבטיחה ייבוא יציב של קבצים גדולים ללא דליפות זיכרון. ללא אסטרטגיית עיבוד אצוותי ברורה, משימות כאלה מובילות ל-OOM וזמני ожидание אינסופיים. הניסיון שלנו מראה: ארכיטקטורה נכונה מקצרת את זמן העיבוד פי 10 ומונעת אובדן נתונים. זה לא משנה אם אתם עובדים עם עשרות או מאות אלפי רשומות — הדפוסים נשארים זהים.

בעיות מרכזיות בעיבוד אצוותי

זיכרון. טעינת כל קובץ ה-CSV למערך היא דרך בטוחה למיצוי זיכרון. הדפוס הנכון הוא קריאת זרם בנתחים. אנו משתמשים ב-LazyCollection של Laravel, אשר קורא את הקובץ שורה אחר שורה ללא טעינה לזיכרון. גישה זו גם מייעלת את איגוד חיבורי מסד הנתונים ואת תפוקת התור.

שגיאות חלקיות. אם מתוך 10,000 שורות 50 אינן תקינות — עצירת התהליך כולו היא שגויה. ההיגיון שלנו: דלג על שורות בעייתיות, רשום ביומן עם הקשר (למשל, האימייל של הרשומה השגויה), והמשך.

יכולת חידוש. אם התהליך נכשל בשורה 7,000 — אנו לא מתחילים מההתחלה. Laravel Batch מאפשר חידוש מנקודת הכשל, תוך שמירה על נתחים שכבר עובדו. אטומיות האצווה נשמרת באמצעות מעקב מצב בטבלת LazyCollection.

מקביליות. עיבוד רציף של 50,000 רשומות ב-100ms כל אחת לוקח כמעט 1.5 שעות. פיצול לעבודות מקבילות (גודל נתח אופטימלי של 500 רשומות) מקצר זאת ל-5–10 דקות על 4 עובדים. מקביליות העובדים מוגדרת לעומס אופטימלי באמצעות ניהול תהליכי עובדי תור (Supervisor עם numprocs=4).

כיצד להימנע מדליפות זיכרון בעת ייבוא CSV גדול?

אנו מיישמים את הדפוס "Batch → Chunks → Jobs". לאחר העלאת הקובץ, משימת אב מפצלת את הנתונים לנתחים; כל נתח מעובד על ידי עבודה נפרדת במקביל. לאחר שכל העבודות מסתיימות, משימת צבירה רצה.

namespace App\Services;

use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;
use Illuminate\Support\LazyCollection;

class CsvImportService
{
    private const CHUNK_SIZE = 500;

    public function startImport(string $filePath, int $importId): string
    {
        $jobs = [];

        LazyCollection::make(function () use ($filePath) {
            $handle = fopen($filePath, 'r');
            $header = fgetcsv($handle);
            while (($row = fgetcsv($handle)) !== false) {
                yield array_combine($header, $row);
            }
            fclose($handle);
        })
        ->chunk(self::CHUNK_SIZE)
        ->each(function ($chunk, $index) use (&$jobs, $importId) {
            $jobs[] = new ProcessCsvChunkJob(
                importId: $importId,
                chunkIndex: $index,
                rows: $chunk->values()->toArray()
            );
        });

        $batch = Bus::batch($jobs)
            ->name("csv-import-{$importId}")
            ->allowFailures()
            ->then(function (Batch $batch) use ($importId) {
                Import::find($importId)?->update(['status' => 'completed']);
                ImportCompletedEvent::dispatch($importId);
            })
            ->catch(function (Batch $batch, \Throwable $e) use ($importId) {
                Import::find($importId)?->update([
                    'status' => 'partially_failed',
                    'error_message' => $e->getMessage(),
                ]);
            })
            ->finally(function (Batch $batch) use ($importId) {
                $import = Import::find($importId);
                $import?->update([
                    'total_jobs' => $batch->totalJobs,
                    'failed_jobs' => $batch->failedJobs,
                    'finished_at' => now(),
                ]);
            })
            ->onQueue('batch-processing')
            ->dispatch();

        Import::find($importId)?->update(['batch_id' => $batch->id]);

        return $batch->id;
    }
}

עבודת עיבוד נתח

class ProcessCsvChunkJob implements ShouldQueue
{
    use Batchable, Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public int $tries = 3;
    public int $timeout = 120;
    public int $backoff = 10;

    public function __construct(
        private int $importId,
        private int $chunkIndex,
        private array $rows
    ) {}

    public function handle(): void
    {
        if ($this->batch()?->cancelled()) {
            return;
        }

        $successCount = 0;
        $errors = [];

        foreach ($this->rows as $lineNum => $row) {
            try {
                $this->processRow($row);
                $successCount++;
            } catch (\Throwable $e) {
                $errors[] = [
                    'chunk' => $this->chunkIndex,
                    'line' => $lineNum,
                    'data' => array_slice($row, 0, 3),
                    'error' => $e->getMessage(),
                ];
            }
        }

        ImportChunkResult::create([
            'import_id' => $this->importId,
            'chunk_index' => $this->chunkIndex,
            'processed' => count($this->rows),
            'succeeded' => $successCount,
            'failed' => count($errors),
            'errors' => $errors,
        ]);

        Import::where('id', $this->importId)->increment('processed_rows', count($this->rows));
        Import::where('id', $this->importId)->increment('success_rows', $successCount);
    }

    private function processRow(array $row): void
    {
        $validated = validator($row, [
            'email' => 'required|email',
            'name' => 'required|string|max:255',
        ])->validate();

        User::updateOrCreate(
            ['email' => $validated['email']],
            ['name' => $validated['name']]
        );
    }
}

מה אם התהליך נקטע?

Laravel Batch שומר מצב בטבלת class ProcessCsvChunkJob implements ShouldQueue { use Batchable, Dispatchable, InteractsWithQueue, Queueable, SerializesModels; public int $tries = 3; public int $timeout = 120; public int $backoff = 10; public function __construct( private int $importId, private int $chunkIndex, private array $rows ) {} public function handle(): void { if ($this->batch()?->cancelled()) { return; } $successCount = 0; $errors = []; foreach ($this->rows as $lineNum => $row) { try { $this->processRow($row); $successCount++; } catch (\Throwable $e) { $errors[] = [ 'chunk' => $this->chunkIndex, 'line' => $lineNum, 'data' => array_slice($row, 0, 3), 'error' => $e->getMessage(), ]; } } ImportChunkResult::create([ 'import_id' => $this->importId, 'chunk_index' => $this->chunkIndex, 'processed' => count($this->rows), 'succeeded' => $successCount, 'failed' => count($errors), 'errors' => $errors, ]); Import::where('id', $this->importId)->increment('processed_rows', count($this->rows)); Import::where('id', $this->importId)->increment('success_rows', $successCount); } private function processRow(array $row): void { $validated = validator($row, [ 'email' => 'required|email', 'name' => 'required|string|max:255', ])->validate(); User::updateOrCreate( ['email' => $validated['email']], ['name' => $validated['name']] ); } } . נתחים שהושלמו מסומנים כמושלמים; לא שלמים מתחדשים אוטומטית בעת הפעלה מחדש של העובד. להפעלה מחדש כפויה, ניתן לשאול אינדקסים של נתחים שלא הושלמו מ-job_batches ולשלוח מחדש עבודות.

התקדמות בזמן אמת נחשפת דרך נקודת קצה:

public function progress(int $importId): JsonResponse
{
    $import = Import::findOrFail($importId);
    $batch = $import->batch_id ? Bus::findBatch($import->batch_id) : null;

    return response()->json([
        'status' => $import->status,
        'processed_rows' => $import->processed_rows,
        'success_rows' => $import->success_rows,
        'total_rows' => $import->total_rows,
        'percentage' => $import->total_rows > 0 ? round($import->processed_rows / $import->total_rows * 100, 1) : 0,
        'batch' => $batch ? [
            'total_jobs' => $batch->totalJobs,
            'pending_jobs' => $batch->pendingJobs,
            'failed_jobs' => $batch->failedJobs,
            'progress' => $batch->progress(),
        ] : null,
    ]);
}

ויסות עומס

עבור תור האצווה, מאגר עובדים ייעודי עם מקביליות מוגבלת מונע הצפה של מסד הנתונים או המעבד:

[program:batch-worker]
command=php artisan queue:work --queue=batch-processing --max-jobs=50 --sleep=3 --timeout=120
numprocs=4
autostart=true
autorestart=true

ImportChunkResult — ארבעה עובדים, כל אחד מעבד נתחים ברצף. public function progress(int $importId): JsonResponse { $import = Import::findOrFail($importId); $batch = $import->batch_id ? Bus::findBatch($import->batch_id) : null; return response()->json([ 'status' => $import->status, 'processed_rows' => $import->processed_rows, 'success_rows' => $import->success_rows, 'total_rows' => $import->total_rows, 'percentage' => $import->total_rows > 0 ? round($import->processed_rows / $import->total_rows * 100, 1) : 0, 'batch' => $batch ? [ 'total_jobs' => $batch->totalJobs, 'pending_jobs' => $batch->pendingJobs, 'failed_jobs' => $batch->failedJobs, 'progress' => $batch->progress(), ] : null, ]); } — לאחר 50 עבודות, העובד מופעל מחדש כדי לפנות זיכרון.

אנו בוחרים ב-500 כאיזון אופטימלי בין מהירות ליציבות. היישום האצוותי שלנו מעבד 50,000 רשומות פי 10 מהר יותר מעיבוד רציף.

תהליך העבודה

1. ניתוח: לימוד פורמט הקובץ, נפח הנתונים, דרישות המהירות.
2. עיצוב: בחירת גודל נתח, הגדרת תורים.
3. יישום: כתיבת קוד עם נתחים, טיפול בשגיאות, התקדמות ויכולת חידוש.
4. בדיקות: הרצה על נתוני בדיקה עם כשלים מדומים.
5. פריסה: הגדרת עובדים, ניטור, מסירת תיעוד.

מה כלול ביישום סוהר

• ארכיטקטורת עיבוד אצוותי עם נתחים ועבודות מקבילות.
• נקודת קצה להתקדמות ושחזור.
• רישום שגיאות מפורט לניתוח.
• תיעוד פריסה ותפעול.
• הדרכת צוות על שימוש במערכת.
• תמיכה חודש לאחר ההשקה.

לוח זמנים ועלות

יישום ייבוא CSV בסיסי עם נתחים והתקדמות — מיום עסקים אחד. הוספת יכולת חידוש, יומנים מפורטים ותמיכה ב-XLSX/JSON — עוד 1–2 ימים. עבור ייבוא CSV סטנדרטי עם נתחים והתקדמות, העלות הסוהר היא $1,500. החיסכון האופייני לעסק בינוני עולה על $20,000 בשנה עקב הפחתת עיבוד ידני. עם 5+ שנים בשוק ולמעלה מ-50 פרויקטי עיבוד אצוותי שהושלמו, אנו מבטיחים יציבות וסקלביליות.

יתרונות של Bus::batch()

תיעוד Laravel Bus::batch() מספק תמיכה מובנית למשימות מקובצות: מעקב מצב, כשלים חלקיים, שרשראות קריאה חוזרת. זה מבטל את הצורך בכתיבת מתזמן משלך ומפחית את סיכון השגיאות.

ייבוא CSV בצד השרת ועיבוד קבצים גדולים ב-PHP

המומחיות שלנו כוללת ייבוא CSV בצד השרת, עיבוד נתונים מקבילי, וקריאת זרם LazyCollection לעיבוד קבצים גדולים ב-PHP. אנו משלבים ניטור התקדמות ייבוא וחידוש אצווה לאחר כשל באמצעות נתחי Bus::batch. היגיון טיפול השגיאות האצוותי שלנו מבטיח שלמות נתונים.

קבלו ייעוץ — כתבו לנו, ונכין ארכיטקטורה לתרחיש שלכם תוך יום. הזמינו ביקורת על תהליך האצווה שלכם — נמצא צווארי בקבוק ונציע אופטימיזציה.