devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

JAVA elasticsearch批量插入_elasticsearch 插入

作者：Monodyee | 2024-02-08 09:54:01

赞

踩

elasticsearch 插入

耐心看，多看代码中写的注释，keyid是文档中的_id

目录

耐心看，多看代码中写的注释，keyid是文档中的_id

1.导入包

2.插入格式

3.插入类

--------附录（新增，删除，更新，插入等）

1.导入包

  1. <properties>
  2.    <java.version>1.8</java.version>
  3.    <elasticsearch.version>7.9.3</elasticsearch.version>
  4. </properties>
  5. <!-- elasticsearch相关 -->
  6. <dependency>
  7.    <groupId>org.elasticsearch.client</groupId>
  8.    <artifactId>elasticsearch-rest-high-level-client</artifactId>
  9.    <version>${elasticsearch.version}</version>
  10. </dependency>
  11. <dependency>
  12.    <groupId>org.elasticsearch</groupId>
  13.    <artifactId>elasticsearch</artifactId>
  14.    <version>${elasticsearch.version}</version>
  15. </dependency>
  16. <dependency>
  17.    <groupId>org.elasticsearch.client</groupId>
  18.    <artifactId>elasticsearch-rest-client-sniffer</artifactId>
  19.    <version>${elasticsearch.version}</version>
  20. </dependency>

2.插入格式

将对象转换为map

  1. public static <T> Map<String, Object> beanToMap(T bean) {
  2.     Map<String, Object> map = Maps.newHashMap();
  3.     if (bean != null) {
  4.         BeanMap beanMap = BeanMap.create(bean);
  5.         for (Object key : beanMap.keySet()) {
  6.             map.put(key.toString(), beanMap.get(key));
  7.         }
  8.     }
  9.     return map;
  10. }

创建map

  1. try {
  2. //下面这行是你要插入es的数据，根据自己的数据来
  3. List<DyingVideoDO> allVideoList = videoMapper.findAllVideoList();
  4.  
  5. List<Map<String, Object>> mapList = new ArrayList<>();
  6. for (DyingVideoDO row:allVideoList){
  7.     Map<String, Object> videoMap = BeanUtil.beanToMap(row);
  8.     mapList.add(videoMap);
  9. }
  10. //调用插入方法
  11. syncVideoEsService.createData(mapList);
  12.  
  13. } catch (Exception e) {
  14. //可有可无
  15.     log.error("sync video data error :{}", e);
  16.     return 0;
  17. }
  1. /**
  2.  * 创建文档
  3.  * @param list list
  4.  */
  5.  
  6. //索引名称
  7. private final String index = "t_dying_video";
  8. private final String alias = "t_dying_video";
  9. @Autowired
  10. private ElasticsearchService elasticsearchService;
  11.  
  12. public void createData(List<Map<String, Object>> list) throws Exception {
  13.     try {
  14.         if (!CollectionUtil.isEmpty(list)) {
  15.             List<String> idList = list.stream().map(it -> String.valueOf(it.get("id"))).collect(Collectors.toList());
  16.  
  17.             List<Map<String,Object>> jointVideoArray = new ArrayList<Map<String,Object>>();
  18.             list.stream().forEach(e -> {
  19.                 Map<String, Object> obj = new HashMap<String, Object>() {{
  20.                     put("keyid", e.get("id"));
  21.                     put("title", e.get("title"));//标题
  22.                     put("video_url", e.get("video_url"));
  23.                     put("pic_url",e.get("pic_url"));
  24.                     put("small_pic_url",e.get("small_pic_url"));
  25.                     put("media_content",e.get("media_content"));
  26.                     put("update_content",e.get("update_content"));
  27.                     put("publish_time",e.get("publish_time"));
  28.                     put("label",e.get("label"));
  29.                     put("title_desc",e.get("title_desc"));//由update_content清洗
  30.                     put("voice_content",e.get("voice_content"));//由media_content清洗
  31.                 }};
  32.                 jointVideoArray.add(obj);
  33.             });
  34. //调用下面插入类中的方法
  35.             //elasticsearchService.batchDeleteRequest(alias, idList);
  36.             //elasticsearchService.batchDeleteRequest(jointAlias, idList);
  37.             elasticsearchService.batchInsertRequest(alias, jointVideoArray);
  38.  
  39.         }
  40.     } catch (Exception e) {
  41.         log.error("插入数据到ES异常",e);
  42.         throw e;
  43.     }
  44. }

3.插入类

  1. /**
  2.  * 批量新增文档
  3.  */
  4. public boolean batchInsertRequest(String index, List<Map<String, Object>> list) throws Exception {
  5.     BulkRequest request = new BulkRequest();
  6.     for (int i = 0; i < list.size(); i++) {
  7.         Map<String, Object> item = list.get(i);
  8.         request.add(new IndexRequest(index).id(String.valueOf(item.get("keyid"))).source(item, XContentType.JSON));
  9.     }
  10.     BulkResponse bulk = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
  11.     if (bulk.status().getStatus() == 200) {
  12.         if (!bulk.hasFailures()) {
  13.             return true;
  14.         }
  15.         log.error("批量创建索引{}文档失败", index);
  16.     }
  17.     return false;
  18. }

--------附录（新增，删除，更新，插入等）

  1. package com.dengtacj.synces.service;
  2.  
  3. import cn.hutool.core.bean.BeanUtil;
  4. import lombok.extern.slf4j.Slf4j;
  5. import org.elasticsearch.action.admin.indices.alias.Alias;
  6. import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
  7. import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
  8. import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
  9. import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
  10. import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
  11. import org.elasticsearch.action.bulk.BulkRequest;
  12. import org.elasticsearch.action.bulk.BulkResponse;
  13. import org.elasticsearch.action.delete.DeleteRequest;
  14. import org.elasticsearch.action.index.IndexRequest;
  15. import org.elasticsearch.action.index.IndexResponse;
  16. import org.elasticsearch.action.search.SearchRequest;
  17. import org.elasticsearch.action.support.master.AcknowledgedResponse;
  18. import org.elasticsearch.action.update.UpdateRequest;
  19. import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
  20. import org.elasticsearch.client.RequestOptions;
  21. import org.elasticsearch.client.RestHighLevelClient;
  22. import org.elasticsearch.client.indices.GetIndexRequest;
  23. import org.elasticsearch.common.settings.Settings;
  24. import org.elasticsearch.common.xcontent.XContentType;
  25. import org.elasticsearch.index.reindex.BulkByScrollResponse;
  26. import org.elasticsearch.index.reindex.DeleteByQueryRequest;
  27. import org.springframework.beans.factory.annotation.Autowired;
  28. import org.springframework.stereotype.Service;
  29.  
  30. import java.io.IOException;
  31. import java.util.List;
  32. import java.util.Map;
  33.  
  34. /**
  35.  * Elasticsearch服务
  36.  *
  37.  */
  38. @Slf4j
  39. @Service
  40. public class ElasticsearchService {
  41.  
  42.  
  43.     @Autowired
  44.     public RestHighLevelClient restHighLevelClient;
  45.  
  46.     protected static final RequestOptions COMMON_OPTIONS;
  47.  
  48.     static {
  49.         RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
  50.         // 默认缓冲限制为100MB，此处修改为30MB。
  51.         builder.setHttpAsyncResponseConsumerFactory(new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
  52.         COMMON_OPTIONS = builder.build();
  53.     }
  54.  
  55.  
  56.     /**
  57.      * 创建索引
  58.      *
  59.      * @param index 索引名称
  60.      * @return 是否成功
  61.      */
  62.     public boolean createIndexRequest(String index) {
  63.         CreateIndexRequest createIndexRequest = new CreateIndexRequest(index)
  64.                 .settings(Settings.builder().put("index.number_of_shards", 3).put("index.number_of_replicas", 0));
  65.  
  66.         try {
  67.             CreateIndexResponse response = restHighLevelClient.indices().create(createIndexRequest, COMMON_OPTIONS);
  68.             log.info("所有节点确认响应 : {}", response.isAcknowledged());
  69.             log.info("所有分片的复制未超时 :{}", response.isShardsAcknowledged());
  70.             log.info("创建索引【{}】成功", index);
  71.             return true;
  72.         } catch (IOException e) {
  73.             log.error("创建索引库【{}】失败", index, e);
  74.         }
  75.         return false;
  76.     }
  77.  
  78.  
  79.     /**
  80.      * 创建索引
  81.      *
  82.      * @param index   索引名称
  83.      * @param mapping 索引结构
  84.      * @return 是否成功
  85.      */
  86.     public boolean createIndexRequest(String index, String mapping) {
  87.         CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
  88.         try {
  89.             createIndexRequest.settings(Settings.builder()
  90.                             // 分片数
  91.                             .put("index.number_of_shards", 3)
  92.                             // 副本数
  93.                             .put("index.number_of_replicas", 0)
  94.                     // 默认分词器
  95. //                    .put("analysis.analyzer.default.tokenizer", "index_ansj")
  96.             );
  97.             createIndexRequest.mapping("_doc", mapping, XContentType.JSON);
  98.             CreateIndexResponse response = restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
  99.             log.info("所有节点确认响应 : {}", response.isAcknowledged());
  100.             log.info("所有分片的复制未超时 :{}", response.isShardsAcknowledged());
  101.             log.info("创建索引【{}】成功", index);
  102.             return true;
  103.         } catch (Exception e) {
  104.             log.error("创建索引库【{}】失败", index, e);
  105.         }
  106.         return false;
  107.     }
  108.  
  109.  
  110.     /**
  111.      * 创建索引
  112.      *
  113.      * @param index    索引名称
  114.      * @param settings 索引设置
  115.      * @param mapping  索引结构
  116.      * @return 是否成功
  117.      */
  118.     public boolean createIndexRequest(String index, String settings, String mapping) {
  119.         CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
  120.         try {
  121.             createIndexRequest.settings(settings, XContentType.JSON);
  122.             createIndexRequest.mapping("_doc", mapping, XContentType.JSON);
  123.             CreateIndexResponse response = restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
  124.             log.info("所有节点确认响应 : {}", response.isAcknowledged());
  125.             log.info("所有分片的复制未超时 :{}", response.isShardsAcknowledged());
  126.             log.info("创建索引【{}】成功", index);
  127.             return true;
  128.         } catch (Exception e) {
  129.             log.error("创建索引库【{}】失败", index, e);
  130.         }
  131.         return false;
  132.     }
  133.  
  134.  
  135.     /**
  136.      * 创建索引
  137.      *
  138.      * @param index    索引名称
  139.      * @param settings 索引设置
  140.      * @param mapping  索引结构
  141.      * @param alias    索引别名
  142.      * @return 是否成功
  143.      */
  144.     public boolean createIndexRequest(String index, String settings, String mapping, String alias) {
  145.         CreateIndexRequest createIndexRequest = new CreateIndexRequest(index);
  146.         try {
  147.             createIndexRequest.settings(settings, XContentType.JSON);
  148.             createIndexRequest.mapping("_doc", mapping, XContentType.JSON);
  149.             createIndexRequest.alias(new Alias(alias));
  150.             CreateIndexResponse response = restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
  151.             log.info("所有节点确认响应 : {}", response.isAcknowledged());
  152.             log.info("所有分片的复制未超时 :{}", response.isShardsAcknowledged());
  153.             log.info("创建索引【{}】成功", index);
  154.             return true;
  155.         } catch (Exception e) {
  156.             log.error("创建索引库【{}】失败", index, e);
  157.         }
  158.         return false;
  159.     }
  160.  
  161.  
  162.     /**
  163.      * 删除索引
  164.      *
  165.      * @param index 索引名称
  166.      */
  167.     public boolean deleteIndexRequest(String index) {
  168.         try {
  169.             boolean exists = restHighLevelClient.indices().exists(new GetIndexRequest(index), RequestOptions.DEFAULT);
  170.             if (exists) {
  171.                 AcknowledgedResponse response = restHighLevelClient.indices().delete(new DeleteIndexRequest(index), COMMON_OPTIONS);
  172.                 // 判断是否确认响应
  173.                 if (response.isAcknowledged()) {
  174.                     return true;
  175.                 }
  176.             } else {
  177.                 log.info("索引【{}】不存在", index);
  178.                 return true;
  179.             }
  180.         } catch (IOException e) {
  181.             log.error("删除索引库【{}】失败", index, e);
  182.         }
  183.         return false;
  184.     }
  185.  
  186.  
  187.     /**
  188.      * 新增文档
  189.      */
  190.     public boolean insertRequest(String index, String id, Object object) {
  191.         IndexRequest indexRequest = new IndexRequest(index).id(id).source(BeanUtil.beanToMap(object), XContentType.JSON);
  192.         try {
  193.             IndexResponse indexResponse = restHighLevelClient.index(indexRequest, COMMON_OPTIONS);
  194.             if (indexResponse.status().getStatus() == 200) {
  195.                 log.info("创建索引{}文档成功", index);
  196.                 return true;
  197.             }
  198.         } catch (IOException e) {
  199.             log.error("创建索引文档 {" + index + "} 数据 {" + object + "} 失败", e);
  200.         }
  201.         return false;
  202.     }
  203.  
  204.  
  205.     /**
  206.      * 新增文档
  207.      */
  208.     public boolean insertRequest(String index, Map<String, Object> item) {
  209.         IndexRequest indexRequest = new IndexRequest(index).id((String) item.get("id")).source(item, XContentType.JSON);
  210.         try {
  211.             IndexResponse indexResponse = restHighLevelClient.index(indexRequest, COMMON_OPTIONS);
  212.             if (indexResponse.status().getStatus() == 200) {
  213.                 log.info("创建索引{}文档成功", index);
  214.                 return true;
  215.             }
  216.         } catch (IOException e) {
  217.             log.error("创建索引文档 {" + index + "} 数据 {" + item + "} 失败", e);
  218.         }
  219.         return false;
  220.     }
  221.  
  222.  
  223.     /**
  224.      * 批量新增文档
  225.      */
  226.     public boolean batchInsertRequest(String index, List<Map<String, Object>> list) throws Exception {
  227.         BulkRequest request = new BulkRequest();
  228.         for (int i = 0; i < list.size(); i++) {
  229.             Map<String, Object> item = list.get(i);
  230.             request.add(new IndexRequest(index).id(String.valueOf(item.get("keyid"))).source(item, XContentType.JSON));
  231.         }
  232.         BulkResponse bulk = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
  233.         if (bulk.status().getStatus() == 200) {
  234.             if (!bulk.hasFailures()) {
  235.                 return true;
  236.             }
  237.             log.error("批量创建索引{}文档失败", index);
  238.         }
  239.         return false;
  240.     }
  241.  
  242.  
  243.     /**
  244.      * 修改文档
  245.      */
  246.     public void updateRequest(String index, String id, Object object) {
  247.         UpdateRequest updateRequest = new UpdateRequest(index, id);
  248.         updateRequest.doc(BeanUtil.beanToMap(object), XContentType.JSON);
  249.         try {
  250.             restHighLevelClient.update(updateRequest, COMMON_OPTIONS);
  251.         } catch (IOException e) {
  252.             log.error("更新索引文档 {" + index + "} 数据 {" + object + "} 失败", e);
  253.         }
  254.     }
  255.  
  256.  
  257.     /**
  258.      * 批量修改文档
  259.      */
  260.     public boolean batchUpdateRequest(String index, List<Map<String, Object>> list) throws Exception {
  261.         BulkRequest request = new BulkRequest();
  262.         for (int i = 0; i < list.size(); i++) {
  263.             Map<String, Object> item = list.get(i);
  264.             request.add(new UpdateRequest(index, (String) item.get("keyid")).doc(item, XContentType.JSON));
  265.         }
  266.         BulkResponse bulk = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
  267.         if (bulk.status().getStatus() == 200) {
  268.             if (!bulk.hasFailures()) {
  269.                 log.info("批量修改索引{}文档成功", index);
  270.                 return true;
  271.             }
  272.             log.error("批量修改索引{}文档失败,失败原因:{}", index, bulk.buildFailureMessage());
  273.         }
  274.         return false;
  275.     }
  276.  
  277.  
  278.     /**
  279.      * 删除文档
  280.      */
  281.     public void deleteRequest(String index, String id) {
  282.         DeleteRequest deleteRequest = new DeleteRequest(index, id);
  283.         try {
  284.             restHighLevelClient.delete(deleteRequest, COMMON_OPTIONS);
  285.         } catch (IOException e) {
  286.             log.error("删除索引文档 {" + index + "} 数据id {" + id + "} 失败", e);
  287.         }
  288.     }
  289.  
  290.  
  291.     /**
  292.      * 批量删除文档
  293.      */
  294.     public boolean batchDeleteRequest(String index, List<String> list) throws Exception {
  295.         BulkRequest request = new BulkRequest();
  296.         for (int i = 0; i < list.size(); i++) {
  297.             request.add(new DeleteRequest(index, list.get(i)));
  298.         }
  299.         BulkResponse bulk = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
  300.         if (bulk.status().getStatus() == 200) {
  301.             if (!bulk.hasFailures()) {
  302.                 log.info("批量删除索引{}文档成功 count:{}", index,list.size());
  303.                 return true;
  304.             }
  305.             log.error("批量删除索引{}文档成功,失败原因:{}", index, bulk.buildFailureMessage());
  306.         }
  307.         return false;
  308.     }
  309.  
  310.  
  311.  
  312.     public boolean cleanIndex(String index){
  313.         try {
  314.             DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest(index);
  315.             deleteRequest.setRefresh(true);
  316.             BulkByScrollResponse response = restHighLevelClient.deleteByQuery(deleteRequest,RequestOptions.DEFAULT);
  317.             return true;
  318.         }catch (IOException e){
  319.             log.error("clean index :{} error : {}",index,e);
  320.             return false;
  321.         }
  322.     }
  323.  
  324.  
  325.     public boolean refreshIndex(String... index){
  326.         try {
  327.             RefreshRequest refreshRequest = new RefreshRequest();
  328.             refreshRequest.indices(index);
  329.             RefreshResponse response = restHighLevelClient.indices().refresh(refreshRequest,RequestOptions.DEFAULT);
  330.             log.info("refresh  {}  response status : {}",index,response.getStatus());
  331.             return true;
  332.         }catch (IOException e){
  333.             log.error("refresh index :{} error : {}",index,e);
  334.             return false;
  335.         }
  336.     }
  337.  
  338.  
  339. }
创建索引的调用方法，mapping和seeting只是读取你的索引配置文件
比如我的文件是:

 

 
 
{ "settings": { "number_of_shards": 5, "number_of_replicas": 1, "index": { "analysis.analyzer.default.type" : "ik_max_word"} } }

 
 
  1. public boolean createIndex() {
  2.     String mapping = null;
  3.     String settings = null;
  4.     try {
  5.         mapping = FileUtil.readString(ResourceUtils.getFile("classpath:index/mapping/t_dying_video.json"), "UTF-8");
  6.         settings = FileUtil.readString(ResourceUtils.getFile("classpath:index/settings/t_dying_video.json"), "UTF-8");
  7.         if (StrUtil.isNotBlank(mapping) && StrUtil.isNotBlank(settings)) {
  8.             return elasticsearchService.createIndexRequest(index, settings, mapping, alias);
  9.         }
  10.     } catch (Exception e) {
  11.         log.error("读取索引【{}】的Mapping文件失败", index, e);
  12.     }
  13.     return false;
  14. }
 
 
 


                

        
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/68075
推荐阅读
相关标签
Copyright ©  2003-2013 www.wpsshop.cn  版权所有，并保留所有权利。

  
闽ICP备14008679号