Connect to an Alibaba Cloud Elasticsearch cluster

// The following sample code shows how to use BulkRequest for batch writing and SearchRequest.
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;

import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.*;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class RestClientTest710 {
    public static void main(String[] args) {
        // Alibaba Cloud Elasticsearch Serverless applications require basic authentication.
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        // Use the username and password that you configured when you created the Alibaba Cloud Elasticsearch serverless application. These are also the logon credentials for the Kibana console.
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("<UserName>", "<YourPassword>"));

        // Use a builder to create a REST client and configure the HttpClientConfigCallback of the HTTP client.
        // To obtain the application endpoint, click the ID of your Elasticsearch serverless application and go to the Basic Information page.
        RestClientBuilder builder = RestClient.builder(new HttpHost("<YourEsHost>", 9200, "http"))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    }
                });

        // A RestHighLevelClient instance is built using a REST low-level client builder.
        RestHighLevelClient highClient = new RestHighLevelClient(builder);

        try {
            // Batch write
            // Create a BulkRequest object and use the timeout() method to set the request timeout period to 10s.
            BulkRequest bulkRequest = new BulkRequest();
            bulkRequest.timeout("10s");

            // Create a request.
            Map<String, Object> jsonMap = new HashMap<>();
            // field_01 and field_02 are field names, and value_01 and value_02 are their values.
            jsonMap.put("{field_01}", "{value_01}");
            jsonMap.put("{field_02}", "{value_02}");
            IndexRequest indexRequest = new IndexRequest("{index_name}").id("{doc_id}").source(jsonMap);
            bulkRequest.add(indexRequest);

            Map<String, Object> jsonMap2 = new HashMap<>();
            jsonMap2.put("{field_01}", "{value_01}");
            jsonMap2.put("{field_02}", "{value_02}");
            IndexRequest indexRequest2 = new IndexRequest("{index_name}").id("{doc_id}").source(jsonMap2);
            bulkRequest.add(indexRequest2);

            BulkResponse bulkResponse = highClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            System.out.println(bulkResponse.hasFailures());


            // Query all documents.
            // Specify the index to search.
            SearchRequest searchRequest = new SearchRequest("{index_name}");
            // Build a query object.
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            // Create a query that matches all documents.
            searchSourceBuilder.query(QueryBuilders.matchAllQuery());
            // Set the query object as the source of SearchRequest.
            searchRequest.source(searchSourceBuilder);

            SearchResponse searchResponse = highClient.search(searchRequest, RequestOptions.DEFAULT);
            // Obtain the SearchHit[] searchHits array of the search results from SearchResponse.
            SearchHit[] searchHits = searchResponse.getHits().getHits();
            
            // Traverse the searchHits array to process each search hit. In this example, the getSourceAsString() method is used to print the source data of the document as a string.
            for (SearchHit hit : searchHits) {
                System.out.println(hit.getSourceAsString());
            }


            // Query a single document.
            GetRequest getRequest = new GetRequest("{index_name}", "{doc_id}");
            GetResponse getResponse = highClient.get(getRequest, RequestOptions.DEFAULT);
            // Print the content of the document.
            System.out.println(getResponse.getSourceAsString());
            // The full response is returned, just as it would be from a command-line request.
            System.out.println(getResponse);

            highClient.close();

        } catch (IOException ioException) {
            // Handle exceptions.
        }
    }
}

# coding="utf-8"
from elasticsearch import Elasticsearch
import time
import string
import random
def stru_doc(index_name,doc_type="_doc",count=100):
    doc = []
    for i in range(count):
        _id = ''.join([random.choice(letters) for _ in range(17)])

        i_doc = [{"create": {"_index": index_name, "_id": _id, "_type": doc_type}},
                 {"field1": "This is a test document for creating an index. It contains various text to simulate real-world data for search and analysis.",
                  "filed2": "By default, Cloud Assistant Agent is installed on ECS instances created from public images after December 1, 2017. If your instance was purchased before this date, you need to manually install the agent to use its features. The installation methods vary based on the operating system.",
                  "field3": "If you choose the default installation path, the client is installed in C:\\ProgramData\\aliyun\\assist\\ on a Windows instance.",
                  "filed4":"This topic provides a quick overview of the billing items, methods, components, and pricing for Elastic Compute Service (ECS).",
                  "filed5":"Note: The basic billing methods for ECS resources are subscription and pay-as-you-go. You can combine these with other cost-effective billing options to reduce costs. For more information, see Billing overview.",
                  "filed6":"The cat is out of the bag! We are proud to have been involved in the sale of the Ai.com domain. We look forward to seeing what they will do with it! On February 16, the AI chatbot ChatGPT became a global sensation. OpenAI, its developer, spent a fortune to redirect the premium domain AI.com to ChatGPT. As we all know, Microsoft is the major backer of OpenAI, and products like Bing are already testing ChatGPT. Microsoft also announced an expansion of its partnership with OpenAI, with the latter receiving a 'multi-year, multi-billion dollar' investment.",
                  "@timestamp": int(time.time())}]
        doc += i_doc
    return doc

def bulk_request():
    """
    Continuously write data without creating an index.
   
    :return:
    """
    index_name = "ali_push_doc_%s" % time.strftime('%H%M%S')

    body = {
        "settings": {"number_of_replicas": 0,
                     "number_of_shards": 1}
    }

    doc_type = "_doc"
    try:
        create_index = es.indices.create(index=index_name, body=body)
        print(create_index)
    except Exception as index:
        print(time.strftime('%H:%M:%s') + str(index))
        time.sleep(20)
    bulk_body = stru_doc(index_name)
    try:
        bulk_request = es.bulk(body=bulk_body, timeout='10s')
    # print(time.strftime('%H:%M:%S') + 'push_test' + str(bulk_request))
    except Exception as u:
        print(u)
        with open('push_doc_exception.txt', 'w') as f:
            f.write(time.strftime('%H:%M:%S') + url + '\n' + str(u) + '\n' + str(bulk_body) + '\n')
    es.indices.refresh()
    doc_account = es.count(index=index_name)
    print("indexcount:",doc_account)
    if doc_account["count"] == 100:
        for times in range(100):
            try:
                query_body = {
                    "query": {
                        "match_all": {}
                    }
                }
                search_request = es.search(query_body, index_name)
                print(search_request)
            except Exception as e:
                with open('asyn_error.txt', 'a') as f:
                    f.write(time.strftime('%H:%M:%S') + str(e) + '\n' + str(e) + '\n')
    else:
        with open('asyn_error.txt', 'a') as f:
            f.write(time.strftime('%H:%M:%S') + str(doc_account) + '\n')

if __name__ == '__main__':
    url = <YourEsHost>
    es = Elasticsearch(url, http_auth=(<UserName>, <YourPassword>))
    letters = string.ascii_lowercase + string.digits
    bulk_request()