es mapping 数据逻辑处理保存

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

@@ -68,61 +68,65 @@ public class CubeController  extends BaseController {
                envelop.setErrorMsg("索引已存在");
                return envelop;
            }
            // 数据集维度
            List<CubeMappingModel> cubeMappingModels = cubeMappingService.findCubeMappingModels(id);
            for(CubeMappingModel cubeMappingModel : cubeMappingModels){
                FieldInfo fieldInfo =  new FieldInfo();
                String fieldCode = cubeMappingModel.getDimensionCode();
                fieldInfo.setField(fieldCode);
                // 子成员 扩展
                List<CubeMemberMappingModel>  cubeMemberMappingModels = cubeMemberMappingService.findCubeMappingModels(cubeMappingModel.getId());
                if(cubeMemberMappingModels != null){
                    //如果存在子集设置 子集属性
                    if(cubeMappingModel.getChildSaveType() != null){
                        if(cubeMappingModel.getChildSaveType().equals(1)){
                            fieldInfo.setDataType("object");
                        }else if(cubeMappingModel.getChildSaveType().equals(2)){
                            fieldInfo.setDataType("nested");
            List<CubeMappingModel> cubeMappingModels = cubeMappingService.findCubeMappingModelsByCubeId(id);
            if(cubeMappingModels != null && cubeMappingModels.size() > 0){
                for(CubeMappingModel cubeMappingModel : cubeMappingModels){
                    FieldInfo fieldInfo =  new FieldInfo();
                    String fieldCode = cubeMappingModel.getDimensionCode();
                    fieldInfo.setField(fieldCode);
                    // 子成员 扩展
                    List<CubeMemberMappingModel>  cubeMemberMappingModels = cubeMemberMappingService.findCubeMemberMappingModelsById(cubeMappingModel.getId());
                    if(cubeMemberMappingModels != null && cubeMemberMappingModels.size() > 0){
                        //如果存在子集设置 子集属性
                        if(cubeMappingModel.getChildSaveType() != null){
                            if(cubeMappingModel.getChildSaveType().equals(1)){
                                fieldInfo.setDataType("object");
                            }else if(cubeMappingModel.getChildSaveType().equals(2)){
                                fieldInfo.setDataType("nested");
                            }
                        }
                    }
                    List<FieldInfo> childFieldiList = new ArrayList<>();
                    for(CubeMemberMappingModel cubeMemberMappingModel : cubeMemberMappingModels){
                        FieldInfo childFieldInfo =  new FieldInfo();
                        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.setDataType("String");
                            childFieldiList.add(dictFieldCode);
                            childFieldiList.add(dictFieldName);
                        List<FieldInfo> childFieldiList = new ArrayList<>();
                        for(CubeMemberMappingModel cubeMemberMappingModel : cubeMemberMappingModels){
                            FieldInfo childFieldInfo =  new FieldInfo();
                            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.setDataType("String");
                                childFieldiList.add(dictFieldCode);
                                childFieldiList.add(dictFieldName);
                            }
                        }
                        fieldInfo.setFieldInfos(childFieldiList);
                    }else {
                        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);
                    }
                    fieldInfo.setFieldInfos(childFieldiList);
                }else {
                    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);
            }else {
                envelop.setSuccessFlg(false);
                envelop.setErrorMsg("没有维度信息");
            }
            esHandler.createIndexAndCreateMapping(index, type, fieldInfoList, esHandler.getTransportClient());
            envelop.setSuccessFlg(true);
            return envelop;
        } catch (Exception e) {
            e.printStackTrace();

src/main/java/com/yihu/quota/kafka/ConsumerListener.java

@@ -11,13 +11,13 @@ import org.springframework.kafka.annotation.KafkaListener;
 */
public class ConsumerListener {
    @Autowired
    private ElasticSearchDataProcessService elasticSearchMappingService;
    private ElasticSearchDataProcessService elasticSearchDataProcessService;
    @KafkaListener(topics = "sep-hbase-data")
    public void loadData(ConsumerRecord<?, ?> record) {
        System.out.println("kafka data: " + record.key() + " - " + record.value());
        if(record.value() != null){
            elasticSearchMappingService.saveData(record.value().toString());
            elasticSearchDataProcessService.saveData(record.value().toString());
        }
    }

src/main/java/com/yihu/quota/model/dimension/Dimension.java

@@ -33,6 +33,11 @@ public class Dimension {
     */
    private String algorithm;
    /**
     * 中间算法参数
     */
    private String parm;
    /**
     * 数据类型
     */
@@ -94,6 +99,7 @@ public class Dimension {
        this.algorithm = algorithm;
    }
    @Column(name = "data_type")
    public String getDataType() {
        return dataType;
    }
@@ -102,6 +108,14 @@ public class Dimension {
        this.dataType = dataType;
    }
    public String getParm() {
        return parm;
    }
    public void setParm(String parm) {
        this.parm = parm;
    }
    @Transient
    public List<DimensionMember> getDimensionMembers() {
        return dimensionMembers;

src/main/java/com/yihu/quota/model/dimension/DimensionMember.java

@@ -40,6 +40,16 @@ public class DimensionMember {
     */
    private String dataType;
    /**
     * 中间算法
     */
    private String algorithm;
    /**
     * 中间算法参数
     */
    private String parm;
    /**
     * 数据字典
     */
@@ -89,10 +99,15 @@ public class DimensionMember {
        this.level = level;
    }
    @Column(name = "data_type")
    public String getDataType() {
        return dataType;
    }
    public void setDataType(String dataType) {
        this.dataType = dataType;
    }
    public String getDict() {
        return dict;
    }
@@ -101,8 +116,22 @@ public class DimensionMember {
        this.dict = dict;
    }
    public void setDataType(String dataType) {
        this.dataType = dataType;
    public String getAlgorithm() {
        return algorithm;
    }
    public void setAlgorithm(String algorithm) {
        this.algorithm = algorithm;
    }
    public String getParm() {
        return parm;
    }
    public void setParm(String parm) {
        this.parm = parm;
    }
    @Formula("(SELECT odm.name FROM olap_dimension_member odm LEFT JOIN olap_dimension od ON od.id = odm.dimension_id WHERE odm.id  = id )")

src/main/java/com/yihu/quota/model/source/DataSourcesTable.java

@@ -20,6 +20,12 @@ public class DataSourcesTable {
     * 表名
     */
    private String tableName;
    /**
     * 表编码
     */
    private String tableCode;
    /**
     * 备注
     */
@@ -62,6 +68,15 @@ public class DataSourcesTable {
        this.tableName = tableName;
    }
    @Column(name = "table_code")
    public String getTableCode() {
        return tableCode;
    }
    public void setTableCode(String tableCode) {
        this.tableCode = tableCode;
    }
    public String getNote() {
        return note;
    }

src/main/java/com/yihu/quota/model/source/DataSourcesTableField.java

@@ -21,6 +21,11 @@ public class DataSourcesTableField {
     * 字段名
     */
    private String fieldName;
    /**
     * 字段编码
     */
    private String fieldCode;
    /**
     * 字段类型
     */
@@ -64,6 +69,16 @@ public class DataSourcesTableField {
        this.fieldName = fieldName;
    }
    @Column(name = "field_code")
    public String getFieldCode() {
        return fieldCode;
    }
    public void setFieldCode(String fieldCode) {
        this.fieldCode = fieldCode;
    }
    @Column(name = "field_type")
    public String getFieldType() {
        return fieldType;

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

@@ -41,7 +41,12 @@ public class CubeMappingService extends BaseJpaService<CubeMapping, CubeMappingD
    }
    public List<CubeMappingModel> findCubeMappingModels(int cubeId) {
    /**
     * 根据数据集 查询对应的维度信息
     * @param cubeId
     * @return
     */
    public List<CubeMappingModel> findCubeMappingModelsByCubeId(int cubeId) {
        String sql = "SELECT ocm.*,od.data_type,od.dict from olap_cube_mapping ocm ,olap_dimension od " +
                " where ocm.dimension_id = od.id AND ocm.cube_id = ? order by ocm.id asc";
        List<CubeMappingModel> cubeMappingModels = jdbcTemplate.query(sql, new BeanPropertyRowMapper(CubeMappingModel.class), cubeId);
@@ -49,4 +54,20 @@ public class CubeMappingService extends BaseJpaService<CubeMapping, CubeMappingD
    }
    /**
     * 根据表编码和字段编码 查询对应的维度信息
     * @param fieldCode
     * @param tableCode
     * @return
     */
    public List<CubeMappingModel> findCubeMappingModelsByFieldCode(String tableCode,String fieldCode) {
        String sql = "SELECT ocm.*,odstf.field_code,od.dict,od.data_type,od.`algorithm`,od.parm from olap_cube_mapping ocm  " +
                " 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};
        List<CubeMappingModel> cubeMappingModels = jdbcTemplate.query(sql, new BeanPropertyRowMapper(CubeMappingModel.class), param);
        return cubeMappingModels;
    }
}

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

@@ -37,10 +37,32 @@ public class CubeMemberMappingService extends BaseJpaService<CubeMemberMapping,
        cubeMemberMappingDao.delete(id);
    }
    public List<CubeMemberMappingModel> findCubeMappingModels(Integer id) {
        String sql = "SELECT ocmm.*,odm.data_type from olap_cube_member_mapping ocmm ,olap_dimension_member odm" +
                " where ocmm.dimension_member_id = odm.id  AND ocmm.cube_mapping_id = ?  order by ocmm.id asc";
    /**
     * 根据数据集维度ID 查询对应的维度成员信息
     * @param id
     * @return
     */
    public List<CubeMemberMappingModel> findCubeMemberMappingModelsById(Integer id) {
        String sql = "SELECT ocmm.*,odm.data_type from olap_cube_member_mapping ocmm ,olap_dimension_member odm " +
                " where ocmm.dimension_member_id = odm.id  AND ocmm.cube_mapping_id = ?  order by ocmm.id asc ";
        List<CubeMemberMappingModel> cubeMemberMappingModels = jdbcTemplate.query(sql, new BeanPropertyRowMapper(CubeMemberMappingModel.class), id);
        return cubeMemberMappingModels;
    }
    /**
     * 根据表编码和字段编码 查询对应的维度信息
     * @param fieldCode
     * @return
     */
    public List<CubeMemberMappingModel> findCubeMemberMappingModels(String tableCode,String fieldCode) {
        String sql = "SELECT ocmm.*,ocm.dimension_code as parentCode,ocm.child_save_type,odstf.field_code,odm.dict,odm.data_type,odm.`algorithm`,odm.parm from olap_cube_member_mapping ocmm " +
                " LEFT JOIN olap_data_sources_table_field odstf ON ocmm.data_field_id = odstf.id " +
                " LEFT JOIN olap_data_sources_table odst ON odst.id = odstf.table_id " +
                " 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};
        List<CubeMemberMappingModel> cubeMemberMappingModels = jdbcTemplate.query(sql, new BeanPropertyRowMapper(CubeMemberMappingModel.class), param);
        return cubeMemberMappingModels;
    }
}

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

@@ -4,14 +4,23 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
import com.yihu.ehr.elasticsearch.ElasticSearchPool;
import com.yihu.ehr.elasticsearch.ElasticSearchUtil;
import com.yihu.ehr.util.datetime.DateUtil;
import com.yihu.quota.etl.formula.DictFunc;
import com.yihu.quota.etl.formula.DivisionFunc;
import com.yihu.quota.util.ElasticSearchHandler;
import com.yihu.quota.vo.CubeMappingModel;
import com.yihu.quota.vo.CubeMemberMappingModel;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.text.NumberFormat;
import java.text.ParseException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
@@ -30,8 +39,6 @@ public class ElasticSearchDataProcessService {
    @Autowired
    private CubeMappingService cubeMappingService;
    @Autowired
    private CubeService cubeService;
    @Autowired
    private CubeMemberMappingService cubeMemberMappingService;
    @Autowired
    private ElasticSearchPool elasticSearchPool;
@@ -74,7 +81,6 @@ public class ElasticSearchDataProcessService {
        //通过表找到 对应的数据集 保存的索引和type
        //TODO 可以维护到数据字典 - 保存到redis 减少去数据库里面查询
        String rowKey = dataMap.get("rowKey").toString();
        String action = dataMap.get("action").toString();
        dataMap.remove("table");
@@ -82,10 +88,93 @@ public class ElasticSearchDataProcessService {
        dataMap.remove("action");
        try {
            if(action.contains(action_put)){
                //保存数据
                //TODO 子属性数据 更新 ，需要将库中数据查询出来，然后合并在保存到库中
                //TODO 扩展的属性值 更新  如 字典 子成员  更加数据类型转换
                String keyValue = "";
                for(String key : dataMap.keySet()){
                    if(dataMap.get(key)!= null){
                         keyValue = dataMap.get(key).toString();
                    }
                    //根据列名 查找出 对应的维度code及是否要数据字典，是否通过算法扩展出来
                    // 是否是子集模式中
                    List<CubeMappingModel> cubeMappingModels = cubeMappingService.findCubeMappingModelsByFieldCode(table, key);
                    if(cubeMappingModels != null && cubeMappingModels.size() > 0){
                        for(CubeMappingModel cubeMappingModel :cubeMappingModels){
                            String cloumnCode = cubeMappingModel.getDimensionCode();
                            //字典扩展
                            if(StringUtils.isNotEmpty(cubeMappingModel.getDict())){
                                DictFunc dictFunc = new DictFunc();
                                String param[] = {cubeMappingModel.getDict(),cloumnCode};
                                String value = dictFunc.execute(param);
                                String dictCode = cloumnCode + ".Code";
                                String dictName = cloumnCode + ".Name";
                                source.put(dictCode,key);
                                source.put(dictName,value);
                            }else if(StringUtils.isNotEmpty(cubeMappingModel.getAlgorithm())){
                                //具体维度 对应具体算法
                            }else {
                                if(StringUtils.isNotEmpty(cubeMappingModel.getDataType())){
                                    String dataType = cubeMappingModel.getDataType();
                                    Object value = dataConver(dataType,keyValue);
                                    source.put(cloumnCode,value);
                                }else{
                                    source.put(cloumnCode,dataMap.get(key));
                                }
                            }
                        }
                    }
                    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 ){//对象方式
                                }
                                if(childSaveType == 2 ){//nested 方式
                                    //查出历史数据 然后组合保存
                                }
                            }else {
                            }
                            //字典扩展
                            if(StringUtils.isNotEmpty(cubeMemberMappingModel.getDict())){
                                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);
                            }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 townParam[] = {dataMap.get(key).toString(),"1"};
                                        String townVal = divisionFunc.execute(townParam);
                                        String cityParam[] = {dataMap.get(key).toString(),"2"};
                                        String cityVal = divisionFunc.execute(cityParam);
                                        String provinceParam[] = {dataMap.get(key).toString(),"3"};
                                        String provinceVal = divisionFunc.execute(provinceParam);
                                        source.put("",townVal);
                                        source.put("",cityVal);
                                        source.put("",provinceVal);
                                    }
                                }
                            }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));
                                }
                            }
                        }
                    }
                }
                elasticSearchUtil.index(index, type, dataMap);
            }else if (action.contains(action_del)){
                for(String key : dataMap.keySet()){
@@ -101,6 +190,9 @@ public class ElasticSearchDataProcessService {
            logger.debug("elasticSearch 执行失败");
            e.printStackTrace();
            e.getMessage();
        } catch (Exception e) {
            logger.debug("数据解析异常");
            e.printStackTrace();
        }
    }
@@ -119,4 +211,31 @@ public class ElasticSearchDataProcessService {
        return source;
    }
    /**
     * 数据类型转换
     * @param dataType
     * @param keyValue
     */
    public Object dataConver(String dataType,String keyValue){
        NumberFormat nf = NumberFormat.getInstance();
        Object value = null;
        dataType = dataType.toLowerCase();
        if(dataType.equals("string")){
            value = keyValue;
        }else if(dataType.equals("int")){
            int intValue = Integer.valueOf(keyValue);
            value = intValue;
        }else if(dataType.equals("double")){
            nf.setGroupingUsed(false);
            nf.setMaximumFractionDigits(2);
            double doubleValue = Double.valueOf(keyValue);
            value = doubleValue;
        }else if(dataType.equals("date")){
            Date dateValue = DateUtil.formatCharDateYMDHMS(keyValue);
            value = dateValue;
        }
        return value;
    }
}

src/main/java/com/yihu/quota/vo/CubeMappingModel.java

@@ -11,10 +11,13 @@ public class CubeMappingModel {
    private Integer dimensionId;      //维度ID
    private String dimensionCode;    //维度编码
    private String fieldName;        //字段名称
    private String dataType;        //数据类型
    private String dict;            //数据字典
    private String algorithm;       // 中间算法
    private String parm;        //中间算法参数
    private String dataType;    //数据类型
    private Integer childSaveType; //子集映射格式类型 1 对象 2 子集
    private String childPrimaryKay; //子集唯一字段
    private String fieldCode;        //字段编码
    public Integer getId() {
        return id;
@@ -95,4 +98,28 @@ public class CubeMappingModel {
    public void setChildSaveType(Integer childSaveType) {
        this.childSaveType = childSaveType;
    }
    public String getFieldCode() {
        return fieldCode;
    }
    public void setFieldCode(String fieldCode) {
        this.fieldCode = fieldCode;
    }
    public String getAlgorithm() {
        return algorithm;
    }
    public void setAlgorithm(String algorithm) {
        this.algorithm = algorithm;
    }
    public String getParm() {
        return parm;
    }
    public void setParm(String parm) {
        this.parm = parm;
    }
}

src/main/java/com/yihu/quota/vo/CubeMemberMappingModel.java

@@ -11,8 +11,13 @@ public class CubeMemberMappingModel {
    private Integer dimensionMemberId;//维度成员ID
    private String dimensionCode;    //维度编码
    private String fieldName;        //字段名称
    private String algorithm;       // 中间算法
    private String parm;            //中间算法参数
    private String dataType;        //数据类型
    private String dict;            //数据字典
    private String fieldCode;      //字段编码
    private String parentCode;      //父级属性code
    private Integer childSaveType;
    public Integer getId() {
        return id;
@@ -78,4 +83,43 @@ public class CubeMemberMappingModel {
        this.dict = dict;
    }
    public String getFieldCode() {
        return fieldCode;
    }
    public void setFieldCode(String fieldCode) {
        this.fieldCode = fieldCode;
    }
    public String getAlgorithm() {
        return algorithm;
    }
    public void setAlgorithm(String algorithm) {
        this.algorithm = algorithm;
    }
    public String getParm() {
        return parm;
    }
    public void setParm(String parm) {
        this.parm = parm;
    }
    public String getParentCode() {
        return parentCode;
    }
    public void setParentCode(String parentCode) {
        this.parentCode = parentCode;
    }
    public Integer getChildSaveType() {
        return childSaveType;
    }
    public void setChildSaveType(Integer childSaveType) {
        this.childSaveType = childSaveType;
    }
}