通过客户端访问阿里云Elasticsearch

//以下代码示例为如何使用BulkRequest进行批量写入和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) {
        // 阿里云Elasticsearch Serverless集群需要basic auth验证。
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        //访问用户名和密码为您创建阿里云Elasticsearch Serverless实例时设置的用户名和密码，也是Kibana控制台的登录用户名和密码。
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("<UserName>", "<YourPassword>"));

        // 通过builder创建rest client，配置http client的HttpClientConfigCallback。
        // 单击所创建的Elasticsearch Serverless实例ID，在基本信息页面获取公网地址，即为ES集群地址。
        RestClientBuilder builder = RestClient.builder(new HttpHost("<YourEsHost>", 9200, "http"))
                .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                    @Override
                    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    }
                });

        // RestHighLevelClient实例通过REST low-level client builder进行构造。
        RestHighLevelClient highClient = new RestHighLevelClient(builder);

        try {
            //批量写入
            //创建BulkRequest 对象，并使用 timeout() 方法设置了请求的超时时间为 "10s"
            BulkRequest bulkRequest = new BulkRequest();
            bulkRequest.timeout("10s");

            // 创建request。
            Map<String, Object> jsonMap = new HashMap<>();
            // field_01、field_02为字段名，value_01、value_02为对应的值。
            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());


            //查询所有文档
            //指定搜索的索引
            SearchRequest searchRequest = new SearchRequest("{index_name}");
            //构建了一个查询对象
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            //创建一个匹配所有文档的查询
            searchSourceBuilder.query(QueryBuilders.matchAllQuery());
            //将查询对象设置为 SearchRequest 的源
            searchRequest.source(searchSourceBuilder);

            SearchResponse searchResponse = highClient.search(searchRequest, RequestOptions.DEFAULT);
            //从 SearchResponse 中获取搜索结果的数组 SearchHit[] searchHits。
            SearchHit[] searchHits = searchResponse.getHits().getHits();
            
            //遍历 searchHits 数组，对每一个搜索命中的文档进行处理。在此示例中，使用 getSourceAsString() 方法将文档的源数据以字符串形式打印出来。
            for (SearchHit hit : searchHits) {
                System.out.println(hit.getSourceAsString());
            }


            //查询单个文档
            GetRequest getRequest = new GetRequest("{index_name}", "{doc_id}");
            GetResponse getResponse = highClient.get(getRequest, RequestOptions.DEFAULT);
            //打印文档的内容
            System.out.println(getResponse.getSourceAsString());
            //返回的全部内容和命令式一样的
            System.out.println(getResponse);

            highClient.close();

        } catch (IOException ioException) {
            // 异常处理。
        }
    }
}

# 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": "测试创建文档dd汽车时代，如果你万一开车撞了什么东西，千万千万要下车查看检查，否则如果撞到的是人，就算是逃逸，只要构成重伤，负主责以上，就构成交通肇事罪哦。",
                  "filed2": "2017年12月01日之后使用公共镜像创建的ECS实例，默认预装云助手客户端。如果您的实例是2017年12月01日之前购买的，若需要使用云助手相关功能，请自行安装云助手客户端。"
                  "不同操作系统可选择的安装方式如下表所示。",
                  "field3": "如果您选择默认安装路径，客户端在Windows实例中的安装目录为 C:\ProgramData\aliyun\assist\。",
                  "filed4":"通过阅读本文，您可以快速了解云服务器ECS的计费项及其计费方式、计费组成、定价等主要计费信息。",
                  "filed5":"说明 ECS资源的基础计费方式为包年包月和按量付费，针对不同的ECS资源，您可以根据需要结合其它优惠的计费方式来降低使用成本。更多信息请参见 计费方式概述。",
                  "filed6":"猫不在袋子里了！我们很自豪地说我们参与了 Ai.com 域名的销售。非常期待看到他们会用该域名做什么！IT之家 2 月 16 日消息，人工智能聊天机器人 ChatGPT 近日火爆全球，该技术的开发商 OpenAI 豪掷千金，将超优质域名 AI.com 链接跳转到了 ChatGPT。众所周知，OpenAI 背后的金主是科技巨头微软，目前必应等产品已开始测试 ChatGPT，资金支持自然也不在话下。微软此前还宣布，将扩大与 OpenAI 的合作关系，后者将获得微软“多年、数十亿美元”的投资，具体数额没有披露，有媒体报道称是 100 亿美元。",
                  "@timestamp": int(time.time())}]
        doc += i_doc
    return doc

def bulk_request():
    """
    持续写入 不创建索引
   
    :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()