瀏覽代碼

add manual load data

kerryzhang 5 年之前
父節點
當前提交
4cee2dc244

+ 4 - 1
elastic-publish-service/req/main-controller.http 查看文件

21
 
21
 
22
 {
22
 {
23
   "agencyPath": "/1",
23
   "agencyPath": "/1",
24
-  "keyword": "b楼",
24
+  "keyword": "科苑一区",
25
   "page": 1,
25
   "page": 1,
26
   "size": 10,
26
   "size": 10,
27
   "regulatoryLevelId": "",
27
   "regulatoryLevelId": "",
38
 
38
 
39
 ### load data
39
 ### load data
40
 
40
 
41
+GET http://{{host}}/set-load
42
+
43
+### load data
41
 
44
 
42
 
45
 
43
 GET http://{{host}}/loadByAgencyIds?agencyIds=190084,196581
46
 GET http://{{host}}/loadByAgencyIds?agencyIds=190084,196581

+ 0 - 4
elastic-publish-service/src/main/java/com/vcarecity/publish/constants/ElasticConstant.java 查看文件

9
 public class ElasticConstant {
9
 public class ElasticConstant {
10
     public static final long CHECK_TIME_INTERVAL = 1000 * 3 * 60;
10
     public static final long CHECK_TIME_INTERVAL = 1000 * 3 * 60;
11
 
11
 
12
-    public static final String UNIT_AGENCY_INDEX = "unit-agency";
13
-
14
-
15
     public static final int QUERY_PAGE_COUNT = 2000;
12
     public static final int QUERY_PAGE_COUNT = 2000;
16
 
13
 
17
 
14
 
22
     public static final DateTimeFormatter ES_DT_FORMATTER = DateTimeFormatter.ofPattern(ES_DATETIME_PATTERN);
19
     public static final DateTimeFormatter ES_DT_FORMATTER = DateTimeFormatter.ofPattern(ES_DATETIME_PATTERN);
23
 
20
 
24
 
21
 
25
-
26
 }
22
 }

+ 5 - 3
elastic-publish-service/src/main/java/com/vcarecity/publish/controller/MainController.java 查看文件

3
 import com.vcarecity.publish.annotation.SkipMethod;
3
 import com.vcarecity.publish.annotation.SkipMethod;
4
 import com.vcarecity.publish.api.ApiResult;
4
 import com.vcarecity.publish.api.ApiResult;
5
 import com.vcarecity.publish.api.ApiStatus;
5
 import com.vcarecity.publish.api.ApiStatus;
6
-import com.vcarecity.publish.constants.ElasticConstant;
7
 import com.vcarecity.publish.elastic.service.ElasticLoadDataService;
6
 import com.vcarecity.publish.elastic.service.ElasticLoadDataService;
8
 import com.vcarecity.publish.elastic.service.ElasticUpdateService;
7
 import com.vcarecity.publish.elastic.service.ElasticUpdateService;
9
 import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
8
 import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
38
     private final ElasticUpdateService elasticUpdateService;
37
     private final ElasticUpdateService elasticUpdateService;
39
     private final LoadDataStatusService loadDataStatusService;
38
     private final LoadDataStatusService loadDataStatusService;
40
 
39
 
40
+    @Value("${app.index.name}")
41
+    private String indexName;
42
+
41
     public MainController(UnitService unitService,
43
     public MainController(UnitService unitService,
42
                           ElasticLoadDataService elasticLoadDataService,
44
                           ElasticLoadDataService elasticLoadDataService,
43
                           LoadDataStatusService loadDataStatusService,
45
                           LoadDataStatusService loadDataStatusService,
119
     @PostMapping("/query-unit-agency")
121
     @PostMapping("/query-unit-agency")
120
     public ApiResult queryUnitAgency(@RequestBody @Valid UnitAgencyQuery unitAgencyQuery) throws IOException {
122
     public ApiResult queryUnitAgency(@RequestBody @Valid UnitAgencyQuery unitAgencyQuery) throws IOException {
121
 
123
 
122
-        TotalResultDTO<List<UnitIdDTO>> totalResult = unitService.getUnitAgencyByQuery(unitAgencyQuery, UnitIdDTO.class, ElasticConstant.UNIT_AGENCY_INDEX);
124
+        TotalResultDTO<List<UnitIdDTO>> totalResult = unitService.getUnitAgencyByQuery(unitAgencyQuery, UnitIdDTO.class, indexName);
123
 
125
 
124
         return ApiResult.builder()
126
         return ApiResult.builder()
125
                 .status(ApiStatus.SUCCESS_CODE)
127
                 .status(ApiStatus.SUCCESS_CODE)
130
 
132
 
131
     @PostMapping("/test/query-unit-agency")
133
     @PostMapping("/test/query-unit-agency")
132
     public ApiResult testQueryUnitAgency(@RequestBody @Valid UnitAgencyQuery unitAgencyQuery) throws IOException {
134
     public ApiResult testQueryUnitAgency(@RequestBody @Valid UnitAgencyQuery unitAgencyQuery) throws IOException {
133
-        TotalResultDTO<List<UnitAgencyDTO>> totalResult = unitService.getUnitAgencyByQuery(unitAgencyQuery, UnitAgencyDTO.class, ElasticConstant.UNIT_AGENCY_INDEX);
135
+        TotalResultDTO<List<UnitAgencyDTO>> totalResult = unitService.getUnitAgencyByQuery(unitAgencyQuery, UnitAgencyDTO.class, indexName);
134
         return ApiResult.builder()
136
         return ApiResult.builder()
135
                 .status(ApiStatus.SUCCESS_CODE)
137
                 .status(ApiStatus.SUCCESS_CODE)
136
                 .data(totalResult.getData())
138
                 .data(totalResult.getData())

+ 5 - 2
elastic-publish-service/src/main/java/com/vcarecity/publish/elastic/service/impl/ElasticLoadDataServiceImpl.java 查看文件

7
 import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
7
 import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
8
 import com.vcarecity.publish.mapper.UnitMapper;
8
 import com.vcarecity.publish.mapper.UnitMapper;
9
 import lombok.extern.slf4j.Slf4j;
9
 import lombok.extern.slf4j.Slf4j;
10
+import org.springframework.beans.factory.annotation.Value;
10
 import org.springframework.stereotype.Service;
11
 import org.springframework.stereotype.Service;
11
 
12
 
12
 import java.io.IOException;
13
 import java.io.IOException;
25
     private final ElasticUpdateService elasticUpdateService;
26
     private final ElasticUpdateService elasticUpdateService;
26
     private final ElasticIndexService elasticIndexService;
27
     private final ElasticIndexService elasticIndexService;
27
     private final int pageCount;
28
     private final int pageCount;
28
-    private final String index;
29
+
30
+
31
+    @Value("${app.index.name}")
32
+    private String index;
29
 
33
 
30
     public ElasticLoadDataServiceImpl(UnitMapper unitMapper,
34
     public ElasticLoadDataServiceImpl(UnitMapper unitMapper,
31
                                       ElasticUpdateService elasticUpdateService,
35
                                       ElasticUpdateService elasticUpdateService,
34
         this.elasticUpdateService = elasticUpdateService;
38
         this.elasticUpdateService = elasticUpdateService;
35
         this.elasticIndexService = elasticIndexService;
39
         this.elasticIndexService = elasticIndexService;
36
         this.pageCount = ElasticConstant.QUERY_PAGE_COUNT;
40
         this.pageCount = ElasticConstant.QUERY_PAGE_COUNT;
37
-        this.index = ElasticConstant.UNIT_AGENCY_INDEX;
38
     }
41
     }
39
 
42
 
40
 
43
 

+ 7 - 2
elastic-publish-service/src/main/java/com/vcarecity/publish/elastic/service/impl/ElasticQueryServiceImpl.java 查看文件

1
 package com.vcarecity.publish.elastic.service.impl;
1
 package com.vcarecity.publish.elastic.service.impl;
2
 
2
 
3
-import com.vcarecity.publish.constants.ElasticConstant;
4
 import com.vcarecity.publish.elastic.service.ElasticQueryService;
3
 import com.vcarecity.publish.elastic.service.ElasticQueryService;
5
 import lombok.extern.slf4j.Slf4j;
4
 import lombok.extern.slf4j.Slf4j;
6
 import org.elasticsearch.action.search.SearchRequest;
5
 import org.elasticsearch.action.search.SearchRequest;
11
 import org.elasticsearch.index.query.TermQueryBuilder;
10
 import org.elasticsearch.index.query.TermQueryBuilder;
12
 import org.elasticsearch.search.SearchHit;
11
 import org.elasticsearch.search.SearchHit;
13
 import org.elasticsearch.search.builder.SearchSourceBuilder;
12
 import org.elasticsearch.search.builder.SearchSourceBuilder;
13
+import org.springframework.beans.factory.annotation.Value;
14
 import org.springframework.stereotype.Service;
14
 import org.springframework.stereotype.Service;
15
 
15
 
16
 import java.io.IOException;
16
 import java.io.IOException;
25
 
25
 
26
     private final RestHighLevelClient restHighLevelClient;
26
     private final RestHighLevelClient restHighLevelClient;
27
 
27
 
28
+
29
+    @Value("${app.index.name}")
30
+    private String indexName;
31
+
28
     public ElasticQueryServiceImpl(RestHighLevelClient restHighLevelClient) {
32
     public ElasticQueryServiceImpl(RestHighLevelClient restHighLevelClient) {
33
+
29
         this.restHighLevelClient = restHighLevelClient;
34
         this.restHighLevelClient = restHighLevelClient;
30
     }
35
     }
31
 
36
 
39
         searchSourceBuilder.query(unitId);
44
         searchSourceBuilder.query(unitId);
40
 
45
 
41
 
46
 
42
-        SearchRequest request = new SearchRequest(ElasticConstant.UNIT_AGENCY_INDEX);
47
+        SearchRequest request = new SearchRequest(indexName);
43
         request.source(searchSourceBuilder);
48
         request.source(searchSourceBuilder);
44
 
49
 
45
         SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);
50
         SearchResponse response = restHighLevelClient.search(request, RequestOptions.DEFAULT);

+ 10 - 7
elastic-publish-service/src/main/java/com/vcarecity/publish/elastic/service/impl/ElasticUpdateServiceImpl.java 查看文件

5
 import com.fasterxml.jackson.databind.DeserializationFeature;
5
 import com.fasterxml.jackson.databind.DeserializationFeature;
6
 import com.fasterxml.jackson.databind.ObjectMapper;
6
 import com.fasterxml.jackson.databind.ObjectMapper;
7
 import com.vcarecity.elastic.util.AgencyPathUtil;
7
 import com.vcarecity.elastic.util.AgencyPathUtil;
8
-import com.vcarecity.elastic.util.SnowFlake;
9
-import com.vcarecity.publish.constants.ElasticConstant;
10
 import com.vcarecity.publish.elastic.entity.UnitAgencyDetailEntity;
8
 import com.vcarecity.publish.elastic.entity.UnitAgencyDetailEntity;
11
 import com.vcarecity.publish.elastic.service.ElasticUpdateService;
9
 import com.vcarecity.publish.elastic.service.ElasticUpdateService;
12
 import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
10
 import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
24
 import org.elasticsearch.index.reindex.UpdateByQueryRequest;
22
 import org.elasticsearch.index.reindex.UpdateByQueryRequest;
25
 import org.elasticsearch.script.Script;
23
 import org.elasticsearch.script.Script;
26
 import org.elasticsearch.script.ScriptType;
24
 import org.elasticsearch.script.ScriptType;
25
+import org.springframework.beans.factory.annotation.Value;
27
 import org.springframework.cglib.beans.BeanCopier;
26
 import org.springframework.cglib.beans.BeanCopier;
28
 import org.springframework.stereotype.Service;
27
 import org.springframework.stereotype.Service;
29
 
28
 
42
 @Service
41
 @Service
43
 public class ElasticUpdateServiceImpl implements ElasticUpdateService {
42
 public class ElasticUpdateServiceImpl implements ElasticUpdateService {
44
 
43
 
45
-    public final String index;
44
+    @Value("${app.index.name}")
45
+    public String index;
46
 
46
 
47
     private final RestHighLevelClient restHighLevelClient;
47
     private final RestHighLevelClient restHighLevelClient;
48
     private final ObjectMapper objectMapper;
48
     private final ObjectMapper objectMapper;
58
         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
58
         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
59
 
59
 
60
         this.objectMapper = objectMapper;
60
         this.objectMapper = objectMapper;
61
-        this.index = ElasticConstant.UNIT_AGENCY_INDEX;
61
+
62
     }
62
     }
63
 
63
 
64
 
64
 
132
     public List<UnitAgencyMergeEntity> updateDataByQuery(String key, List<UnitAgencyMergeEntity> data) throws IOException {
132
     public List<UnitAgencyMergeEntity> updateDataByQuery(String key, List<UnitAgencyMergeEntity> data) throws IOException {
133
         List<UnitAgencyMergeEntity> addList = new ArrayList<>();
133
         List<UnitAgencyMergeEntity> addList = new ArrayList<>();
134
         final BeanCopier beanCopier = BeanCopier.create(UnitAgencyMergeEntity.class, UnitAgencyDetailEntity.class, false);
134
         final BeanCopier beanCopier = BeanCopier.create(UnitAgencyMergeEntity.class, UnitAgencyDetailEntity.class, false);
135
-        for (UnitAgencyMergeEntity item : data) {
135
+
136
+
137
+        for (int i = 0; i < data.size(); i++) {
138
+            UnitAgencyMergeEntity item = data.get(i);
136
 
139
 
137
             Map<String, Object> params = createElasticDataMap(beanCopier, item);
140
             Map<String, Object> params = createElasticDataMap(beanCopier, item);
138
             Object value = params.get(key);
141
             Object value = params.get(key);
144
 
147
 
145
             BulkByScrollResponse response = restHighLevelClient.updateByQuery(request, RequestOptions.DEFAULT);
148
             BulkByScrollResponse response = restHighLevelClient.updateByQuery(request, RequestOptions.DEFAULT);
146
 
149
 
147
-            logger.info("response:{}", response);
150
+            logger.info("updateDataByQuery.index: {} resp:{}", i, response);
148
 
151
 
149
             if (response.getUpdated() == 0L) {
152
             if (response.getUpdated() == 0L) {
150
                 addList.add(item);
153
                 addList.add(item);
205
 
208
 
206
     private IndexRequest createIndexRequest(String index, Map<String, Object> value) {
209
     private IndexRequest createIndexRequest(String index, Map<String, Object> value) {
207
         IndexRequest indexRequest = new IndexRequest(index);
210
         IndexRequest indexRequest = new IndexRequest(index);
208
-        indexRequest.id(SnowFlake.nextId() + "");
211
+        // indexRequest.id(SnowFlake.nextId() + "");
209
         indexRequest.source(value, XContentType.JSON);
212
         indexRequest.source(value, XContentType.JSON);
210
         return indexRequest;
213
         return indexRequest;
211
     }
214
     }

+ 38 - 23
elastic-publish-service/src/main/java/com/vcarecity/publish/task/CheckUpdateTask.java 查看文件

9
 import com.vcarecity.publish.service.LoadDataStatusService;
9
 import com.vcarecity.publish.service.LoadDataStatusService;
10
 import com.vcarecity.publish.util.DateUtil;
10
 import com.vcarecity.publish.util.DateUtil;
11
 import lombok.extern.slf4j.Slf4j;
11
 import lombok.extern.slf4j.Slf4j;
12
+import org.springframework.beans.factory.annotation.Value;
12
 import org.springframework.scheduling.annotation.Scheduled;
13
 import org.springframework.scheduling.annotation.Scheduled;
13
 import org.springframework.stereotype.Component;
14
 import org.springframework.stereotype.Component;
14
 
15
 
32
     private final UnitMapper unitMapper;
33
     private final UnitMapper unitMapper;
33
 
34
 
34
 
35
 
36
+    @Value("${app.index.update-unit}")
37
+    private Boolean updateUnit;
38
+
39
+
40
+    @Value("${app.index.update-agency}")
41
+    private Boolean updateAgency;
42
+
43
+
35
     public CheckUpdateTask(CheckTimeService checkTimeService,
44
     public CheckUpdateTask(CheckTimeService checkTimeService,
36
                            ElasticUpdateService elasticUpdateService,
45
                            ElasticUpdateService elasticUpdateService,
37
                            LoadDataStatusService loadDataStatusService, AgencyMapper agencyMapper,
46
                            LoadDataStatusService loadDataStatusService, AgencyMapper agencyMapper,
60
         List<UnitAgencyMergeEntity> updateUaList;
69
         List<UnitAgencyMergeEntity> updateUaList;
61
         String fromDateTime;
70
         String fromDateTime;
62
 
71
 
63
-        // 获取上一次的时间
64
-        fromDateTime = checkTimeService.getLastDateTime("T_UNIT");
72
+        if (updateUnit) {
73
+
74
+
75
+            // 获取上一次的时间
76
+            fromDateTime = checkTimeService.getLastDateTime("T_UNIT");
65
 
77
 
66
-        logger.info("start checkUpdateTask on T_UNIT from:{}. to:{}", fromDateTime, toDateTime);
78
+            logger.info("start checkUpdateTask on T_UNIT from:{}. to:{}", fromDateTime, toDateTime);
67
 
79
 
68
-        try {
69
-            updateUaList = unitMapper.findUpdateUaList(fromDateTime, toDateTime);
80
+            try {
81
+                updateUaList = unitMapper.findUpdateUaList(fromDateTime, toDateTime);
70
 
82
 
71
-            logger.debug("T_UNIT update list size = {}", updateUaList.size());
83
+                logger.debug("T_UNIT update list size = {}", updateUaList.size());
72
 
84
 
73
-            if (!updateUaList.isEmpty()) {
74
-                elasticUpdateService.handlerDataChange("unitId", updateUaList);
85
+                if (!updateUaList.isEmpty()) {
86
+                    elasticUpdateService.handlerDataChange("unitId", updateUaList);
87
+                }
88
+                checkTimeService.setLastDateTime("T_UNIT", curCheckTime);
89
+                updateUaList.clear();
90
+            } catch (Exception e) {
91
+                e.printStackTrace();
75
             }
92
             }
76
-            checkTimeService.setLastDateTime("T_UNIT", curCheckTime);
77
-            updateUaList.clear();
78
-        } catch (Exception e) {
79
-            e.printStackTrace();
80
         }
93
         }
81
 
94
 
82
-        fromDateTime = checkTimeService.getLastDateTime("T_AGENCY");
95
+        if (updateAgency) {
96
+            fromDateTime = checkTimeService.getLastDateTime("T_AGENCY");
83
 
97
 
84
-        logger.info("start checkUpdateTask on T_AGENCY from:{}. to:{}", fromDateTime, toDateTime);
98
+            logger.info("start checkUpdateTask on T_AGENCY from:{}. to:{}", fromDateTime, toDateTime);
85
 
99
 
86
-        try {
87
-            updateUaList = agencyMapper.findUpdateUaList(fromDateTime, toDateTime);
100
+            try {
101
+                updateUaList = agencyMapper.findUpdateUaList(fromDateTime, toDateTime);
88
 
102
 
89
-            logger.debug("T_AGENCY update list size = {}", updateUaList.size());
103
+                logger.debug("T_AGENCY update list size = {}", updateUaList.size());
90
 
104
 
91
-            if (!updateUaList.isEmpty()) {
92
-                elasticUpdateService.handlerDataChange("agencyId", updateUaList);
105
+                if (!updateUaList.isEmpty()) {
106
+                    elasticUpdateService.handlerDataChange("agencyId", updateUaList);
107
+                }
108
+                checkTimeService.setLastDateTime("T_AGENCY", curCheckTime);
109
+                updateUaList.clear();
110
+            } catch (Exception e) {
111
+                e.printStackTrace();
93
             }
112
             }
94
-            checkTimeService.setLastDateTime("T_AGENCY", curCheckTime);
95
-            updateUaList.clear();
96
-        } catch (Exception e) {
97
-            e.printStackTrace();
98
         }
113
         }
99
     }
114
     }
100
 
115
 

+ 3 - 0
elastic-publish-service/src/main/resources/application-beta.yml 查看文件

25
 debug: true
25
 debug: true
26
 app:
26
 app:
27
   index:
27
   index:
28
+    name: unit-agency
28
     config: index-text
29
     config: index-text
29
     reload: true
30
     reload: true
31
+    update-unit: true
32
+    update-agency: false
30
   scheduling:
33
   scheduling:
31
     enable: true
34
     enable: true

+ 3 - 0
elastic-publish-service/src/main/resources/application-dev.yml 查看文件

26
 debug: true
26
 debug: true
27
 app:
27
 app:
28
   index:
28
   index:
29
+    name: unit-agency
29
     config: index-text
30
     config: index-text
30
     reload: true
31
     reload: true
32
+    update-unit: true
33
+    update-agency: false
31
   scheduling:
34
   scheduling:
32
     enable: false
35
     enable: false

+ 3 - 0
elastic-publish-service/src/main/resources/application-production.yml 查看文件

27
     map-underscore-to-camel-case: true
27
     map-underscore-to-camel-case: true
28
 app:
28
 app:
29
   index:
29
   index:
30
+    name: unit-agency
30
     config: index-text
31
     config: index-text
31
     reload: false
32
     reload: false
33
+    update-unit: true
34
+    update-agency: false
32
   scheduling:
35
   scheduling:
33
     enable: true
36
     enable: true

+ 3 - 2
elastic-publish-service/src/test/java/com/vcarecity/publish/elastic/service/impl/ElasticIndexServiceImplTest.java 查看文件

1
 package com.vcarecity.publish.elastic.service.impl;
1
 package com.vcarecity.publish.elastic.service.impl;
2
 
2
 
3
-import com.vcarecity.publish.constants.ElasticConstant;
4
 import com.vcarecity.publish.elastic.service.ElasticIndexService;
3
 import com.vcarecity.publish.elastic.service.ElasticIndexService;
5
 import org.junit.jupiter.api.Test;
4
 import org.junit.jupiter.api.Test;
6
 import org.springframework.beans.factory.annotation.Autowired;
5
 import org.springframework.beans.factory.annotation.Autowired;
6
+import org.springframework.beans.factory.annotation.Value;
7
 import org.springframework.boot.test.context.SpringBootTest;
7
 import org.springframework.boot.test.context.SpringBootTest;
8
 import org.springframework.test.context.TestPropertySource;
8
 import org.springframework.test.context.TestPropertySource;
9
 
9
 
16
     @Autowired
16
     @Autowired
17
     private ElasticIndexService elasticIndexService;
17
     private ElasticIndexService elasticIndexService;
18
 
18
 
19
-    public static final String INDEX = ElasticConstant.UNIT_AGENCY_INDEX;
19
+    @Value("${app.index.name}")
20
+    public String INDEX;
20
 
21
 
21
     @Test
22
     @Test
22
     void indexExist() throws Exception {
23
     void indexExist() throws Exception {