|
@ -29,13 +29,14 @@ public class ElasticSearchDataProcessService {
|
|
|
private static Logger logger = LoggerFactory.getLogger(ElasticSearchDataProcessService.class);
|
|
|
private static String dataSource_hbase = "hbase";
|
|
|
private static String dataSource_mysql = "mysql";
|
|
|
private static String action_del = "DeleteFamily";//删除整行
|
|
|
private static String action_put = "Put"; //添加和修改单个字段值
|
|
|
private static String action_delFamily = "DeleteFamily";//删除整行
|
|
|
|
|
|
private static String dataSource_k = "dataSource";
|
|
|
private static String table_k = "table";
|
|
|
private static String id_k = "_id";
|
|
|
private static String rowKey_k = "rowkey";
|
|
|
private static String profileId_k = "profileId";
|
|
|
private static String subRowKey_k = "subRowkey";
|
|
|
private static String action_k = "action";
|
|
|
|
|
@ -46,8 +47,6 @@ public class ElasticSearchDataProcessService {
|
|
|
@Autowired
|
|
|
private CubeMemberMappingService cubeMemberMappingService;
|
|
|
@Autowired
|
|
|
private ElasticSearchPool elasticSearchPool;
|
|
|
@Autowired
|
|
|
private ElasticSearchUtil elasticSearchUtil;
|
|
|
|
|
|
/**
|
|
@ -75,152 +74,78 @@ public class ElasticSearchDataProcessService {
|
|
|
|
|
|
/**
|
|
|
* @param dataMap
|
|
|
* //如果是子集的数据 hbase 过来要指定父级数据
|
|
|
*/
|
|
|
public void hbaseDataProcess(Map<String, Object> dataMap){
|
|
|
public void hbaseDataProcess(Map<String, Object> dataMap) throws Exception{
|
|
|
Map<String, Object> source = new HashMap<>();
|
|
|
String index = "patient_event";
|
|
|
String type = "info";
|
|
|
String table = dataMap.get(table_k).toString();
|
|
|
//通过表找到 对应的数据集 保存的索引和type
|
|
|
//TODO 可以维护到数据字典 - 保存到redis 减少去数据库里面查询
|
|
|
String table = "";
|
|
|
String rowKey = "";
|
|
|
String profileId = "";
|
|
|
String action = "";
|
|
|
if(dataMap.containsKey(table_k)){
|
|
|
table = dataMap.get(table_k).toString();
|
|
|
}
|
|
|
if(dataMap.containsKey(rowKey_k)){
|
|
|
rowKey = dataMap.get(rowKey_k).toString();
|
|
|
source.put(id_k,rowKey);
|
|
|
source.put(rowKey_k,rowKey);
|
|
|
}
|
|
|
if(dataMap.containsKey(profileId_k)){
|
|
|
profileId = dataMap.get(profileId_k).toString();
|
|
|
}
|
|
|
if(dataMap.containsKey(action_k)){
|
|
|
action = dataMap.get(action_k).toString();
|
|
|
}
|
|
|
|
|
|
|
|
|
//如果是子集的数据 hbase 过来要指定父级数据
|
|
|
|
|
|
String rowKey = dataMap.get(rowKey_k).toString();
|
|
|
String action = dataMap.get(action_k).toString();
|
|
|
dataMap.remove(table_k);
|
|
|
dataMap.remove(rowKey_k);
|
|
|
dataMap.remove(action_k);
|
|
|
try {
|
|
|
if(action.contains(action_put)){
|
|
|
String keyValue = "";
|
|
|
for(String key : dataMap.keySet()){
|
|
|
if(dataMap.get(key)!= null){
|
|
|
keyValue = dataMap.get(key).toString();
|
|
|
}
|
|
|
source.put(id_k,rowKey);
|
|
|
source.put(rowKey_k,rowKey);
|
|
|
//根据列名 查找出 对应的维度code及是否要数据字典,是否通过算法扩展出来
|
|
|
List<CubeMappingModel> cubeMappingModels = cubeMappingService.findCubeMappingModelsByFieldCode(table, key);
|
|
|
if(cubeMappingModels != null && cubeMappingModels.size() > 0){
|
|
|
for(CubeMappingModel cubeMappingModel :cubeMappingModels){
|
|
|
dataMap.remove(table_k);
|
|
|
dataMap.remove(rowKey_k);
|
|
|
dataMap.remove(action_k);
|
|
|
String keyValue = "";
|
|
|
for(String hbaseColCode : dataMap.keySet()){
|
|
|
if(dataMap.get(hbaseColCode)!= null){
|
|
|
keyValue = dataMap.get(hbaseColCode).toString();
|
|
|
}
|
|
|
List<CubeMappingModel> cubeMappingModels = cubeMappingService.findCubeMappingModelsByFieldCode(table, hbaseColCode);
|
|
|
if(cubeMappingModels != null && cubeMappingModels.size() > 0){
|
|
|
for(CubeMappingModel cubeMappingModel :cubeMappingModels){
|
|
|
String index = cubeMappingModel.getIndexName();
|
|
|
String type = cubeMappingModel.getIndexType();
|
|
|
if(action.contains(action_put)){
|
|
|
String cloumnCode = cubeMappingModel.getDimensionCode();
|
|
|
String dataType = cubeMappingModel.getDataType();
|
|
|
String dict = cubeMappingModel.getDict();
|
|
|
String algorithm = cubeMappingModel.getAlgorithm();
|
|
|
String algorithmParm = cubeMappingModel.getParm();
|
|
|
Map<String, Object> esDataMap = dataToMap(keyValue,cloumnCode,dataType ,dict,algorithm,algorithmParm);
|
|
|
//维度数据扩展保存
|
|
|
Map<String, Object> esDataMap = dimensionDataExtendToMap(keyValue, cloumnCode, dataType, dict, algorithm, algorithmParm);
|
|
|
source.putAll(esDataMap);
|
|
|
saveElasticSearchData(index, type,rowKey,source);
|
|
|
}else if(action.contains(action_del)){
|
|
|
elasticSearchUtil.delete(index,type,rowKey);
|
|
|
}
|
|
|
}
|
|
|
// 是否是 子集属性
|
|
|
List<CubeMemberMappingModel> cubeMemberMappingModels = cubeMemberMappingService.findCubeMemberMappingModels(table,key);
|
|
|
if(cubeMemberMappingModels != null && cubeMemberMappingModels.size() > 0){
|
|
|
for(CubeMemberMappingModel cubeMemberMappingModel :cubeMemberMappingModels){
|
|
|
String cloumnCode = cubeMemberMappingModel.getDimensionCode();
|
|
|
String parentCode = cubeMemberMappingModel.getParentCode();
|
|
|
if(cubeMemberMappingModel.getChildSaveType() != null){
|
|
|
int childSaveType = cubeMemberMappingModel.getChildSaveType();
|
|
|
if(childSaveType == 1 ){//对象方式
|
|
|
Map<String, Object> objChildMap = new HashMap<>();
|
|
|
objChildMap.put(cloumnCode,keyValue);
|
|
|
//字典扩展
|
|
|
if(StringUtils.isNotEmpty(cubeMemberMappingModel.getDict()) && StringUtils.isEmpty(cubeMemberMappingModel.getAlgorithm())){
|
|
|
String value = "";
|
|
|
String param[] = {cubeMemberMappingModel.getDict(),keyValue};
|
|
|
// value = dictFunc.execute(param);
|
|
|
value = key + "测试值";
|
|
|
source.put(cloumnCode,keyValue);
|
|
|
objChildMap.put(cloumnCode + "Name",value);
|
|
|
|
|
|
}else if(StringUtils.isNotEmpty(cubeMemberMappingModel.getAlgorithm())){
|
|
|
if(cubeMemberMappingModel.getAlgorithm().equals("DivisionFunc") && StringUtils.isNotEmpty(cubeMemberMappingModel.getParm())){
|
|
|
if(cubeMemberMappingModel.getParm().equals(key)){
|
|
|
DivisionFunc divisionFunc = new DivisionFunc();
|
|
|
String level = "1";
|
|
|
String divisionVal = "";
|
|
|
if(cloumnCode.toLowerCase().equals("town")){
|
|
|
level = "1";
|
|
|
divisionVal = "信州区";
|
|
|
}else if(cloumnCode.toLowerCase().equals("city")){
|
|
|
level = "2";
|
|
|
divisionVal = "上饶市";
|
|
|
}else if(cloumnCode.toLowerCase().equals("province")){
|
|
|
level = "3";
|
|
|
divisionVal = "江西省";
|
|
|
}
|
|
|
String divisionParam[] = {dataMap.get(key).toString(),level};
|
|
|
// divisionVal = divisionFunc.execute(divisionParam);
|
|
|
objChildMap.put(cloumnCode,divisionVal);
|
|
|
}
|
|
|
}
|
|
|
//其他算法 --
|
|
|
}
|
|
|
source.put(parentCode,objChildMap);
|
|
|
saveElasticSearchData(index, type,rowKey,source);
|
|
|
}else if(childSaveType == 2 ){//nested 方式
|
|
|
List<Map<String,Object>> nestedList = new ArrayList<>();
|
|
|
//查出历史数据 然后组合保存
|
|
|
String field = parentCode + "." + subRowKey_k;
|
|
|
List<Map<String, Object>> subDataList = elasticSearchUtil.findByField(index, type, field, rowKey);
|
|
|
if(subDataList != null && subDataList.size() > 0){
|
|
|
String parentRowkey = subDataList.get(0).get(rowKey_k).toString();
|
|
|
List<Map<String, Object>> dataList = elasticSearchUtil.findByField(index, type, rowKey_k, parentRowkey);
|
|
|
if(dataList != null && dataList.size() > 0){
|
|
|
//组装 子集历史数据,更改当前字段值 在添加
|
|
|
for(Map<String, Object> map : dataList){
|
|
|
if(map.get(subRowKey_k).equals(rowKey)){
|
|
|
for(String colKey :map.keySet()){
|
|
|
if(colKey.equals(cloumnCode)){
|
|
|
map.put(key,keyValue);
|
|
|
}
|
|
|
}
|
|
|
nestedList.add(map);
|
|
|
}else{
|
|
|
nestedList.add(map);
|
|
|
}
|
|
|
}
|
|
|
}else {
|
|
|
//库中没有记录,单条添加
|
|
|
Map<String, Object> nestedChildMap = new HashMap<>();
|
|
|
nestedChildMap.put(cloumnCode,keyValue);
|
|
|
nestedChildMap.put(subRowKey_k,rowKey);
|
|
|
nestedList.add(nestedChildMap);
|
|
|
}
|
|
|
}
|
|
|
source.put(parentCode,nestedList);
|
|
|
saveElasticSearchData(index, type,rowKey,source);
|
|
|
}
|
|
|
}else {
|
|
|
//字典扩展
|
|
|
if(StringUtils.isNotEmpty(cubeMemberMappingModel.getDict()) && StringUtils.isEmpty(cubeMemberMappingModel.getAlgorithm())){
|
|
|
String value = "";
|
|
|
String param[] = {cubeMemberMappingModel.getDict(),keyValue};
|
|
|
// value = dictFunc.execute(param);
|
|
|
value = cloumnCode + "测试值";
|
|
|
source.put(cloumnCode,keyValue);
|
|
|
source.put(cloumnCode + "Name",value);
|
|
|
}else if(StringUtils.isNotEmpty(cubeMemberMappingModel.getAlgorithm())){
|
|
|
|
|
|
}else {
|
|
|
if(StringUtils.isNotEmpty(cubeMemberMappingModel.getDataType())){
|
|
|
String dataType = cubeMemberMappingModel.getDataType();
|
|
|
Object value = dataConver(dataType,keyValue);
|
|
|
source.put(cloumnCode,value);
|
|
|
}else{
|
|
|
source.put(cloumnCode,dataMap.get(key));
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
saveElasticSearchData(index, type,rowKey,source);
|
|
|
}
|
|
|
// 是否是子集属性
|
|
|
List<CubeMemberMappingModel> cubeMemberMappingModels = cubeMemberMappingService.findCubeMemberMappingModels(table,hbaseColCode);
|
|
|
if(cubeMemberMappingModels != null && cubeMemberMappingModels.size() > 0){
|
|
|
for(CubeMemberMappingModel cubeMemberMappingModel :cubeMemberMappingModels){
|
|
|
String index = cubeMemberMappingModel.getIndexName();
|
|
|
String type = cubeMemberMappingModel.getIndexType();
|
|
|
if(action.contains(action_put)){
|
|
|
//维度成员数据扩展保存
|
|
|
source = dimensionMemberDataExtendToMap(cubeMemberMappingModel,hbaseColCode,keyValue,rowKey);
|
|
|
saveElasticSearchData(index, type,profileId,source);
|
|
|
}else if(action.contains(action_del)){
|
|
|
elasticSearchUtil.delete(index,type,profileId);
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
}else if(action.contains(action_delFamily)){
|
|
|
elasticSearchUtil.delete(index,type,rowKey);
|
|
|
}
|
|
|
|
|
|
|
|
|
}catch (ParseException e){
|
|
|
logger.debug("elasticSearch 执行失败");
|
|
|
e.printStackTrace();
|
|
@ -231,52 +156,29 @@ public class ElasticSearchDataProcessService {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
public Map<String,Object> dataToMap(String sourceValue,String cloumnCode,String dataType ,String dict,String algorithm,String algorithmParm ){
|
|
|
/**
|
|
|
* 维度数据扩展 转map
|
|
|
* @param sourceValue
|
|
|
* @param cloumnCode
|
|
|
* @param dataType
|
|
|
* @param dict
|
|
|
* @param algorithm
|
|
|
* @param algorithmParm
|
|
|
* @return
|
|
|
*/
|
|
|
public Map<String,Object> dimensionDataExtendToMap(String sourceValue,String cloumnCode,String dataType ,String dict,String algorithm,String algorithmParm ){
|
|
|
Map<String, Object> source = new HashMap<>();
|
|
|
DictFunc dictFunc = new DictFunc();
|
|
|
//字典扩展
|
|
|
if(StringUtils.isNotEmpty(dict) && StringUtils.isEmpty(algorithm)){
|
|
|
String value = "";
|
|
|
String param[] = {dict,sourceValue};
|
|
|
// value = dictFunc.execute(param);
|
|
|
value = cloumnCode + "测试值";
|
|
|
source.put(cloumnCode,sourceValue);
|
|
|
source.put(cloumnCode + "Name",value);
|
|
|
source = extendDictData(source,cloumnCode,dict,sourceValue);
|
|
|
}else if(StringUtils.isNotEmpty(algorithm)){
|
|
|
//计算后 又经过字典
|
|
|
// 如:年龄段
|
|
|
if(algorithm.equals("AgeGroupFunc") && StringUtils.isNotEmpty(algorithmParm)){
|
|
|
String value = "";
|
|
|
AgeGroupFunc ageGroupFunc = new AgeGroupFunc();
|
|
|
String ageGroup = ageGroupFunc.execute(Integer.valueOf(sourceValue));
|
|
|
String param[] = {dict,ageGroup};
|
|
|
// value = dictFunc.execute(param);
|
|
|
value = "年龄段=" + ageGroup;
|
|
|
if(StringUtils.isNotEmpty(dict)){
|
|
|
source.put(cloumnCode,ageGroup);
|
|
|
source.put(cloumnCode + "Name",value);
|
|
|
}
|
|
|
//年龄段
|
|
|
if(algorithm.equals("AgeGroupFunc")){
|
|
|
source = extendAgeGroupData(source,cloumnCode,dict,sourceValue);
|
|
|
}
|
|
|
//区域算法
|
|
|
if(algorithm.equals("DivisionFunc") && StringUtils.isNotEmpty(algorithmParm)){
|
|
|
if(algorithmParm.equals(cloumnCode)){
|
|
|
DivisionFunc divisionFunc = new DivisionFunc();
|
|
|
String level = "1";
|
|
|
String divisionVal = "";
|
|
|
if(cloumnCode.toLowerCase().equals("town")){
|
|
|
level = "1";
|
|
|
divisionVal = "信州区";
|
|
|
}else if(cloumnCode.toLowerCase().equals("city")){
|
|
|
level = "2";
|
|
|
divisionVal = "上饶市";
|
|
|
}else if(cloumnCode.toLowerCase().equals("province")){
|
|
|
level = "3";
|
|
|
divisionVal = "江西省";
|
|
|
}
|
|
|
String divisionParam[] = {sourceValue,level};
|
|
|
// divisionVal = divisionFunc.execute(divisionParam);
|
|
|
source.put(cloumnCode,divisionVal);
|
|
|
}
|
|
|
source = extendDivisionData(source, cloumnCode, algorithmParm, sourceValue);
|
|
|
}
|
|
|
//其他算法 --
|
|
|
}else {
|
|
@ -290,6 +192,156 @@ public class ElasticSearchDataProcessService {
|
|
|
return source;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 维度成员数据扩展 转map
|
|
|
* @param cubeMemberMappingModel
|
|
|
* @param key
|
|
|
* @param sourceValue
|
|
|
* @param rowKey
|
|
|
* @return
|
|
|
*/
|
|
|
public Map<String,Object> dimensionMemberDataExtendToMap(CubeMemberMappingModel cubeMemberMappingModel ,String key,String sourceValue,String rowKey) {
|
|
|
Map<String, Object> source = new HashMap<>();
|
|
|
String cloumnCode = cubeMemberMappingModel.getDimensionCode();
|
|
|
String parentCode = cubeMemberMappingModel.getParentCode();
|
|
|
String dataType = cubeMemberMappingModel.getDataType();
|
|
|
String dict = cubeMemberMappingModel.getDict();
|
|
|
String algorithm = cubeMemberMappingModel.getAlgorithm();
|
|
|
String algorithmParm = cubeMemberMappingModel.getParm();
|
|
|
int childSaveType = cubeMemberMappingModel.getChildSaveType();
|
|
|
String index = cubeMemberMappingModel.getIndexName();
|
|
|
String type = cubeMemberMappingModel.getIndexType();
|
|
|
|
|
|
if(childSaveType == 1 ){//对象方式
|
|
|
Map<String, Object> objChildMap = new HashMap<>();
|
|
|
objChildMap.put(cloumnCode,sourceValue);
|
|
|
if(StringUtils.isNotEmpty(dict)){
|
|
|
objChildMap = extendDictData(objChildMap,cloumnCode,dict,sourceValue);
|
|
|
source.put(parentCode,objChildMap);
|
|
|
}else if(StringUtils.isNotEmpty(algorithm)){
|
|
|
//年龄段
|
|
|
if(algorithm.equals("AgeGroupFunc")){
|
|
|
objChildMap = extendAgeGroupData(objChildMap,cloumnCode,dict,sourceValue);
|
|
|
}
|
|
|
//区域算法
|
|
|
if(algorithm.equals("DivisionFunc") && StringUtils.isNotEmpty(algorithmParm)){
|
|
|
objChildMap = extendDivisionData(objChildMap, cloumnCode, algorithmParm, sourceValue);
|
|
|
}
|
|
|
source.put(parentCode,objChildMap);
|
|
|
//其他算法 --
|
|
|
}
|
|
|
source.put(parentCode,objChildMap);
|
|
|
}else if(childSaveType == 2 ){//nested 方式
|
|
|
List<Map<String,Object>> nestedList = new ArrayList<>();
|
|
|
//查出历史数据 然后组合保存
|
|
|
String field = parentCode + "." + subRowKey_k;
|
|
|
List<Map<String, Object>> subDataList = elasticSearchUtil.findByField(index, type, field, rowKey);
|
|
|
if(subDataList != null && subDataList.size() > 0){
|
|
|
String parentRowkey = subDataList.get(0).get(rowKey_k).toString();
|
|
|
List<Map<String, Object>> dataList = elasticSearchUtil.findByField(index, type, rowKey_k, parentRowkey);
|
|
|
if(dataList != null && dataList.size() > 0){
|
|
|
//组装 子集历史数据,更改当前字段值 在添加
|
|
|
for(Map<String, Object> map : dataList){
|
|
|
if(map.get(subRowKey_k).equals(rowKey)){
|
|
|
for(String colKey :map.keySet()){
|
|
|
if(colKey.equals(cloumnCode)){
|
|
|
map.put(key,sourceValue);
|
|
|
}
|
|
|
}
|
|
|
nestedList.add(map);
|
|
|
}else{
|
|
|
nestedList.add(map);
|
|
|
}
|
|
|
}
|
|
|
}else {
|
|
|
//库中没有记录,单条添加
|
|
|
Map<String, Object> nestedChildMap = new HashMap<>();
|
|
|
nestedChildMap.put(cloumnCode,sourceValue);
|
|
|
nestedChildMap.put(subRowKey_k,rowKey);
|
|
|
nestedList.add(nestedChildMap);
|
|
|
}
|
|
|
}else{
|
|
|
Map<String, Object> map = new HashMap<>();
|
|
|
map.put("subRowkey",rowKey);
|
|
|
map.put(cloumnCode,sourceValue);
|
|
|
nestedList.add(map);
|
|
|
}
|
|
|
source.put(parentCode,nestedList);
|
|
|
}
|
|
|
return source;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 字典数据扩展
|
|
|
* @param source 数据集合
|
|
|
* @param cloumnCode 列code
|
|
|
* @param dict 字典ID
|
|
|
* @param code 字典编码
|
|
|
* @return
|
|
|
*/
|
|
|
public Map<String, Object> extendDictData(Map<String, Object> source ,String cloumnCode,String dict,String code){
|
|
|
DictFunc dictFunc = new DictFunc();
|
|
|
String value = "";
|
|
|
String param[] = {dict,code};
|
|
|
// value = dictFunc.execute(param);
|
|
|
value = cloumnCode + "测试值";
|
|
|
source.put(cloumnCode,code);
|
|
|
source.put(cloumnCode + "Name",value);
|
|
|
return source;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 年龄扩展年龄段 及年龄段字典
|
|
|
* @param source 数据集合
|
|
|
* @param cloumnCode 列code
|
|
|
* @param dict 字典ID
|
|
|
* @param sourceValue 年龄
|
|
|
* @return
|
|
|
*/
|
|
|
public Map<String, Object> extendAgeGroupData(Map<String, Object> source ,String cloumnCode,String dict,String sourceValue){
|
|
|
String value = "";
|
|
|
DictFunc dictFunc = new DictFunc();
|
|
|
AgeGroupFunc ageGroupFunc = new AgeGroupFunc();
|
|
|
String ageGroup = ageGroupFunc.execute(Integer.valueOf(sourceValue));
|
|
|
String param[] = {dict,ageGroup};
|
|
|
// value = dictFunc.execute(param);
|
|
|
value = "年龄段=" + ageGroup;
|
|
|
if(StringUtils.isNotEmpty(dict)){
|
|
|
source.put(cloumnCode,ageGroup);
|
|
|
source.put(cloumnCode + "Name",value);
|
|
|
}
|
|
|
return source;
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 区域扩展
|
|
|
* @param source 数据集合
|
|
|
* @param cloumnCode 列code
|
|
|
* @param algorithmParm 算法参数
|
|
|
* @param sourceValue 年龄
|
|
|
* @return
|
|
|
*/
|
|
|
public Map<String, Object> extendDivisionData(Map<String, Object> source ,String cloumnCode,String algorithmParm,String sourceValue){
|
|
|
DivisionFunc divisionFunc = new DivisionFunc();
|
|
|
String level = "1";
|
|
|
String divisionVal = "";
|
|
|
if(cloumnCode.toLowerCase().equals("town")){
|
|
|
level = "1";
|
|
|
divisionVal = "信州区";
|
|
|
}else if(cloumnCode.toLowerCase().equals("city")){
|
|
|
level = "2";
|
|
|
divisionVal = "上饶市";
|
|
|
}else if(cloumnCode.toLowerCase().equals("province")){
|
|
|
level = "3";
|
|
|
divisionVal = "江西省";
|
|
|
}
|
|
|
String divisionParam[] = {sourceValue,level};
|
|
|
// divisionVal = divisionFunc.execute(divisionParam);
|
|
|
source.put(cloumnCode,divisionVal);
|
|
|
return source;
|
|
|
}
|
|
|
|
|
|
|
|
|
/**
|
|
|
* 添加修改 数据
|
|
|
* @param index
|