如何通过Java High Level Rest Client批量API实现ES索引创建与多库SQL数据导入

Hey there! Let's break down your Elasticsearch integration challenges and fix up your code. I've noticed a few critical issues in your implementation, plus I'll clarify exactly how to use CreateIndexResponse effectively.

核心问题梳理 #

First, let's highlight the main problems in your current code:

• You're using UpdateRequest to add new documents (this is for updating existing docs—use IndexRequest instead)
• Your index settings include cluster-level parameters (like cluster.name, http.enabled) that don't belong in index-specific config
• The counter variable is positioned incorrectly, leading to duplicate document IDs
• You're not leveraging CreateIndexResponse to verify if your index was created successfully
• JDBC resources (Statement, ResultSet) aren't being managed properly to avoid leaks

修正后的完整代码 #

Here's the revised code with all fixes and best practices applied:

public static void importDataToElasticsearch(Connection con) throws Exception {
    // Use try-with-resources to auto-close JDBC resources
    try (Statement statement = con.createStatement();
         ResultSet result = statement.executeQuery("SELECT Field_1, Field_2, Field_3 from Table_1")) {

        // 1. Create Index with correct settings
        CreateIndexRequest createIndexRequest = new CreateIndexRequest("index_name");
        // Only include index-level settings (cluster settings are managed on ES nodes, not here)
        createIndexRequest.settings(Settings.builder()
                .put("index.number_of_shards", 3)
                .put("index.number_of_replicas", 1)
                .build());

        // Execute index creation and handle response
        CreateIndexResponse createIndexResponse = ElasticSearch.eclient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
        
        // Verify index creation result
        if (createIndexResponse.isAcknowledged()) {
            System.out.println("Index created successfully!");
        } else {
            System.err.println("Index creation request was not acknowledged by the cluster.");
            return; // Exit if index creation failed
        }

        // 2. Prepare bulk request for data import
        BulkRequest bulkRequest = new BulkRequest();
        int counter = 1; // Initialize counter inside try block, increment in loop

        while (result.next()) {
            String field_1 = result.getString("Field_1");
            int field_2 = result.getInt("Field_2");
            String field_3 = result.getString("Field_3");

            // Build JSON document
            XContentBuilder builder = XContentFactory.jsonBuilder()
                    .startObject()
                    .field("Field 1", field_1)
                    .field("Field 2", field_2)
                    .field("Field 3", field_3)
                    .endObject();

            // Use IndexRequest for NEW documents (UpdateRequest is for existing docs)
            IndexRequest indexRequest = new IndexRequest("index_name")
                    .id(Integer.toString(counter))
                    .source(builder);
            
            bulkRequest.add(indexRequest);
            counter++; // Increment counter for each new document
        }

        // Execute bulk request
        if (!bulkRequest.requests().isEmpty()) {
            BulkResponse response = ElasticSearch.eclient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (response.hasFailures()) {
                for (BulkItemResponse item : response.getItems()) {
                    System.err.println("Bulk item failure: " + item.getFailureMessage());
                }
            } else {
                System.out.println("Successfully imported " + (counter-1) + " documents to Elasticsearch!");
            }
        } else {
            System.out.println("No data to import from the query result.");
        }

    } catch (Exception ex) {
        ex.printStackTrace();
        throw ex; // Re-throw to let caller handle if needed
    }
}

关键知识点详解 #

Let's dive into the key improvements and clarifications:

1. How to use CreateIndexResponse #

This object gives you critical feedback about your index creation request:

• isAcknowledged(): Returns true if the cluster acknowledged the index creation request
• isShardsAcknowledged(): Returns true if all required shards were allocated successfully
• index(): Returns the name of the index that was created

You should always check these values to ensure your index was set up correctly before proceeding to import data.

2. Correct Index Settings Configuration #

Your original code included cluster-level settings (like cluster.name, http.enabled, node.data) in the index request—these are not index-specific. Index settings should only include parameters that apply to the index itself, such as:

• index.number_of_shards: Number of primary shards (can't change after index creation)
• index.number_of_replicas: Number of replica shards (can change dynamically later)
• index.refresh_interval: How often the index refreshes to make new docs visible
• index.codec: Compression codec for stored data

Cluster settings are configured in your Elasticsearch node's elasticsearch.yml file, not via the index creation API.

3. Bulk Import Best Practices #

• Use IndexRequest for adding new documents: UpdateRequest is only for modifying existing documents (it will throw an error if the document doesn't exist unless you set upsert()).
• Always check if the bulk request is empty before executing it to avoid unnecessary API calls.
• Handle bulk failures gracefully by iterating over BulkItemResponse to debug individual failures.

4. Resource Management #

Using try-with-resources (try (Statement s = con.createStatement(); ...)) ensures that JDBC resources (Statement, ResultSet) are automatically closed, even if an exception occurs. This prevents memory leaks and resource exhaustion.

5. Extending to Multiple Databases #

To import data from multiple databases, you can refactor the method to accept a Connection and a String sqlQuery as parameters. For example:

public static void importFromDatabase(Connection con, String sqlQuery, String indexName) throws Exception {
    // Reuse the same logic as above, replacing the hardcoded query and index name
}

Then call this method with different database connections and queries to populate the same Elasticsearch index.

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