Couchbase基于Java的批量插入最优方案及示例代码咨询

Hey there! Let's walk through the best way to handle bulk inserts of large JSON and CSV datasets into Couchbase for your Java project. I've tackled similar bulk loading tasks before, so here's a practical, optimized approach with example code to get you started.

First, let's cover the core principles that'll keep your bulk load efficient and reliable:

• Stick to Couchbase Java SDK 3.x or newer—it's built with reactive/async support that's way better for large-scale operations than older versions.
• Use batch processing (100-500 documents per batch works for most cases; tweak based on your document size).
• Leverage the SDK's reactive APIs to avoid blocking threads and maximize throughput.
• For CSV data, first parse rows into Couchbase-compatible JsonObjects (use a tried-and-true library like Apache Commons CSV or OpenCSV).
• Handle transient errors with the SDK's built-in retry mechanisms (no need to reinvent the wheel here).

Example 1: Bulk Insert JSON Files #

This example reads multiple JSON files, parses them into JsonObjects, and inserts them in batches using the reactive SDK for non-blocking performance.

import com.couchbase.client.java.*;
import com.couchbase.client.java.json.JsonObject;
import com.couchbase.client.java.kv.UpsertOptions;
import reactor.core.publisher.Flux;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class JsonBulkLoader {
    public static void main(String[] args) {
        // Initialize your Couchbase connection (update with your cluster details)
        Cluster cluster = Cluster.connect("your-cluster-host", "username", "password");
        Bucket bucket = cluster.bucket("target-bucket");
        Collection collection = bucket.defaultCollection();

        try {
            // Replace with your list of JSON file paths
            List<String> jsonFiles = List.of(
                "/data/json/user-data-1.json",
                "/data/json/user-data-2.json",
                "/data/json/user-data-3.json"
            );

            Flux.fromIterable(jsonFiles)
                .flatMap(filePath -> {
                    try {
                        // Read and parse the JSON file into a JsonObject
                        String jsonContent = Files.readString(Paths.get(filePath));
                        JsonObject doc = JsonObject.fromJson(jsonContent);
                        
                        // Assume your document ID is stored in the "documentId" field—adjust as needed
                        String docId = doc.getString("documentId");
                        
                        // Return the reactive upsert operation
                        return collection.reactive().upsert(docId, doc, UpsertOptions.upsertOptions()
                            .timeout(15, TimeUnit.SECONDS));
                    } catch (Exception e) {
                        return Flux.error(new RuntimeException("Failed to process file: " + filePath, e));
                    }
                })
                .buffer(250) // Batch size: adjust based on your document size and cluster capacity
                .doOnNext(batchResults -> {
                    System.out.printf("Successfully inserted %d documents in this batch%n", batchResults.size());
                })
                .doOnError(error -> {
                    System.err.printf("Bulk insert hit an error: %s%n", error.getMessage());
                })
                .blockLast(); // Wait for all operations to finish (remove if using non-blocking context)

        } finally {
            // Always clean up the cluster connection
            cluster.disconnect();
        }
    }
}

Example 2: Bulk Insert CSV Data #

For CSV, we'll first parse each row into a JsonObject, then use the same reactive bulk pattern. We'll use Apache Commons CSV for parsing—it's lightweight and easy to work with.

First, add the dependency to your pom.xml (if using Maven):

<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-csv</artifactId>
    <version>1.10.0</version>
</dependency>

Then the loader code:

import com.couchbase.client.java.*;
import com.couchbase.client.java.json.JsonObject;
import com.couchbase.client.java.kv.UpsertOptions;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import reactor.core.publisher.Flux;
import java.io.FileReader;
import java.util.concurrent.TimeUnit;

public class CsvBulkLoader {
    public static void main(String[] args) {
        Cluster cluster = Cluster.connect("your-cluster-host", "username", "password");
        Bucket bucket = cluster.bucket("target-bucket");
        Collection collection = bucket.defaultCollection();

        String csvPath = "/data/csv/user-records.csv";

        // Use try-with-resources to auto-close readers/parsers
        try (FileReader reader = new FileReader(csvPath);
             CSVParser csvParser = new CSVParser(reader, CSVFormat.DEFAULT.withHeader())) {

            Flux.fromIterable(csvParser)
                .map(csvRecord -> {
                    // Map CSV columns to JsonObject fields—adjust to match your CSV schema
                    JsonObject doc = JsonObject.create()
                        .put("id", csvRecord.get("user_id")) // Use this as the document ID
                        .put("full_name", csvRecord.get("full_name"))
                        .put("email", csvRecord.get("email"))
                        .put("signup_date", csvRecord.get("signup_date"))
                        .put("is_active", Boolean.parseBoolean(csvRecord.get("is_active")));

                    return new DocPair(csvRecord.get("user_id"), doc);
                })
                .flatMap(pair -> collection.reactive().upsert(pair.id, pair.doc, UpsertOptions.upsertOptions()
                    .timeout(15, TimeUnit.SECONDS)))
                .buffer(300) // Adjust batch size based on your needs
                .doOnNext(batch -> System.out.printf("Processed %d CSV records%n", batch.size()))
                .doOnError(err -> System.err.printf("CSV bulk load failed: %s%n", err.getMessage()))
                .blockLast();

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            cluster.disconnect();
        }
    }

    // Simple helper class to pair document ID with its JsonObject
    private static class DocPair {
        String id;
        JsonObject doc;

        DocPair(String id, JsonObject doc) {
            this.id = id;
            this.doc = doc;
        }
    }
}

Pro Tips for Success #

• Tweak batch size: If you're dealing with large documents (1MB+), drop the batch size to 100 or less. For tiny documents, you can go up to 1000 per batch.
• Avoid blocking in production: If this is part of a web app or non-blocking service, replace blockLast() with subscribe() and handle success/error callbacks appropriately.
• Log effectively: Enable SLF4J logging for the Couchbase SDK to track slow operations or errors during bulk loads.
• Duplicate ID handling: The examples use upsert which overwrites existing documents. If you want to fail on duplicates, use insert instead (but be prepared to handle conflict errors).
• Cluster prep: Temporarily scale up your Couchbase cluster's KV workers or memory if you're loading a massive dataset—it'll speed things up and avoid timeouts.

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