Series Laravel & Elasticsearch Thực chiến #2: Nghệ thuật Mapping và Tuyệt kỹ nhồi dữ liệu (Bulk Index)

1. Cuốn từ điển của Quái vật (Mapping & Data Types)

Trong MySQL, trước khi insert dữ liệu, bạn phải tạo Bảng (Table) và định nghĩa kiểu dữ liệu (Varchar, Int, DateTime). Trong Elasticsearch, quá trình này gọi là tạo Index và định nghĩa Mapping.

Điều quan trọng nhất mọi Backend Dev phải khắc cốt ghi tâm khi làm việc với ES là sự khác biệt giữa hai kiểu dữ liệu chuỗi:

• keyword: ES sẽ lưu nguyên xi chuỗi đó thành một khối. Chỉ dùng cho các trường cần lọc chính xác (Exact Match), gom nhóm (Aggregation) hoặc sắp xếp. Ví dụ: card_id (Mã thẻ), station_id (Mã ga), status (Trạng thái).
• text: ES sẽ mang chuỗi đi "băm" (Analyze) thành các từ khóa nhỏ để phục vụ tìm kiếm toàn văn bản (Full-text search). Ví dụ: error_notes (Ghi chú lỗi của giao dịch). Tuyệt đối không dùng text để gom nhóm (GROUP BY).

2. Khởi tạo Index cho Giao dịch AFC

Chúng ta sẽ viết một Console Command để khởi tạo Index afc_transactions với cấu trúc Mapping chuẩn xác.

docker-compose exec app php artisan make:command ElasticInitIndex

Mở file app/Console/Commands/ElasticInitIndex.php và viết logic: Tạo command mới:

namespace App\Console\Commands;

use Elastic\Elasticsearch\Client;
use Illuminate\Console\Command;

class ElasticInitIndex extends Command
{
    protected $signature = 'elastic:init-index';
    protected $description = 'Khởi tạo Index afc_transactions với cấu trúc Mapping';

    public function handle(Client $client)
    {
        $indexName = 'afc_transactions';

        // Xóa Index cũ nếu tồn tại (cẩn thận trên Production nhé!)
        if ($client->indices()->exists(['index' => $indexName])->asBool()) {
            $client->indices()->delete(['index' => $indexName]);
            $this->warn("Đã xóa Index cũ: {$indexName}");
        }

        // Định nghĩa cấu trúc
        $params = [
            'index' => $indexName,
            'body' => [
                'mappings' => [
                    'properties' => [
                        'transaction_id' => ['type' => 'keyword'],
                        'card_id'        => ['type' => 'keyword'],
                        'station_id'     => ['type' => 'keyword'],
                        'amount'         => ['type' => 'integer'],
                        'status'         => ['type' => 'keyword'],
                        'tapped_at'      => ['type' => 'date', 'format' => 'yyyy-MM-dd HH:mm:ss'],
                        'error_notes'    => ['type' => 'text'] // Phục vụ Full-text search
                    ]
                ]
            ]
        ];

        // Gửi lệnh tạo Index
        $client->indices()->create($params);
        $this->info("✅ Đã khởi tạo thành công Index: {$indexName}");
    }
}

Chạy lệnh này qua Terminal, ES đã sẵn sàng một "cái kho" để đón dữ liệu.

3. Nỗi đau của vòng lặp foreach và Sự cứu rỗi của Bulk API

Giả sử bảng transactions trong MySQL của bạn đang có 1 triệu dòng. Nếu bạn query toàn bộ ra, rồi dùng vòng lặp foreach gọi API của ES để nạp từng dòng một... thì xin chúc mừng, server PHP của bạn sẽ bị tràn RAM (Out of Memory) ngay lập tức, và mất cả ngày trời cũng chưa nạp xong.

• chunk(1000): Lấy dữ liệu từ MySQL ra từng đợt, mỗi đợt 1000 dòng để không làm nghẽn RAM.
• Bulk API: Gói 1000 dòng đó thành một "cục" (payload) và ném sang ES trong đúng 1 lần gọi mạng (1 HTTP Request).

Tuyệt kỹ ở đây là kết hợp hàm chunk() của Laravel và Bulk API của Elasticsearch.

• chunk(1000): Lấy dữ liệu từ MySQL ra từng đợt, mỗi đợt 1000 dòng để không làm nghẽn RAM.
• Bulk API: Gói 1000 dòng đó thành một "cục" (payload) và ném sang ES trong đúng 1 lần gọi mạng (1 HTTP Request).

4. Thực chiến: Viết lệnh Đồng bộ dữ liệu (Sync)

Tiếp tục tạo một command mới để đồng bộ:

docker-compose exec app php artisan make:command ElasticSyncData

Mở file app/Console/Commands/ElasticSyncData.php:

namespace App\Console\Commands;

use Elastic\Elasticsearch\Client;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;

class ElasticSyncData extends Command
{
    protected $signature = 'elastic:sync';
    protected $description = 'Đồng bộ dữ liệu từ MySQL sang Elasticsearch bằng Bulk API';

    public function handle(Client $client)
    {
        $this->info('Bắt đầu quá trình đồng bộ dữ liệu...');
        
        $indexName = 'afc_transactions';
        $totalSynced = 0;

        // Chia nhỏ dữ liệu, mỗi lần lấy 1000 dòng để tối ưu RAM
        DB::table('transactions')->orderBy('id')->chunk(1000, function ($transactions) use ($client, $indexName, &$totalSynced) {
            $params = ['body' => []];

            foreach ($transactions as $txn) {
                // 1. Khai báo hành động (index) và ID của document
                $params['body'][] = [
                    'index' => [
                        '_index' => $indexName,
                        '_id'    => $txn->id // Trói buộc ID của ES giống hệt ID của MySQL
                    ]
                ];

                // 2. Nội dung dữ liệu đẩy lên
                $params['body'][] = [
                    'transaction_id' => $txn->transaction_code,
                    'card_id'        => $txn->card_id,
                    'station_id'     => $txn->station_id,
                    'amount'         => (int) $txn->amount,
                    'status'         => $txn->status,
                    'tapped_at'      => $txn->created_at,
                    'error_notes'    => $txn->error_message ?? ''
                ];
            }

            // Bắn một phát 1000 dòng sang ES
            $response = $client->bulk($params);

            if ($response['errors']) {
                $this->error('Có lỗi xảy ra trong quá trình Bulk Index!');
            } else {
                $totalSynced += count($transactions);
                $this->line("Đã đồng bộ được: {$totalSynced} dòng...");
            }
        });

        $this->info("✅ Đã hoàn tất đồng bộ tổng cộng {$totalSynced} dòng.");
    }
}

Bây giờ bạn chỉ cần gõ lệnh php artisan elastic:sync. Bạn sẽ thấy tiến trình nhảy số vèo vèo, 1 triệu dòng dữ liệu có thể được xử lý gọn gàng chỉ trong vòng vài chục giây mà server vẫn thở đều.

Tổng kết

Trong bài này, bạn đã làm chủ được nghệ thuật khai báo cấu trúc dữ liệu (keyword vs text) và phương pháp tối ưu hóa bộ nhớ kết hợp băng thông mạng (Chunk + Bulk Index).

Dữ liệu giờ đây đã nằm gọn gàng bên trong bụng của Quái vật Elasticsearch. Ở Bài 3, chúng ta sẽ bắt đầu "vắt sức" nó bằng Elasticsearch Query DSL.