//以下代碼示例為如何使用BulkRequest進(jìn)行批量寫入和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();
        //訪問用戶名和密碼為您創(chuàng)建阿里云Elasticsearch Serverless實例時設(shè)置的用戶名和密碼，也是Kibana控制臺的登錄用戶名和密碼。
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("<UserName>", "<YourPassword>"));

        // 通過builder創(chuàng)建rest client，配置http client的HttpClientConfigCallback。
        // 單擊所創(chuàng)建的Elasticsearch Serverless實例ID，在基本信息頁面獲取公網(wǎng)地址，即為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進(jìn)行構(gòu)造。
        RestHighLevelClient highClient = new RestHighLevelClient(builder);

        try {
            //批量寫入
            //創(chuàng)建BulkRequest 對象，并使用 timeout() 方法設(shè)置了請求的超時時間為 "10s"
            BulkRequest bulkRequest = new BulkRequest();
            bulkRequest.timeout("10s");

            // 創(chuàng)建request。
            Map<String, Object> jsonMap = new HashMap<>();
            // field_01、field_02為字段名，value_01、value_02為對應(yīng)的值。
            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}");
            //構(gòu)建了一個查詢對象
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            //創(chuàng)建一個匹配所有文檔的查詢
            searchSourceBuilder.query(QueryBuilders.matchAllQuery());
            //將查詢對象設(shè)置為 SearchRequest 的源
            searchRequest.source(searchSourceBuilder);

            SearchResponse searchResponse = highClient.search(searchRequest, RequestOptions.DEFAULT);
            //從 SearchResponse 中獲取搜索結(jié)果的數(shù)組 SearchHit[] searchHits。
            SearchHit[] searchHits = searchResponse.getHits().getHits();
            
            //遍歷 searchHits 數(shù)組，對每一個搜索命中的文檔進(jìn)行處理。在此示例中，使用 getSourceAsString() 方法將文檔的源數(shù)據(jù)以字符串形式打印出來。
            for (SearchHit hit : searchHits) {
                System.out.println(hit.getSourceAsString());
            }


            //查詢單個文檔
            GetRequest getRequest = new GetRequest("{index_name}", "{doc_id}");
            GetResponse getResponse = highClient.get(getRequest, RequestOptions.DEFAULT);
            //打印文檔的內(nèi)容
            System.out.println(getResponse.getSourceAsString());
            //返回的全部內(nèi)容和命令式一樣的
            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": "測試創(chuàng)建文檔dd汽車時代，如果你萬一開車撞了什么東西，千萬千萬要下車查看檢查，否則如果撞到的是人，就算是逃逸，只要構(gòu)成重傷，負(fù)主責(zé)以上，就構(gòu)成交通肇事罪哦。",
                  "filed2": "2017年12月01日之后使用公共鏡像創(chuàng)建的ECS實例，默認(rèn)預(yù)裝云助手客戶端。如果您的實例是2017年12月01日之前購買的，若需要使用云助手相關(guān)功能，請自行安裝云助手客戶端。"
                  "不同操作系統(tǒng)可選擇的安裝方式如下表所示。",
                  "field3": "如果您選擇默認(rèn)安裝路徑，客戶端在Windows實例中的安裝目錄為 C:\ProgramData\aliyun\assist\。",
                  "filed4":"通過閱讀本文，您可以快速了解云服務(wù)器ECS的計費(fèi)項及其計費(fèi)方式、計費(fèi)組成、定價等主要計費(fèi)信息。",
                  "filed5":"說明 ECS資源的基礎(chǔ)計費(fèi)方式為包年包月和按量付費(fèi)，針對不同的ECS資源，您可以根據(jù)需要結(jié)合其它優(yōu)惠的計費(fèi)方式來降低使用成本。更多信息請參見 計費(fèi)方式概述。",
                  "filed6":"貓不在袋子里了！我們很自豪地說我們參與了 Ai.com 域名的銷售。非常期待看到他們會用該域名做什么！IT之家 2 月 16 日消息，人工智能聊天機(jī)器人 ChatGPT 近日火爆全球，該技術(shù)的開發(fā)商 OpenAI 豪擲千金，將超優(yōu)質(zhì)域名 AI.com 鏈接跳轉(zhuǎn)到了 ChatGPT。眾所周知，OpenAI 背后的金主是科技巨頭微軟，目前必應(yīng)等產(chǎn)品已開始測試 ChatGPT，資金支持自然也不在話下。微軟此前還宣布，將擴(kuò)大與 OpenAI 的合作關(guān)系，后者將獲得微軟“多年、數(shù)十億美元”的投資，具體數(shù)額沒有披露，有媒體報道稱是 100 億美元。",
                  "@timestamp": int(time.time())}]
        doc += i_doc
    return doc

def bulk_request():
    """
    持續(xù)寫入 不創(chuàng)建索引
   
    :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()