简单实用的分布式大数据处理框架，特点是零基础操作，支持批处理和流式处理

package org.eocencle.magnet.jsonbuilder.builder; import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import org.eocencle.magnet.core.exception.UnsupportedException; import org.eocencle.magnet.core.mapping.*; import org.eocencle.magnet.core.util.CoreTag; import org.eocencle.magnet.jsonbuilder.session.JsonProjectConfig; import org.eocencle.magnet.jsonbuilder.util.JSONBuilderTag; import java.util.Iterator; /** * 工作流建构类 * @author: huan * @Date: 2020-05-31 * @Description: */ public class WorkFlowBuilder implements JSONParser { // 单例实体 private static WorkFlowBuilder BUILDER = new WorkFlowBuilder(); private WorkFlowBuilder() { } /** * 获取单例实体 * @Author huan * @Date 2020-05-31 * @Param [] * @Return org.eocencle.magnet.jsonbuilder.builder.WorkFlowBuilder * @Exception * @Description */ public static WorkFlowBuilder getInstance() { return BUILDER; } @Override public void parse(Object parser, JsonProjectConfig config) { JSONObject jsonObejct = (JSONObject) parser; JSONArray paramArray = jsonObejct.getJSONArray(JSONBuilderTag.JSON_ATTR_WORKFLOW); Iterator<Object> iterator = paramArray.iterator(); JSONObject jsonObj = null; String type = null; while (iterator.hasNext()) { jsonObj = (JSONObject) iterator.next(); type = jsonObj.getString(JSONBuilderTag.JSON_ATTR_CTYPE); if (JSONBuilderTag.WORKFLOW_TYPE_SQL.equalsIgnoreCase(type)) { this.parseSQLElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_OUTPUT.equalsIgnoreCase(type)) { this.parseOutputElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_GROUP.equalsIgnoreCase(type)) { this.parseGroupElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_FILTER.equalsIgnoreCase(type)) { this.parseFilterElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_DISTINCT.equalsIgnoreCase(type)) { this.parseDistinctElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_ORDER.equalsIgnoreCase(type)) { this.parseOrderElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_UNION.equalsIgnoreCase(type)) { this.parseUnionElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_JOIN.equalsIgnoreCase(type)) { this.parseJoinElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_SCHEMA.equalsIgnoreCase(type)) { this.parseSchemaElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_VALUEMAPPERS.equalsIgnoreCase(type)) { this.parseValueMappersElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_SPLITFIELDTOROWS.equalsIgnoreCase(type)) { this.parseSplitFieldToRowsElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_STRINGCUTS.equalsIgnoreCase(type)) { this.parseStringCutsElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_ADDFIELDS.equalsIgnoreCase(type)) { this.parseAddFieldsElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_ADDSEQUENCE.equalsIgnoreCase(type)) { this.parseAddSequenceElements(jsonObj, config); } else if (JSONBuilderTag.WORKFLOW_TYPE_ROWNUM.equalsIgnoreCase(type)) { this.parseRowNumElements(jsonObj, config); } else { throw new UnsupportedException(type + " component is not supported"); } } } /** * 解析SQL元素 * @Author huan * @Date 2020-06-13 * @Param [jsonObj, config] * @Return void * @Exception * @Description */ private void parseSQLElements(JSONObject jsonObj, JsonProjectConfig config) { SQLInfo sqlInfo = new SQLInfo(); sqlInfo.setId(jsonObj.getString(JSONBuilderTag.JSON_ATTR_ID)); sqlInfo.setAlias(jsonObj.getString(JSONBuilderTag.JSON_ATTR_ALIAS)); sqlInfo.setSql(jsonObj.getString(JSONBuilderTag.JSON_ATTR_SQL)); config.putWorkFlowInfo(sqlInfo.getId(), sqlInfo); } /** * 解析输出元素 * @Author huan * @Date 2020-06-13 * @Param [jsonObj, config] * @Return void * @Exception * @Description */ private void parseOutputElements(JSONObject jsonObj, JsonProjectConfig config) { OutputInfo outputInfo = new OutputInfo(); outputInfo.setId(jsonObj.getString(JSONBuilderTag.JSON_ATTR_ID)); outputInfo.setRef(jsonObj.getString(JSONBuilderTag.JSON_ATTR_REF)); String style = jsonObj.getString(JSONBuilderTag.JSON_ATTR_STYLE); if (null == style) { outputInfo.setStyle(JSONBuilderTag.OUTPUT_STYLE_FILE); } else { outputInfo.setStyle(style); } outputInfo.setTarget(jsonObj.getString(JSONBuilderTag.JSON_ATTR_TARGET)); String type = jsonObj.getString(JSONBuilderTag.JSON_ATTR_TYPE); if (null == type) { outputInfo.setType(JSONBuilderTag.OUTPUT_TYPE_CREATE); } else { outputInfo.setType(type); } String compress = jsonObj.getString(JSONBuilderTag.JSON_ATTR_COMPRESS); if (null == compress) { outputInfo.setCompress(JSONBuilderTag.COMPRESS_NONE); } else { outputInfo.setCompress(compress); } String separator = jsonObj.getString(JSONBuilderTag.JSON_ATTR_SEPARATOR); if (null == separator) { outputInfo.setSeparator(JSONBuilderTag.SPLIT_INVISIBLE1); } else { outputInfo.setSeparator(separator); } config.putWorkFlowInfo(outputInfo.getId(), outputInfo); } /** * 解析分组元素 * @Author huan * @Date 2020-06-13 * @Param [jsonObj, config] * @Return void * @Exception * @Description */ private void parseGroupElements(JSONObject jsonObj, JsonProjectConfig config) { GroupInfo groupInfo = new GroupInfo(); groupInfo.setId(jsonObj.getString(JSONBuilderTag.JSON_ATTR_ID)); groupInfo.setAlias(jsonObj.getString(JSONBuilderTag.JSON_ATTR_ALIAS)); groupInfo.setRef(jsonObj.getString(JSONBuilderTag.JSON_ATTR_REF)); String field = jsonObj.getString(JSONBuilderTag.JSON_ATTR_FIELD); String[] fields = field.trim().split(JSONBuilderTag.SPLIT_COMMA); for (String val: fields) { groupInfo.addGroupField(new GroupInfo.GroupField(val.trim())); } String[] orders = jsonObj.getString(JSONBuilderTag.JSON_ATTR_ORDER).split(CoreTag.SPLIT_COMMA); String[] key = null; for (String order: orders) { key = order.trim().split(CoreTag.SPLIT_BLANK); if (2 == key.length) { groupInfo.addOrderField(new GroupInfo.OrderField(key[0].trim(), key[1].trim())); } else { groupInfo.addOrderField(new GroupInfo.OrderField(key[0].trim())); } } groupInfo.setRownumField(jsonObj.getString(JSONBuilderTag.JSON_ATTR_ROWNUM)); groupInfo.setStreamState(jsonObj.getString(JSONBuilderTag.JSON_ATTR_STREAM_STATE)); config.putWorkFlowInfo(groupInfo.getId(), groupInfo); } /** * 解析过滤元素 * @Author huan * @Date 2020-06-13 * @Param [jsonObj, config] * @Return void * @Exception * @Description */ private void parseFilterElements(JSONObject jsonObj, JsonProjectConfig config) { FilterInfo filterInfo = new FilterInfo(); filterInfo.setId(jsonObj.getString(JSONBuilderTag.JSON_ATTR_ID)); filterInfo.setAlias(jsonObj.getString(JSONBuild