es mapping 数据逻辑处理保存

src/main/java/com/yihu/quota/controller/CubeController.java

@@ -82,8 +82,16 @@ public class CubeController  extends BaseController {
                rowKeyFieldInfo.setDataType("String");
                fieldInfoList.add(rowKeyFieldInfo);
                for(CubeMappingModel cubeMappingModel : cubeMappingModels){
                    FieldInfo fieldInfo =  new FieldInfo();
                    String fieldCode = cubeMappingModel.getDimensionCode();
                    //数据字典名称作为一个属性 扩展
                    if(StringUtils.isNotEmpty(cubeMappingModel.getDict())){
                        FieldInfo dictFieldName =  new FieldInfo();
                        dictFieldName.setField(fieldCode + "Name");
                        dictFieldName.setDataType("String");
                        fieldInfoList.add(dictFieldName);
                    }
                    FieldInfo fieldInfo =  new FieldInfo();
                    fieldInfo.setField(fieldCode);
                    // 子成员 扩展
                    List<CubeMemberMappingModel>  cubeMemberMappingModels = cubeMemberMappingService.findCubeMemberMappingModelsById(cubeMappingModel.getId());
@@ -106,15 +114,11 @@ public class CubeController  extends BaseController {
                            childFieldInfo.setField(cubeMemberMappingModel.getDimensionCode());
                            childFieldInfo.setDataType(cubeMemberMappingModel.getDataType());
                            childFieldiList.add(childFieldInfo);
                            //数据字典名称作为一个子属性 扩展
                            if(StringUtils.isNotEmpty(cubeMemberMappingModel.getDict())){
                                //子集 的 数据字典 扩展
                                FieldInfo dictFieldCode =  new FieldInfo();
                                dictFieldCode.setField(fieldCode + "Name");
                                dictFieldCode.setDataType("String");
                                FieldInfo dictFieldName =  new FieldInfo();
                                dictFieldName.setField(fieldCode + "Code");
                                dictFieldName.setField(fieldCode + "Name");
                                dictFieldName.setDataType("String");
                                childFieldiList.add(dictFieldCode);
                                childFieldiList.add(dictFieldName);
                            }
                        }
@@ -123,17 +127,6 @@ public class CubeController  extends BaseController {
                        fieldInfo.setDataType(cubeMappingModel.getDataType());
                    }
                    fieldInfoList.add(fieldInfo);
                    if(StringUtils.isNotEmpty(cubeMappingModel.getDict())){
                        //数据字典 扩展
                        FieldInfo dictFieldCode =  new FieldInfo();
                        dictFieldCode.setField(fieldCode + "Name");
                        dictFieldCode.setDataType("String");
                        FieldInfo dictFieldName =  new FieldInfo();
                        dictFieldName.setField(fieldCode + "Code");
                        dictFieldName.setDataType("String");
                        fieldInfoList.add(dictFieldCode);
                        fieldInfoList.add(dictFieldName);
                    }
                }
                esHandler.createIndexAndCreateMapping(index, type, fieldInfoList, esHandler.getTransportClient());
                envelop.setSuccessFlg(true);
@@ -153,7 +146,7 @@ public class CubeController  extends BaseController {
    @ApiOperation(value = "测试保存mapping 数据 ")
    @RequestMapping(value = "/saveElasticSearchMappingData", method = RequestMethod.GET)
    public Envelop saveElasticSearchMappingData(
            @ApiParam(name = "table", value = "表", required = true)
            @ApiParam(name = "table", value = "表 Healtharchive / HealtharchiveSub", required = true)
            @RequestParam(value = "table", required = true) String table,
            @ApiParam(name = "rowkey", value = "行数据唯一键值", required = true)
            @RequestParam(value = "rowkey", required = true) String rowkey,
@@ -162,7 +155,7 @@ public class CubeController  extends BaseController {
            @ApiParam(name = "cloumnValue", value = "列值", required = true)
            @RequestParam(value = "cloumnValue", required = true) Object cloumnValue,
            @ApiParam(name = "action", value = "执行动作 1:put 添加修改数据 2:del 删除单个字段值 3:delAll 删除整行", required = true)
            @RequestParam(value = "action", required = true) Object action
            @RequestParam(value = "action", required = true) int action
    ){
        Envelop envelop = new Envelop();
@@ -173,7 +166,15 @@ public class CubeController  extends BaseController {
            source.put("dataSource","hbase");
            source.put("table",table);
            source.put("rowkey",rowkey);
            source.put("action",action);
            String actionStr = "Put";
            if(action == 1){
                actionStr = "Put";
            }else if(action == 2){
                actionStr = "DeleteColumn";
            }else if(action == 3){
                actionStr = "DeleteFamily";
            }
            source.put("action",actionStr);
            source.put(cloumn,cloumnValue);
            String jsonData = gson.toJson(source);
            elasticSearchDataProcessService.saveData(jsonData);

src/main/java/com/yihu/quota/service/cube/CubeMappingService.java

@@ -65,7 +65,7 @@ public class CubeMappingService extends BaseJpaService<CubeMapping, CubeMappingD
                " LEFT JOIN olap_data_sources_table_field odstf ON ocm.data_field_id = odstf.id "+
                " LEFT JOIN olap_data_sources_table odst ON odst.id = odstf.table_id " +
                " LEFT JOIN olap_dimension od ON ocm.dimension_id = od.id where odstf.field_code = ? and odst.table_code = ? ";
        String param[] = {tableCode,fieldCode};
        String param[] = {fieldCode,tableCode};
        List<CubeMappingModel> cubeMappingModels = jdbcTemplate.query(sql, new BeanPropertyRowMapper(CubeMappingModel.class), param);
        return cubeMappingModels;
    }

src/main/java/com/yihu/quota/service/cube/CubeMemberMappingService.java

@@ -61,7 +61,7 @@ public class CubeMemberMappingService extends BaseJpaService<CubeMemberMapping,
                " LEFT JOIN olap_dimension_member odm ON ocmm.dimension_member_id = odm.id  " +
                " LEFT JOIN olap_cube_mapping ocm ON ocm.id = ocmm.cube_mapping_id " +
                " where odstf.field_code = ? and odst.table_code = ? ";
        String param[] = {tableCode,fieldCode};
        String param[] = {fieldCode,tableCode};
        List<CubeMemberMappingModel> cubeMemberMappingModels = jdbcTemplate.query(sql, new BeanPropertyRowMapper(CubeMemberMappingModel.class), param);
        return cubeMemberMappingModels;
    }

src/main/java/com/yihu/quota/service/cube/ElasticSearchDataProcessService.java

@@ -35,6 +35,7 @@ public class ElasticSearchDataProcessService {
    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 subRowKey_k = "subRowkey";
    private static String action_k = "action";
@@ -78,8 +79,8 @@ public class ElasticSearchDataProcessService {
     */
    public void hbaseDataProcess(Map<String, Object> dataMap){
        Map<String, Object> source  = new HashMap<>();
        String index = "";
        String type = "";
        String index = "patient_event";
        String type = "info";
        String table = dataMap.get(table_k).toString();
        //通过表找到 对应的数据集 保存的索引和type
        //TODO 可以维护到数据字典 - 保存到redis 减少去数据库里面查询
@@ -96,6 +97,8 @@ public class ElasticSearchDataProcessService {
                    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){
@@ -104,24 +107,24 @@ public class ElasticSearchDataProcessService {
                            DictFunc dictFunc = new DictFunc();
                            //字典扩展
                            if(StringUtils.isNotEmpty(cubeMappingModel.getDict()) && StringUtils.isEmpty(cubeMappingModel.getAlgorithm())){
                                String value = "";
                                String param[] = {cubeMappingModel.getDict(),keyValue};
                                String value = dictFunc.execute(param);
                                String dictCode = cloumnCode + ".Code";
                                String dictName = cloumnCode + ".Name";
                                source.put(dictCode,key);
                                source.put(dictName,value);
//                                value = dictFunc.execute(param);
                                value =  key + "测试值";
                                source.put(cloumnCode,keyValue);
                                source.put(cloumnCode + "Name",value);
                            }else if(StringUtils.isNotEmpty(cubeMappingModel.getAlgorithm())){
                                //计算后 又经过字典 如:年龄段
                                if(cubeMappingModel.getAlgorithm().equals("AgeGroupFunc") && StringUtils.isNotEmpty(cubeMappingModel.getParm())){
                                    String value = "";
                                    AgeGroupFunc ageGroupFunc = new AgeGroupFunc();
                                    String ageGroup = ageGroupFunc.execute(keyValue);
                                    String ageGroup = ageGroupFunc.execute(Integer.valueOf(keyValue));
                                    String param[] = {cubeMappingModel.getDict(),ageGroup};
                                    String value = dictFunc.execute(param);
//                                    value = dictFunc.execute(param);
                                    value = "年龄段=" + ageGroup;
                                    if(StringUtils.isNotEmpty(cubeMappingModel.getDict())){
                                        String dictCode = cloumnCode + ".Code";
                                        String dictName = cloumnCode + ".Name";
                                        source.put(dictCode,key);
                                        source.put(dictName,value);
                                        source.put(cloumnCode,ageGroup);
                                        source.put(cloumnCode + "Name",value);
                                    }
                                }
                                //其他具体维度 对应具体算法
@@ -184,13 +187,13 @@ public class ElasticSearchDataProcessService {
                            }else {
                                //字典扩展
                                if(StringUtils.isNotEmpty(cubeMemberMappingModel.getDict())){
                                    String value = "";
                                    DictFunc dictFunc = new DictFunc();
                                    String param[] = {cubeMemberMappingModel.getDict(),cloumnCode};
                                    String value = dictFunc.execute(param);
                                    String dictCode = cloumnCode + ".Code";
                                    String dictName = cloumnCode + ".Name";
                                    source.put(dictCode,key);
                                    source.put(dictName,value);
//                                    value = dictFunc.execute(param);
                                    value =  key + "测试值";
                                    source.put(cloumnCode + "Name",value);
                                    source.put(cloumnCode,keyValue);
                                }else if(StringUtils.isNotEmpty(cubeMemberMappingModel.getAlgorithm())){
                                    if(cubeMemberMappingModel.getAlgorithm().equals("DivisionFunc") && StringUtils.isNotEmpty(cubeMemberMappingModel.getParm())){
                                        if(cubeMemberMappingModel.getParm().equals(key)){
@@ -220,13 +223,13 @@ public class ElasticSearchDataProcessService {
                    }
                }
                elasticSearchUtil.index(index, type, dataMap);
                elasticSearchUtil.update(index, type,rowKey, source);
            }else if (action.contains(action_del)){
                for(String key : dataMap.keySet()){
                    dataMap.put(key,"");
                }
                //保存数据
               elasticSearchUtil.index(index, type, dataMap);
               elasticSearchUtil.update(index, type, rowKey,dataMap);
            }else if(action.contains(action_delFamily)){
                elasticSearchUtil.delete(index,type,rowKey);
            }

src/main/java/com/yihu/quota/util/ElasticSearchHandler.java

@@ -52,8 +52,10 @@ public class ElasticSearchHandler {
            Settings settings = Settings.builder().put("cluster.name", clusterName)
        /*            .put("client.transport.sniff", true)*/
                    .put("client.transport.ping_timeout", "30s").build();
            String host = ip.split(";")[0];
            host = host.substring(0,host.indexOf(":"));
            client = TransportClient.builder().settings(settings).build()
                    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(ip), 9300));
                    .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host), 9300));
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }