|
|
@ -1,30 +1,52 @@
|
|
|
package com.yihu.iot.service.common;
|
|
|
|
|
|
|
|
|
import com.alibaba.druid.sql.ast.SQLExpr;
|
|
|
import com.alibaba.druid.sql.ast.expr.SQLQueryExpr;
|
|
|
import com.alibaba.druid.sql.parser.SQLExprParser;
|
|
|
import com.alibaba.fastjson.JSONArray;
|
|
|
import com.alibaba.fastjson.JSONObject;
|
|
|
import com.yihu.elasticsearch.ElasticSearchPool;
|
|
|
import com.yihu.iot.datainput.enums.DataTypeEnum;
|
|
|
import com.yihu.iot.datainput.service.DataStandardConvertService;
|
|
|
import com.yihu.jw.datainput.Data;
|
|
|
import com.yihu.jw.datainput.StepInfoDO;
|
|
|
import org.apache.commons.lang.StringUtils;
|
|
|
//import org.apache.hadoop.hbase.util.CollectionUtils;
|
|
|
import org.elasticsearch.action.search.SearchResponse;
|
|
|
import org.elasticsearch.index.query.*;
|
|
|
import org.elasticsearch.search.builder.SearchSourceBuilder;
|
|
|
import org.elasticsearch.search.sort.FieldSortBuilder;
|
|
|
import org.elasticsearch.search.sort.SortOrder;
|
|
|
import org.nlpcn.es4sql.domain.Select;
|
|
|
import org.nlpcn.es4sql.jdbc.ObjectResult;
|
|
|
import org.nlpcn.es4sql.jdbc.ObjectResultsExtractor;
|
|
|
import org.nlpcn.es4sql.parse.ElasticSqlExprParser;
|
|
|
import org.nlpcn.es4sql.parse.SqlParser;
|
|
|
import org.nlpcn.es4sql.query.AggregationQueryAction;
|
|
|
import org.nlpcn.es4sql.query.DefaultQueryAction;
|
|
|
import org.nlpcn.es4sql.query.SqlElasticSearchRequestBuilder;
|
|
|
import org.slf4j.Logger;
|
|
|
import org.slf4j.LoggerFactory;
|
|
|
import org.springframework.beans.BeanUtils;
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
import org.springframework.stereotype.Component;
|
|
|
import org.springframework.util.CollectionUtils;
|
|
|
|
|
|
import javax.annotation.PostConstruct;
|
|
|
import java.beans.PropertyDescriptor;
|
|
|
import java.lang.reflect.Field;
|
|
|
import java.sql.Timestamp;
|
|
|
import java.text.SimpleDateFormat;
|
|
|
import java.util.*;
|
|
|
|
|
|
//import org.apache.hadoop.hbase.util.CollectionUtils;
|
|
|
|
|
|
@Component
|
|
|
public class ElasticSearchQueryGenerator {
|
|
|
|
|
|
private Logger logger = LoggerFactory.getLogger(ElasticSearchQueryGenerator.class);
|
|
|
private static Map<String,Set<String>> fieldsMap = new HashMap<>();
|
|
|
@Autowired
|
|
|
private ElasticSearchPool elasticSearchPool;
|
|
|
|
|
|
@PostConstruct
|
|
|
public static void init(){
|
|
|
@ -42,6 +64,245 @@ public class ElasticSearchQueryGenerator {
|
|
|
fieldsMap.put("stepInfoList",fieldsSet2);
|
|
|
}
|
|
|
|
|
|
|
|
|
public Long excuteForLong(String sql, String esType, String esIndex) {
|
|
|
try {
|
|
|
SQLExprParser parser = new ElasticSqlExprParser(sql);
|
|
|
SQLExpr expr = parser.expr();
|
|
|
SQLQueryExpr queryExpr = (SQLQueryExpr) expr;
|
|
|
|
|
|
Select select = null;
|
|
|
select = new SqlParser().parseSelect(queryExpr);
|
|
|
|
|
|
//通过抽象语法树，封装成自定义的Select，包含了select、from、where group、limit等
|
|
|
AggregationQueryAction action = null;
|
|
|
DefaultQueryAction queryAction = null;
|
|
|
SqlElasticSearchRequestBuilder requestBuilder = null;
|
|
|
if (select.isAgg) {
|
|
|
//包含计算的的排序分组的
|
|
|
action = new AggregationQueryAction(elasticSearchPool.getClient(), select);
|
|
|
requestBuilder = action.explain();
|
|
|
} else {
|
|
|
//封装成自己的Select对象
|
|
|
queryAction = new DefaultQueryAction(elasticSearchPool.getClient(), select);
|
|
|
requestBuilder = queryAction.explain();
|
|
|
}
|
|
|
SearchResponse response = (SearchResponse) requestBuilder.get();
|
|
|
ObjectResult temp = new ObjectResultsExtractor(true, true, true).extractResults(response.getAggregations(), true);
|
|
|
Long Longvalue = ((Double) temp.getLines().get(0).get(0)).longValue();
|
|
|
return Longvalue;
|
|
|
} catch (Exception e) {
|
|
|
logger.error(e.getMessage());
|
|
|
}
|
|
|
return 0L;
|
|
|
}
|
|
|
|
|
|
public List excute(String sql, Class clazz, String esType, String esIndex) {
|
|
|
List saveModels = new ArrayList<>();
|
|
|
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ssXX");
|
|
|
SimpleDateFormat dateFormat1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
|
|
|
|
|
try {
|
|
|
//解决 group by之后默认是200的问题
|
|
|
if (sql.toLowerCase().contains("group by")) {
|
|
|
sql = sql + " limit 0,2000";
|
|
|
}
|
|
|
|
|
|
SQLExprParser parser = new ElasticSqlExprParser(sql);
|
|
|
SQLExpr expr = parser.expr();
|
|
|
SQLQueryExpr queryExpr = (SQLQueryExpr) expr;
|
|
|
|
|
|
Select select = null;
|
|
|
select = new SqlParser().parseSelect(queryExpr);
|
|
|
|
|
|
//通过抽象语法树，封装成自定义的Select，包含了select、from、where group、limit等
|
|
|
AggregationQueryAction action = null;
|
|
|
DefaultQueryAction queryAction = null;
|
|
|
SqlElasticSearchRequestBuilder requestBuilder = null;
|
|
|
if (select.isAgg) {
|
|
|
//包含计算的的排序分组的
|
|
|
action = new AggregationQueryAction(elasticSearchPool.getClient(), select);
|
|
|
requestBuilder = action.explain();
|
|
|
} else {
|
|
|
//封装成自己的Select对象
|
|
|
queryAction = new DefaultQueryAction(elasticSearchPool.getClient(), select);
|
|
|
requestBuilder = queryAction.explain();
|
|
|
}
|
|
|
SearchResponse response = (SearchResponse) requestBuilder.get();
|
|
|
Object queryResult = null;
|
|
|
if (sql.toUpperCase().indexOf("GROUP") != -1 || sql.toUpperCase().indexOf("SUM") != -1 || select.isAgg) {
|
|
|
queryResult = response.getAggregations();
|
|
|
} else {
|
|
|
queryResult = response.getHits();
|
|
|
}
|
|
|
ObjectResult temp = new ObjectResultsExtractor(true, true, true).extractResults(queryResult, true);
|
|
|
List<String> heads = temp.getHeaders();
|
|
|
temp.getLines().forEach(one -> {
|
|
|
Object saveModel = null;
|
|
|
try {
|
|
|
|
|
|
saveModel = clazz.newInstance();
|
|
|
} catch (Exception e) {
|
|
|
logger.error(e.getMessage());
|
|
|
}
|
|
|
for (int i = 0; i < one.size(); i++) {
|
|
|
try {
|
|
|
String key = null;
|
|
|
Object value = one.get(i);
|
|
|
if (heads.get(i).startsWith("_")) {
|
|
|
if(heads.get(i).contains("_id")){
|
|
|
clazz.getMethod("setId", String.class).invoke(saveModel, value);
|
|
|
}
|
|
|
continue;
|
|
|
}
|
|
|
key = "set" + UpFirstStr(heads.get(i));
|
|
|
if (heads.get(i).contains("quotaDate") || heads.get(i).contains("createTime") || heads.get(i).contains("date_histogram") || heads.get(i).contains("czrq")) {
|
|
|
if (heads.get(i).contains("date_histogram")) {
|
|
|
key = "setQuotaDate";
|
|
|
}
|
|
|
if(heads.get(i).contains("czrq") && one.get(i) == null ){
|
|
|
continue;
|
|
|
}
|
|
|
try {
|
|
|
//yyyy-MM-dd'T'HH:mm:ssXX
|
|
|
value = dateFormat.parse(String.valueOf(one.get(i)));
|
|
|
} catch (Exception e) {
|
|
|
//yyyy-MM-dd HH:mm:ss
|
|
|
try {
|
|
|
value = dateFormat1.parse(String.valueOf(one.get(i)));
|
|
|
}catch (Exception e1){
|
|
|
Timestamp ts = new Timestamp(Long.parseLong(String.valueOf(one.get(i))));
|
|
|
try {
|
|
|
Date date = new Date();
|
|
|
date = ts;
|
|
|
value =date;
|
|
|
|
|
|
} catch (Exception e2) {
|
|
|
value = String.valueOf(one.get(i));
|
|
|
}
|
|
|
}
|
|
|
|
|
|
}
|
|
|
// value = DateUtil.strToDate(String.valueOf(value).replace("T00:00:00+0800", " 00:00:00"), "yyyy-MM-dd HH:mm:ss");
|
|
|
}
|
|
|
|
|
|
if (value instanceof String) {
|
|
|
clazz.getMethod(key, String.class).invoke(saveModel, value);
|
|
|
} else if (value instanceof Integer) {
|
|
|
Field field = clazz.getDeclaredField(heads.get(i));
|
|
|
if(field.getGenericType().getTypeName().equals("java.lang.Long")) {
|
|
|
clazz.getMethod(key, Long.class).invoke(saveModel, Long.valueOf(value + ""));
|
|
|
} else {
|
|
|
clazz.getMethod(key, Integer.class).invoke(saveModel, value);
|
|
|
}
|
|
|
} else if (value instanceof Double) {
|
|
|
clazz.getMethod(key, Double.class).invoke(saveModel, value);
|
|
|
} else if (value instanceof java.util.Date) {
|
|
|
clazz.getMethod(key, java.util.Date.class).invoke(saveModel, value);
|
|
|
} else if (value instanceof java.util.List) {
|
|
|
Field field = clazz.getDeclaredField(heads.get(i));
|
|
|
if(field.getGenericType().getTypeName().equals("java.util.Set<java.lang.String>")){
|
|
|
Set<String> set = new HashSet<>();
|
|
|
for(Object s : (ArrayList) value){
|
|
|
set.add(s + "");
|
|
|
}
|
|
|
clazz.getMethod(key, java.util.Set.class).invoke(saveModel, set);
|
|
|
}else {
|
|
|
clazz.getMethod(key, java.util.List.class).invoke(saveModel, value);
|
|
|
}
|
|
|
}
|
|
|
} catch (Exception e) {
|
|
|
logger.warn(e.getMessage());
|
|
|
}
|
|
|
}
|
|
|
saveModels.add(saveModel);
|
|
|
});
|
|
|
} catch (Exception e) {
|
|
|
logger.error(e.getMessage());
|
|
|
}
|
|
|
return saveModels;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 首字母大写
|
|
|
*
|
|
|
* @param str
|
|
|
* @return
|
|
|
*/
|
|
|
private String UpFirstStr(String str) {
|
|
|
return str.replaceFirst(str.substring(0, 1), str.substring(0, 1).toUpperCase());
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 构造es查询参数
|
|
|
* @param jsonData
|
|
|
*
|
|
|
{
|
|
|
"filter":[{"andOr":"and|or","condition":">|=|<|>=|<=|?","field":"<filed>","value":"<value>"},<{...}>],
|
|
|
- 参数说明：andOr跟数据库的中的AND和OR相似；condition指条件匹配程度，?相当于数据库中的like；filed指检索的字段；value为检索的值
|
|
|
"page":1, - 参数格式：页码，默认1，int类型 不需要分页，传""
|
|
|
"size":10, - 参数格式：条数，默认10，int类型 不需要分页，传""
|
|
|
"sort":[
|
|
|
{"key":{"order":"asc|desc"}}, - 参数格式：排序， key要排序的字段，order固定，取值asc或desc，不需要排序，传""
|
|
|
{"key":{"order":"asc|desc"}}
|
|
|
]
|
|
|
}
|
|
|
* @return
|
|
|
*/
|
|
|
public SearchSourceBuilder getQueryBuilder(String nestedPath, String jsonData,String measureData) {
|
|
|
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
|
|
|
//什么都不传的情况下，构造查询所有的语句
|
|
|
if(StringUtils.isEmpty(jsonData)){
|
|
|
MatchAllQueryBuilder allQueryBuilder = QueryBuilders.matchAllQuery();
|
|
|
return searchSourceBuilder.query(allQueryBuilder);
|
|
|
}
|
|
|
|
|
|
JSONObject json = JSONObject.parseObject(jsonData);
|
|
|
List<Map<String, Object>> filter = (List)json.getJSONArray("filter");
|
|
|
int page = json.getIntValue("page") == 0 ? 1:json.getIntValue("page"); //从第一页开始
|
|
|
int size = json.getIntValue("size") == 0 ? 1:json.getIntValue("size"); //默认值为1，最少获取一条记录
|
|
|
JSONArray sort = json.getJSONArray("sort");
|
|
|
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
|
|
|
|
|
|
boolQueryBuilder = getBoolQueryBuilder(nestedPath,jsonData,false);//非嵌套的数据查询不需要nested
|
|
|
|
|
|
searchSourceBuilder.from((page -1)*size);
|
|
|
searchSourceBuilder.size(size);
|
|
|
//排序
|
|
|
if(!CollectionUtils.isEmpty(sort)){
|
|
|
for(Object obj:sort){
|
|
|
JSONObject object = JSONObject.parseObject(obj.toString());
|
|
|
FieldSortBuilder fieldSortBuilder = null;
|
|
|
for(String key:object.keySet()){
|
|
|
if(!CollectionUtils.isEmpty(fieldsMap.get(nestedPath)) && fieldsMap.get(nestedPath).contains(key)){
|
|
|
fieldSortBuilder = new FieldSortBuilder("measure_data." + key);
|
|
|
}else{
|
|
|
fieldSortBuilder = new FieldSortBuilder(key);
|
|
|
}
|
|
|
JSONObject sortValue = object.getJSONObject(key);
|
|
|
if(StringUtils.equalsIgnoreCase(SortOrder.ASC.toString(),sortValue.getString("order"))){
|
|
|
fieldSortBuilder.order(SortOrder.ASC);
|
|
|
}else if(StringUtils.equalsIgnoreCase(SortOrder.DESC.toString(),sortValue.getString("order"))){
|
|
|
fieldSortBuilder.order(SortOrder.DESC);
|
|
|
}
|
|
|
fieldSortBuilder.setNestedPath("measure_data");
|
|
|
searchSourceBuilder.sort(fieldSortBuilder);
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
if(StringUtils.isNotEmpty(nestedPath)){
|
|
|
NestedQueryBuilder nestedQueryBuilder = getNestedBuilder(nestedPath,jsonData);//嵌套的数据查询
|
|
|
QueryFilterBuilder filterBuilder = QueryBuilders.queryFilter(nestedQueryBuilder);
|
|
|
QueryBuilder filteredQueryBuilder = QueryBuilders.filteredQuery(boolQueryBuilder,filterBuilder);
|
|
|
searchSourceBuilder.query(filteredQueryBuilder);
|
|
|
}else{
|
|
|
searchSourceBuilder.query(boolQueryBuilder);
|
|
|
}
|
|
|
|
|
|
return searchSourceBuilder;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 构造es查询参数
|
|
|
* @param jsonData
|
mengkang