feat: 添加企业集团优化

- 添加相关工具类
- 添加生产环节debug
- 计算逻辑的优化调整
- 添加子集团识别（待深入测试、优化）
- 添加机关单位识别（待深入测试、优化）

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/EnterpriseGroupAggregator.java

@@ -37,7 +37,7 @@ public class EnterpriseGroupAggregator extends Aggregator<EnterpriseGroupAggValu
 
     @Override
     public boolean terminate(WorkerContext context, EnterpriseGroupAggValue enterpriseGroupAggValue) throws IOException {
-        System.out.println("step: " + context.getSuperstep() + ", 本轮迭代节点数：" + enterpriseGroupAggValue.getCount() + " ,flag=" + enterpriseGroupAggValue.getFlag().get());
+        System.out.println("step: " + context.getSuperstep() + ", iterations：" + enterpriseGroupAggValue.getCount() + " ,flag=" + enterpriseGroupAggValue.getFlag().get());
         return !enterpriseGroupAggValue.getFlag().get();
     }
 }

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/EnterpriseGroupJob.java

@@ -4,6 +4,8 @@ import com.aliyun.odps.data.TableInfo;
 import com.aliyun.odps.graph.GraphJob;
 import com.aliyun.odps.graph.RemoveDuplicatesLoadingResolver;
 import com.winhc.max.compute.graph.job.pagerank.CompanyComputingVertexResolver;
+import com.winhc.max.compute.graph.util.BaseUtils;
+import com.winhc.max.compute.graph.util.DateUtils;
 import com.winhc.max.compute.graph.util.ParameterTool;
 
 import java.io.IOException;
@@ -14,26 +16,54 @@ import java.io.IOException;
  */
 public class EnterpriseGroupJob {
 
+
+    private static final String vertexTableName;
+    private static final String edgeTableName;
+    private static final String outputTable;
+
+    static {
+        if (BaseUtils.isWindows()) {
+            vertexTableName = "tmp_xjk_test_enterprise_group_input_vertex";
+            edgeTableName = "tmp_xjk_test_enterprise_group_input_edge";
+            outputTable = "tmp_xjk_test_enterprise_group_out";
+        } else {
+            vertexTableName = "tmp_xjk_enterprise_group_input_vertex";
+            edgeTableName = "tmp_xjk_enterprise_group_input_edge";
+            outputTable = "tmp_xjk_enterprise_group_out";
+        }
+    }
+
+
     public static void main(String[] args) throws IOException {
+        String nowDate = DateUtils.nowDate(DateUtils.YYYYMMDDHHMMSS).replace(" ", ".").replace(":", "");
+
         ParameterTool parameterTool = ParameterTool.fromArgs(args);
         String numWorkers = parameterTool.getOrDefault("numWorkers", "70");
         String workerCPU = parameterTool.getOrDefault("workerCPU", "2");
         String workerMem = parameterTool.getOrDefault("workerMem", "8192");
-
+        String outPart = parameterTool.getOrDefault("outPart", nowDate);
+        String debugVertexId = parameterTool.getOrDefault("debugVertexId", null);
         System.out.println("input args: " + String.join(" ", args));
         GraphJob job = new GraphJob();
 
 
-
         job.setMaxIteration(-1);
         job.setNumWorkers(Integer.parseInt(numWorkers));
-        if(parameterTool.has("workerCPU")){
+        if (parameterTool.has("workerCPU")) {
             job.setWorkerCPU(Integer.parseInt(workerCPU) * 100);
         }
-        if(parameterTool.has("workerMem")){
+        if (parameterTool.has("workerMem")) {
             job.setWorkerMemory(Integer.parseInt(workerMem));
         }
 
+        if (parameterTool.has("enableDebug")) {
+           String debugIndex =  parameterTool.getOrDefault("debugIndex", null);
+            job.set("winhc.debug", "true");
+            job.set("winhc.debug.index", debugIndex);
+            job.set("winhc.debug.debugVertexId", debugVertexId);
+        } else {
+            job.set("winhc.debug", "false");
+        }
 
 
         job.setGraphLoaderClass(EnterpriseGroupReader.class);
@@ -47,15 +77,16 @@ public class EnterpriseGroupJob {
 
         job.addInput(TableInfo.builder()
                 .projectName("winhc_ng")
-                .tableName("tmp_xjk_enterprise_group_input_vertex")
+                .tableName(vertexTableName)
                 .build());
         job.addInput(TableInfo.builder()
                 .projectName("winhc_ng")
-                .tableName("tmp_xjk_enterprise_group_input_edge")
+                .tableName(edgeTableName)
                 .build());
         job.addOutput(TableInfo.builder()
                 .projectName("winhc_ng")
-                .tableName("tmp_xjk_enterprise_group_out")
+                .tableName(outputTable)
+                .partSpec("ds='" + outPart + "'")
                 .build());
 
 

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/EnterpriseGroupReader.java

@@ -15,6 +15,7 @@ import com.winhc.max.compute.graph.job.enterprise_group.entity.EntGroupMsg;
 import com.winhc.max.compute.graph.job.enterprise_group.entity.EnterpriseGroupVertexValue;
 import com.winhc.max.compute.graph.job.enterprise_group.entity.HolderEdge;
 import com.winhc.max.compute.graph.util.JsonUtils;
+import com.winhc.max.compute.graph.util.BaseUtils;
 import com.winhc.max.compute.graph.util.WritableRecordExtensions;
 import lombok.experimental.ExtensionMethod;
 
@@ -34,6 +35,7 @@ import java.util.stream.Collectors;
         EnterpriseGroupUtils.class
         , JsonUtils.class
         , WritableRecordExtensions.class
+        , BaseUtils.class
 })
 public class EnterpriseGroupReader extends
         GraphLoader<Text, EnterpriseGroupVertexValue, HolderEdge, EntGroupMsg> {
@@ -55,24 +57,36 @@ public class EnterpriseGroupReader extends
             Text companyId = record.getTextOrNull("company_id");
             Text holderKeyno = record.getTextOrNull("holder_keyno");
             Text holderName = record.getTextOrNull("holder_name");
-            DoubleWritable investmentProportion = record.getOrNull("investment_proportion", DoubleWritable.class);
+            LongWritable holderType = record.getOrNull("holder_type", LongWritable.class);
+            DoubleWritable investmentProportion = record.getOrNullInstance("investment_proportion", DoubleWritable.class);
             if (investmentProportion == null) {
                 return;
             }
-            HolderEdge edgeValue = HolderEdge.of(holderKeyno.toString(), investmentProportion.get());
+            if (!EnterpriseGroupUtils.isHolder(holderKeyno.toString(), holderName.toString())) {
+                return;
+            }
 
-            Edge<Text, HolderEdge> edge = new Edge<Text, HolderEdge>(
-                    holderKeyno, edgeValue);
-            context.addEdgeRequest(companyId, edge);
+
+            int holderTypeVal = 1;
+            if (holderKeyno.toString().length() == 33) {
+                holderTypeVal = 3;
+            } else if (holderType != null && holderType.get() == 5) {
+                holderTypeVal = 2;
+            }
 
             if (holderKeyno.toString().length() == 33) {
                 //添加人的顶点
                 EnterpriseGroupVertex enterpriseGroupVertex = new EnterpriseGroupVertex();
                 enterpriseGroupVertex.setId(holderKeyno);
-                enterpriseGroupVertex.setValue(EnterpriseGroupVertexValue.of(holderName.toString()));
+                enterpriseGroupVertex.setValue(EnterpriseGroupVertexValue.of(holderName.toString(), 6, null, null));
                 context.addVertexRequest(enterpriseGroupVertex);
             }
 
+            HolderEdge edgeValue = HolderEdge.of(holderKeyno.toString(), holderTypeVal, investmentProportion.get());
+
+            Edge<Text, HolderEdge> edge = new Edge<Text, HolderEdge>(
+                    holderKeyno, edgeValue);
+            context.addEdgeRequest(companyId, edge);
         } else {
             //加载点，个人的顶点也要算在内
             Text companyId = record.getTextOrNull("company_id");
@@ -82,20 +96,20 @@ public class EnterpriseGroupReader extends
 
 
             boolean partnership = false;
+            boolean branch = false;
 
             List<String> companyOrgTypeNewList = gson.parseListStr(companyOrgTypeNew.toString());
-            if (companyOrgTypeNewList != null)
-                for (String tmpText : companyOrgTypeNewList) {
-                    boolean flag = tmpText.contains("合伙");
-                    if (flag) {
-                        partnership = true;
-                        break;
-                    }
-                }
+            if (companyOrgTypeNewList != null) {
+                partnership = companyOrgTypeNewList.containsStr("合伙");
+                branch = companyName.toString().contains("分公司");
+            }
+
 
             EnterpriseGroupVertex enterpriseGroupVertex = new EnterpriseGroupVertex();
             enterpriseGroupVertex.setId(companyId);
-            if (partnership) {
+            if (branch) {
+                enterpriseGroupVertex.setValue(EnterpriseGroupVertexValue.of(companyName.toString(), 2, null, null));
+            } else if (partnership) {
                 List<Map<String, String>> le = gson.parseListStrByMap(legalEntity.toString());
 
                 List<String> list = le.stream().map(e -> {

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/EnterpriseGroupUtils.java

@@ -8,6 +8,8 @@ import com.winhc.max.compute.graph.job.enterprise_group.entity.HolderEdge;
 import com.winhc.max.compute.graph.util.WinhcTuple;
 
 import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -199,7 +201,25 @@ public class EnterpriseGroupUtils {
         return stringStream.collect(Collectors.joining(","));
     }
 
+
+    private static final Pattern holderPart = Pattern.compile(".+条件.*股.*");
+
+    public static boolean isHolder(String holderKeyno, String holderName) {
+        if (holderName == null || holderKeyno == null) {
+            return false;
+        }
+        if (holderKeyno.length() == 33) {
+            return true;
+        }
+        Matcher matcher = holderPart.matcher(holderName);
+        if(matcher.matches()){
+            return false;
+        }
+        return true;
+    }
+
+
     public static void main(String[] args) {
-        System.out.println(stockChainTrim("b", "(a)<-[1]-(b),(c)<-[1]-(b),(d)<-[1]-(b),(c)<-[1]-(d)", 3));
+        System.out.println(isHolder("","无限售条件外资股（H股）"));
     }
 }

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/EnterpriseGroupVertex.java

@@ -4,10 +4,7 @@ import com.aliyun.odps.graph.ComputeContext;
 import com.aliyun.odps.graph.Edge;
 import com.aliyun.odps.graph.Vertex;
 import com.aliyun.odps.graph.WorkerContext;
-import com.aliyun.odps.io.BooleanWritable;
-import com.aliyun.odps.io.LongWritable;
-import com.aliyun.odps.io.NullWritable;
-import com.aliyun.odps.io.Text;
+import com.aliyun.odps.io.*;
 import com.aliyun.odps.utils.StringUtils;
 import com.google.gson.Gson;
 import com.winhc.max.compute.graph.job.enterprise_group.entity.EntGroupMsg;
@@ -34,12 +31,15 @@ public class EnterpriseGroupVertex extends
         Vertex<Text, EnterpriseGroupVertexValue, HolderEdge, EntGroupMsg> {
 
     private static Gson gson;
+    private static ProdEnvDebug prodEnvDebug;
 
     @Override
     public void setup(WorkerContext<Text, EnterpriseGroupVertexValue, HolderEdge, EntGroupMsg> context) throws IOException {
         gson = new Gson();
+        prodEnvDebug = ProdEnvDebug.build(context.getConfiguration());
     }
 
+
     @Override
     public void compute(ComputeContext<Text, EnterpriseGroupVertexValue, HolderEdge, EntGroupMsg> context, Iterable<EntGroupMsg> messages) throws IOException {
         context.aggregate(NullWritable.get());
@@ -61,6 +61,7 @@ public class EnterpriseGroupVertex extends
             //获取下游目标节点
             List<Edge<Text, HolderEdge>> edges = getEdges();
             majorityShareholder = edges.getMajorityShareholder(list);
+
             thisVertexHoldKeyno = new ArrayList<>(majorityShareholder.keySet());
             thisVertexHoldCompanyId = thisVertexHoldKeyno.stream().filter(e -> e.length() == 32).collect(Collectors.toList());
             thisVertexOtherHolderKeyno = edges.getOtherHolder(thisVertexHoldKeyno);
@@ -74,13 +75,26 @@ public class EnterpriseGroupVertex extends
             if (!collect.isEmpty()) {
                 thisVertexHoldCompanyId = thisVertexHoldCompanyId.stream().filter(e -> !collect.contains(e)).collect(Collectors.toList());
             }
+
+            //todo:如果该节点需要拆分子节点，则不向下传递
+            if (getValue().getManualStop().get()) {
+                thisVertexHoldCompanyId.clear();
+                throw new RuntimeException("vId: " + getId() + "\n" + getValue().toString());
+            }
         }
 
+
+        //todo:debug
+        prodEnvDebug.debug_1(getId()
+                , "thisVertexHoldCompanyId: {}\nmajorityShareholder: {}"
+                , thisVertexHoldCompanyId, majorityShareholder
+        );
+
         //将本节点传递到下一节点
         if (!getValue().isSendMsg()) {
             for (String destVertexId : thisVertexHoldCompanyId) {
                 //传递控股企业
-                EntGroupMsg entGroupMsg = EntGroupMsg.ofByType_1(getId(), getValue().getCompanyName(), thisVertexOtherHolderCompanyId);
+                EntGroupMsg entGroupMsg = EntGroupMsg.ofByType_1(getId(), getValue().getCompanyName(), getValue().getCompanyType(), thisVertexOtherHolderCompanyId);
                 boolean flag = entGroupMsg.routeLog(destVertexId);
                 if (!flag) {
 //                    getValue().setOutput();
@@ -154,14 +168,22 @@ public class EnterpriseGroupVertex extends
         List<String> holdCompanyId = vertexEntity.getHoldCompanyIds().tuple2List();
         List<String> groupInvestmentCompanyId = vertexEntity.getGroupInvestmentCompanyIds().tuple2List();
         List<String> groupHolderCompanyId = vertexEntity.getGroupHolderCompanyIds().tuple2List();
+        List<String> subgroup = vertexEntity.getSubgroup().tuple2List();
         List<String> stockRightControlChain = Arrays.stream(vertexEntity.getStockRightControlChain().toString().split(",")).collect(Collectors.toList());
 
+        if (subgroup == null) {
+            subgroup = new ArrayList<>();
+        }
+
         for (EntGroupMsg message : messages) {
             holdNum += message.getHoldNum().get();
             holdCompanyId.addAll(message.getHoldCompanyIds().tuple2List());
             groupHolderCompanyId.addAll(message.getGroupHolderCompanyIds().tuple2List());
             groupInvestmentCompanyId.addAll(message.getGroupInvestmentCompanyIds().tuple2List());
             stockRightControlChain.add(message.getStockRightControlChain().toString());
+            if (message.getSourceVertexId() != null && StringUtils.isNotBlank(message.getSourceVertexId().toString())) {
+                subgroup.add(message.getSourceVertexId().toString());
+            }
         }
         Text groupControllerKeyno = new Text();
         if (!thisVertexHoldKeyno.isEmpty()) {
@@ -176,6 +198,7 @@ public class EnterpriseGroupVertex extends
         vertexEntity.setGroupHolderCompanyIds(groupHolderCompanyId);
         vertexEntity.setGroupControllerKeyno(groupControllerKeyno);
         vertexEntity.setStockRightControlChain(new Text(collect));
+        vertexEntity.setSubgroup(subgroup.list2Tuple());
         vertexEntity.setOutput();
     }
 
@@ -186,6 +209,12 @@ public class EnterpriseGroupVertex extends
             return;
         }
 
+        //todo:debug
+        prodEnvDebug.debug_2(getId()
+                , "vertex value:\n{}"
+                , getValue()
+        );
+
         //判断是否是集团企业
 //        if (hasEdges()) {
 //            List<Edge<Text, HolderEdge>> edges = getEdges();
@@ -199,7 +228,7 @@ public class EnterpriseGroupVertex extends
 //            }
 //        }
 
-        if(!getValue().isOutput()){
+        if (!getValue().isOutput()) {
             return;
         }
 
@@ -214,6 +243,13 @@ public class EnterpriseGroupVertex extends
         Text group_controller_keyno = getValue().getGroupControllerKeyno();
         Text stockRightControlChain = getValue().getStockRightControlChain();
 
+        List<String> subgroupList = getValue().getSubgroup().tuple2List();
+        Text subgroup = new Text();
+        if (getValue().getCompanyName().toString().contains("集团")) {
+            subgroup = new Text(subgroupList.stream().filter(e -> !e.equals(getId().toString())).collect(Collectors.joining(",")));
+        }
+
+
 //        System.out.println("save:"
 //                + enterprise_group_hold_company_id + ","
 //                + enterprise_group_hold_company_name + ","
@@ -225,6 +261,9 @@ public class EnterpriseGroupVertex extends
 //                + stockRightControlChain
 //        );
 
+        if (hold_num.get() < 3) {
+            return;
+        }
         if (hold_num.get() > 200) {
             String s = EnterpriseGroupUtils.stockChainTrim(enterprise_group_hold_company_id.toString(), stockRightControlChain.toString(), 10);
             stockRightControlChain = new Text(s);
@@ -240,6 +279,7 @@ public class EnterpriseGroupVertex extends
                 , group_holder_company_id
                 , group_controller_keyno
                 , stockRightControlChain
+                , subgroup
         );
     }
 }

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/ProdEnvDebug.java

@@ -0,0 +1,120 @@
+package com.winhc.max.compute.graph.job.enterprise_group;
+
+
+import com.aliyun.odps.conf.Configuration;
+import com.aliyun.odps.io.Text;
+import com.winhc.max.compute.graph.util.BaseUtils;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * @author: XuJiakai
+ * 2023/3/1 17:05
+ */
+public class ProdEnvDebug {
+
+    private static class WinhcDebugInfo extends RuntimeException {
+        public WinhcDebugInfo() {
+        }
+
+        public WinhcDebugInfo(String message) {
+            super(message);
+        }
+
+        public WinhcDebugInfo(String message, Throwable cause) {
+            super(message, cause);
+        }
+
+        public WinhcDebugInfo(Throwable cause) {
+            super(cause);
+        }
+
+        public WinhcDebugInfo(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
+            super(message, cause, enableSuppression, writableStackTrace);
+        }
+    }
+
+    private boolean enableDebug;
+
+    /**
+     * debug位置
+     */
+    private Set<String> debugIndices;
+
+    /**
+     * debug 顶点id
+     */
+    private String debugVertexId;
+
+
+    public static ProdEnvDebug build(Configuration configuration) {
+        ProdEnvDebug prodEnvDebug = new ProdEnvDebug();
+        prodEnvDebug.enableDebug = Boolean.parseBoolean(configuration.get("winhc.debug", "false"));
+        String debugIndex = configuration.get("winhc.debug.index", null);
+        if (debugIndex != null) {
+            prodEnvDebug.debugIndices = Arrays.stream(debugIndex.split(",")).collect(Collectors.toSet());
+        }
+        prodEnvDebug.debugVertexId = configuration.get("winhc.debug.debugVertexId", null);
+        return prodEnvDebug;
+    }
+
+    public void debug(Text vertexId, String template, Object... obj) {
+        if (!enableDebug) {
+            return;
+        }
+        String str = BaseUtils.format(template, obj);
+        if (vertexId.toString().equals(debugVertexId)) {
+            throw new WinhcDebugInfo(BaseUtils.format("\nvertex id: {}\n{}\n", vertexId, str));
+        }
+    }
+
+    public void debugByIndex(String index, Text vertexId, String template, Object... obj) {
+        if (!enableDebug) {
+            return;
+        }
+        String str = BaseUtils.format(template, obj);
+        String format = BaseUtils.format("\ndebug index: {}  vertex id: {}\n{}\n", index, vertexId, str);
+        if (debugIndices == null || debugIndices.isEmpty()) {
+            throw new WinhcDebugInfo(format);
+        }
+        if (vertexId.toString().equals(debugVertexId) && debugIndices.contains(index)) {
+            throw new WinhcDebugInfo(format);
+        }
+    }
+
+    public void debug_1(Text vertexId, String template, Object... obj) {
+        debugByIndex("1", vertexId, template, obj);
+    }
+
+    public void debug_2(Text vertexId, String template, Object... obj) {
+        debugByIndex("2", vertexId, template, obj);
+    }
+
+    public void debug_3(Text vertexId, String template, Object... obj) {
+        debugByIndex("3", vertexId, template, obj);
+    }
+
+
+    public static void debugByCondition(boolean condition, String template, Object... obj) {
+        if (condition) {
+            throw new WinhcDebugInfo(BaseUtils.format(template, obj));
+        }
+    }
+
+
+    public static void main(String[] args) {
+        ProdEnvDebug prodEnvDebug = new ProdEnvDebug();
+
+        prodEnvDebug.debugIndices = new HashSet<String>() {{
+            add("1");
+        }};
+
+        prodEnvDebug.enableDebug = true;
+        prodEnvDebug.debugVertexId = "aaa";
+
+        prodEnvDebug.debug_1(new Text("aaa"),"{}","a");
+    }
+}

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/entity/EntGroupMsg.java

@@ -105,10 +105,12 @@ public class EntGroupMsg implements Writable {
         }
         routeLog.append(new Text(vertexId));
         return true;
+
+
     }
 
 
-    public static EntGroupMsg ofByType_1(Text thisVertexCompanyId, Text thisVertexCompanyName, List<String> groupHolderCompanyIds) {
+    public static EntGroupMsg ofByType_1(Text thisVertexCompanyId, Text thisVertexCompanyName, IntWritable companyType, List<String> groupHolderCompanyIds) {
         //发给控股企业消息
         EntGroupMsg entGroupMsg = new EntGroupMsg();
         entGroupMsg.msgType = new IntWritable(1);
@@ -116,14 +118,14 @@ public class EntGroupMsg implements Writable {
         entGroupMsg.holdCompanyIds = new Tuple(1);
         entGroupMsg.groupHolderCompanyIds = new Tuple();
 
-        entGroupMsg.holdCompanyIds.append(new Text(thisVertexCompanyId));
+        entGroupMsg.holdCompanyIds.append(thisVertexCompanyId);
         if (groupHolderCompanyIds != null) {
             for (String s : groupHolderCompanyIds) {
                 entGroupMsg.groupHolderCompanyIds.append(new Text(s));
             }
         }
 
-        addOtherMsg(entGroupMsg, thisVertexCompanyId, thisVertexCompanyId);
+        entGroupMsg.addOtherMsg(thisVertexCompanyId, thisVertexCompanyName, companyType.get());
         return entGroupMsg;
     }
 
@@ -136,20 +138,32 @@ public class EntGroupMsg implements Writable {
         entGroupMsg.groupInvestmentCompanyIds = new Tuple(1);
         entGroupMsg.groupInvestmentCompanyIds.append(new Text(thisVertexCompanyId));
 
-        addOtherMsg(entGroupMsg, thisVertexCompanyId, thisVertexCompanyId);
+        entGroupMsg.addOtherMsg(thisVertexCompanyId, null, null);
         return entGroupMsg;
     }
 
 
-    private static void addOtherMsg(EntGroupMsg entGroupMsg, Text thisVertexCompanyId, Text thisVertexCompanyName) {
-        entGroupMsg.sourceVertexId = thisVertexCompanyId;
-        entGroupMsg.sourceVertexName = thisVertexCompanyName;
+    private void addOtherMsg(Text thisVertexCompanyId, Text thisVertexCompanyName, Integer companyType) {
+        if (thisVertexCompanyName != null && thisVertexCompanyId != null && companyType != null) {
+            if (thisVertexCompanyName.toString().contains("集团") && companyType != 2) {
+                this.sourceVertexId = thisVertexCompanyId;
+                this.sourceVertexName = thisVertexCompanyName;
+            }
+        }
+
+        this.routeLog = new Tuple();
+        this.routeLog.append(thisVertexCompanyId);
+        this.stockRightControlChain = new Text();
+    }
+
 
-        entGroupMsg.routeLog = new Tuple();
-        entGroupMsg.routeLog.append(thisVertexCompanyId);
-        entGroupMsg.stockRightControlChain = new Text();
+    public Text getSourceVertexId() {
+        return sourceVertexId;
     }
 
+    public Text getSourceVertexName() {
+        return sourceVertexName;
+    }
 
     public int getMsgType() {
         return msgType.get();
@@ -194,6 +208,12 @@ public class EntGroupMsg implements Writable {
         if (stockRightControlChain == null)
             stockRightControlChain = new Text();
 
+        if (sourceVertexId == null)
+            sourceVertexId = new Text();
+        if (sourceVertexName == null) {
+            sourceVertexName = new Text();
+        }
+
         msgType.write(out);
         holdNum.write(out);
         holdCompanyIds.write(out);
@@ -201,6 +221,8 @@ public class EntGroupMsg implements Writable {
         groupHolderCompanyIds.write(out);
         routeLog.write(out);
         stockRightControlChain.write(out);
+        this.sourceVertexId.write(out);
+        this.sourceVertexName.write(out);
     }
 
     @Override
@@ -214,6 +236,10 @@ public class EntGroupMsg implements Writable {
         routeLog = new Tuple();
         stockRightControlChain = new Text();
 
+        sourceVertexId = new Text();
+        sourceVertexName = new Text();
+
+
         msgType.readFields(in);
         holdNum.readFields(in);
         holdCompanyIds.readFields(in);
@@ -221,6 +247,8 @@ public class EntGroupMsg implements Writable {
         groupHolderCompanyIds.readFields(in);
         routeLog.readFields(in);
         stockRightControlChain.readFields(in);
+        this.sourceVertexId.readFields(in);
+        this.sourceVertexName.readFields(in);
 
     }
 }

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/entity/EnterpriseGroupVertexValue.java

@@ -22,7 +22,7 @@ public class EnterpriseGroupVertexValue implements Writable {
 
     private Text companyName;
     /**
-     * 公司类型，1为正常企业，2为分公司，3为合伙，4为机关单位，5为事业单位
+     * 公司类型，1为正常企业，2为分公司，3为合伙，4为机关单位，5为事业单位，6为人员节点
      */
     private IntWritable companyType;
     /**
@@ -37,6 +37,11 @@ public class EnterpriseGroupVertexValue implements Writable {
 
     //以上为初始化信息，以下是计算相关信息
 
+    /**
+     * 子集团id
+     */
+    private Tuple subgroup;
+
 
     /**
      * 下属控股企业数
@@ -95,6 +100,10 @@ public class EnterpriseGroupVertexValue implements Writable {
     }
 
 
+    public IntWritable getCompanyType() {
+        return companyType;
+    }
+
     public Text getStockRightControlChain() {
         return stockRightControlChain;
     }
@@ -103,6 +112,18 @@ public class EnterpriseGroupVertexValue implements Writable {
         this.stockRightControlChain = stockRightControlChain;
     }
 
+    public BooleanWritable getManualStop() {
+        return manualStop;
+    }
+
+    public void setSubgroup(Tuple subgroup) {
+        this.subgroup = subgroup;
+    }
+
+    public Tuple getSubgroup() {
+        return subgroup;
+    }
+
     public Text getCompanyName() {
         return companyName;
     }
@@ -214,6 +235,10 @@ public class EnterpriseGroupVertexValue implements Writable {
         if (legalEntityIds == null) {
             manualStop = new BooleanWritable();
         }
+
+        if (subgroup == null) {
+            subgroup = new Tuple();
+        }
     }
 
 
@@ -222,13 +247,14 @@ public class EnterpriseGroupVertexValue implements Writable {
         companyName.write(out);
         companyType.write(out);
         legalEntityIds.write(out);
+        manualStop.write(out);
+        subgroup.write(out);
         holdNum.write(out);
         holdCompanyIds.write(out);
         groupInvestmentCompanyIds.write(out);
         groupHolderCompanyIds.write(out);
         groupControllerKeyno.write(out);
         stockRightControlChain.write(out);
-        manualStop.write(out);
         endFlag.write(out);
     }
 
@@ -237,25 +263,46 @@ public class EnterpriseGroupVertexValue implements Writable {
         companyName = new Text();
         companyType = new IntWritable();
         legalEntityIds = new Tuple();
+        manualStop = new BooleanWritable();
+        subgroup = new Tuple();
         holdNum = new LongWritable();
         holdCompanyIds = new Tuple();
         groupInvestmentCompanyIds = new Tuple();
         groupHolderCompanyIds = new Tuple();
         groupControllerKeyno = new Text();
         stockRightControlChain = new Text();
-        manualStop = new BooleanWritable();
         endFlag = new IntWritable();
 
         companyName.readFields(in);
         companyType.readFields(in);
         legalEntityIds.readFields(in);
+        manualStop.readFields(in);
+        subgroup.readFields(in);
         holdNum.readFields(in);
         holdCompanyIds.readFields(in);
         groupInvestmentCompanyIds.readFields(in);
         groupHolderCompanyIds.readFields(in);
         groupControllerKeyno.readFields(in);
         stockRightControlChain.readFields(in);
-        manualStop.readFields(in);
         endFlag.readFields(in);
     }
+
+
+    @Override
+    public String toString() {
+        return "EnterpriseGroupVertexValue{" +
+                "companyName=" + companyName +
+                ", companyType=" + companyType +
+                ", legalEntityIds=" + legalEntityIds +
+                ", manualStop=" + manualStop +
+                ", subgroup=" + subgroup +
+                ", holdNum=" + holdNum +
+                ", holdCompanyIds=" + holdCompanyIds +
+                ", groupInvestmentCompanyIds=" + groupInvestmentCompanyIds +
+                ", groupHolderCompanyIds=" + groupHolderCompanyIds +
+                ", groupControllerKeyno=" + groupControllerKeyno +
+                ", stockRightControlChain=" + stockRightControlChain +
+                ", endFlag=" + endFlag +
+                '}';
+    }
 }

src/main/java/com/winhc/max/compute/graph/job/enterprise_group/entity/HolderEdge.java

@@ -40,6 +40,13 @@ public class HolderEdge implements Writable {
         return holderEdge;
     }
 
+    public static HolderEdge of(String holderId,Integer holderType, Double investmentProportion) {
+        HolderEdge holderEdge = new HolderEdge();
+        holderEdge.holderType = new IntWritable(holderType);
+        holderEdge.investmentProportion = new DoubleWritable(investmentProportion);
+        return holderEdge;
+    }
+
 
     @Override
     public void write(DataOutput out) throws IOException {

src/main/java/com/winhc/max/compute/graph/util/ArrayUtil.java

@@ -0,0 +1,74 @@
+package com.winhc.max.compute.graph.util;
+
+import java.util.Arrays;
+
+/**
+ * @author: XuJiakai
+ * 2023/3/1 17:14
+ */
+public class ArrayUtil {
+
+    /**
+     * 数组是否为空
+     *
+     * @param <T> 数组元素类型
+     * @param array 数组
+     * @return 是否为空
+     */
+    public static <T> boolean isEmpty(T[] array) {
+        return array == null || array.length == 0;
+    }
+
+    /**
+     * 对象是否为数组对象
+     *
+     * @param obj 对象
+     * @return 是否为数组对象，如果为{@code null} 返回false
+     */
+    public static boolean isArray(Object obj) {
+        if (null == obj) {
+            // throw new NullPointerException("Object check for isArray is null");
+            return false;
+        }
+        return obj.getClass().isArray();
+    }
+
+    /**
+     * 数组或集合转String
+     *
+     * @param obj 集合或数组对象
+     * @return 数组字符串，与集合转字符串格式相同
+     */
+    public static String toString(Object obj) {
+        if (null == obj) {
+            return null;
+        }
+
+        if(obj instanceof long[]){
+            return Arrays.toString((long[]) obj);
+        } else if(obj instanceof int[]){
+            return Arrays.toString((int[]) obj);
+        } else if(obj instanceof short[]){
+            return Arrays.toString((short[]) obj);
+        } else if(obj instanceof char[]){
+            return Arrays.toString((char[]) obj);
+        } else if(obj instanceof byte[]){
+            return Arrays.toString((byte[]) obj);
+        } else if(obj instanceof boolean[]){
+            return Arrays.toString((boolean[]) obj);
+        } else if(obj instanceof float[]){
+            return Arrays.toString((float[]) obj);
+        } else if(obj instanceof double[]){
+            return Arrays.toString((double[]) obj);
+        } else if (ArrayUtil.isArray(obj)) {
+            // 对象数组
+            try {
+                return Arrays.deepToString((Object[]) obj);
+            } catch (Exception ignore) {
+                //ignore
+            }
+        }
+
+        return obj.toString();
+    }
+}

src/main/java/com/winhc/max/compute/graph/util/BaseUtils.java

@@ -0,0 +1,89 @@
+package com.winhc.max.compute.graph.util;
+
+import java.util.List;
+
+/**
+ * @author: XuJiakai
+ * 2023/2/28 14:56
+ */
+public class BaseUtils {
+    public static Boolean isWindows() {
+        return System.getProperty("os.name").contains("Windows");
+    }
+
+    public static boolean containsStr(List<String> list, String str) {
+        for (String tmpText : list) {
+            boolean flag = tmpText.contains(str);
+            if (flag) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * 格式化字符串<br>
+     * 此方法只是简单将占位符 {} 按照顺序替换为参数<br>
+     * 如果想输出 {} 使用 \\转义 { 即可，如果想输出 {} 之前的 \ 使用双转义符 \\\\ 即可<br>
+     * 例：<br>
+     * 通常使用：format("this is {} for {}", "a", "b") =》 this is a for b<br>
+     * 转义{}： format("this is \\{} for {}", "a", "b") =》 this is \{} for a<br>
+     * 转义\： format("this is \\\\{} for {}", "a", "b") =》 this is \a for b<br>
+     *
+     * @param strPattern 字符串模板
+     * @param argArray 参数列表
+     * @return 结果
+     */
+
+    public static String format(final String strPattern, final Object... argArray) {
+        if (StrUtil.isBlank(strPattern) || ArrayUtil.isEmpty(argArray)) {
+            return strPattern;
+        }
+        final int strPatternLength = strPattern.length();
+
+        // 初始化定义好的长度以获得更好的性能
+        StringBuilder sbuf = new StringBuilder(strPatternLength + 50);
+
+        int handledPosition = 0;// 记录已经处理到的位置
+        int delimIndex;// 占位符所在位置
+        for (int argIndex = 0; argIndex < argArray.length; argIndex++) {
+            delimIndex = strPattern.indexOf(StrUtil.EMPTY_JSON, handledPosition);
+            if (delimIndex == -1) {// 剩余部分无占位符
+                if (handledPosition == 0) { // 不带占位符的模板直接返回
+                    return strPattern;
+                }
+                // 字符串模板剩余部分不再包含占位符，加入剩余部分后返回结果
+                sbuf.append(strPattern, handledPosition, strPatternLength);
+                return sbuf.toString();
+            }
+
+            // 转义符
+            if (delimIndex > 0 && strPattern.charAt(delimIndex - 1) == StrUtil.C_BACKSLASH) {// 转义符
+                if (delimIndex > 1 && strPattern.charAt(delimIndex - 2) == StrUtil.C_BACKSLASH) {// 双转义符
+                    // 转义符之前还有一个转义符，占位符依旧有效
+                    sbuf.append(strPattern, handledPosition, delimIndex - 1);
+                    sbuf.append(StrUtil.utf8Str(argArray[argIndex]));
+                    handledPosition = delimIndex + 2;
+                } else {
+                    // 占位符被转义
+                    argIndex--;
+                    sbuf.append(strPattern, handledPosition, delimIndex - 1);
+                    sbuf.append(StrUtil.C_DELIM_START);
+                    handledPosition = delimIndex + 1;
+                }
+            } else {// 正常占位符
+                sbuf.append(strPattern, handledPosition, delimIndex);
+                sbuf.append(StrUtil.utf8Str(argArray[argIndex]));
+                handledPosition = delimIndex + 2;
+            }
+        }
+
+        // append the characters following the last {} pair.
+        // 加入最后一个占位符后所有的字符
+        sbuf.append(strPattern, handledPosition, strPattern.length());
+
+        return sbuf.toString();
+    }
+
+
+}

src/main/java/com/winhc/max/compute/graph/util/CharUtil.java

@@ -0,0 +1,331 @@
+package com.winhc.max.compute.graph.util;
+
+/**
+ * @author: XuJiakai
+ * 2023/3/1 17:24
+ */
+public class CharUtil {
+
+
+    public static final char SPACE = ' ';
+    public static final char TAB = '	';
+    public static final char DOT = '.';
+    public static final char SLASH = '/';
+    public static final char BACKSLASH = '\\';
+    public static final char CR = '\r';
+    public static final char LF = '\n';
+    public static final char UNDERLINE = '_';
+    public static final char DASHED = '-';
+    public static final char COMMA = ',';
+    public static final char DELIM_START = '{';
+    public static final char DELIM_END = '}';
+    public static final char BRACKET_START = '[';
+    public static final char BRACKET_END = ']';
+    public static final char COLON = ':';
+    public static final char DOUBLE_QUOTES = '"';
+    public static final char SINGLE_QUOTE = '\'';
+    public static final char AMP = '&';
+
+    /**
+     * 是否为ASCII字符，ASCII字符位于0~127之间
+     *
+     * <pre>
+     *   CharUtil.isAscii('a')  = true
+     *   CharUtil.isAscii('A')  = true
+     *   CharUtil.isAscii('3')  = true
+     *   CharUtil.isAscii('-')  = true
+     *   CharUtil.isAscii('\n') = true
+     *   CharUtil.isAscii('&copy;') = false
+     * </pre>
+     *
+     * @param ch 被检查的字符处
+     * @return true表示为ASCII字符，ASCII字符位于0~127之间
+     */
+    public static boolean isAscii(char ch) {
+        return ch < 128;
+    }
+
+    /**
+     * 是否为可见ASCII字符，可见字符位于32~126之间
+     *
+     * <pre>
+     *   CharUtil.isAsciiPrintable('a')  = true
+     *   CharUtil.isAsciiPrintable('A')  = true
+     *   CharUtil.isAsciiPrintable('3')  = true
+     *   CharUtil.isAsciiPrintable('-')  = true
+     *   CharUtil.isAsciiPrintable('\n') = false
+     *   CharUtil.isAsciiPrintable('&copy;') = false
+     * </pre>
+     *
+     * @param ch 被检查的字符处
+     * @return true表示为ASCII可见字符，可见字符位于32~126之间
+     */
+    public static boolean isAsciiPrintable(char ch) {
+        return ch >= 32 && ch < 127;
+    }
+
+    /**
+     * 是否为ASCII控制符（不可见字符），控制符位于0~31和127
+     *
+     * <pre>
+     *   CharUtil.isAsciiControl('a')  = false
+     *   CharUtil.isAsciiControl('A')  = false
+     *   CharUtil.isAsciiControl('3')  = false
+     *   CharUtil.isAsciiControl('-')  = false
+     *   CharUtil.isAsciiControl('\n') = true
+     *   CharUtil.isAsciiControl('&copy;') = false
+     * </pre>
+     *
+     * @param ch 被检查的字符
+     * @return true表示为控制符，控制符位于0~31和127
+     */
+    public static boolean isAsciiControl(final char ch) {
+        return ch < 32 || ch == 127;
+    }
+
+    /**
+     * 判断是否为字母（包括大写字母和小写字母）<br>
+     * 字母包括A~Z和a~z
+     *
+     * <pre>
+     *   CharUtil.isLetter('a')  = true
+     *   CharUtil.isLetter('A')  = true
+     *   CharUtil.isLetter('3')  = false
+     *   CharUtil.isLetter('-')  = false
+     *   CharUtil.isLetter('\n') = false
+     *   CharUtil.isLetter('&copy;') = false
+     * </pre>
+     *
+     * @param ch 被检查的字符
+     * @return true表示为字母（包括大写字母和小写字母）字母包括A~Z和a~z
+     */
+    public static boolean isLetter(char ch) {
+        return isLetterUpper(ch) || isLetterLower(ch);
+    }
+
+    /**
+     * <p>
+     * 判断是否为大写字母，大写字母包括A~Z
+     * </p>
+     *
+     * <pre>
+     *   CharUtil.isLetterUpper('a')  = false
+     *   CharUtil.isLetterUpper('A')  = true
+     *   CharUtil.isLetterUpper('3')  = false
+     *   CharUtil.isLetterUpper('-')  = false
+     *   CharUtil.isLetterUpper('\n') = false
+     *   CharUtil.isLetterUpper('&copy;') = false
+     * </pre>
+     *
+     * @param ch 被检查的字符
+     * @return true表示为大写字母，大写字母包括A~Z
+     */
+    public static boolean isLetterUpper(final char ch) {
+        return ch >= 'A' && ch <= 'Z';
+    }
+
+    /**
+     * <p>
+     * 检查字符是否为小写字母，小写字母指a~z
+     * </p>
+     *
+     * <pre>
+     *   CharUtil.isLetterLower('a')  = true
+     *   CharUtil.isLetterLower('A')  = false
+     *   CharUtil.isLetterLower('3')  = false
+     *   CharUtil.isLetterLower('-')  = false
+     *   CharUtil.isLetterLower('\n') = false
+     *   CharUtil.isLetterLower('&copy;') = false
+     * </pre>
+     *
+     * @param ch 被检查的字符
+     * @return true表示为小写字母，小写字母指a~z
+     */
+    public static boolean isLetterLower(final char ch) {
+        return ch >= 'a' && ch <= 'z';
+    }
+
+    /**
+     * <p>
+     * 检查是否为数字字符，数字字符指0~9
+     * </p>
+     *
+     * <pre>
+     *   CharUtil.isNumber('a')  = false
+     *   CharUtil.isNumber('A')  = false
+     *   CharUtil.isNumber('3')  = true
+     *   CharUtil.isNumber('-')  = false
+     *   CharUtil.isNumber('\n') = false
+     *   CharUtil.isNumber('&copy;') = false
+     * </pre>
+     *
+     * @param ch 被检查的字符
+     * @return true表示为数字字符，数字字符指0~9
+     */
+    public static boolean isNumber(char ch) {
+        return ch >= '0' && ch <= '9';
+    }
+
+    /**
+     * 是否为16进制规范的字符，判断是否为如下字符
+     * <pre>
+     * 1. 0~9
+     * 2. a~f
+     * 4. A~F
+     * </pre>
+     *
+     * @param c 字符
+     * @return 是否为16进制规范的字符
+     * @since 4.1.5
+     */
+    public static boolean isHexChar(char c) {
+        return isNumber(c) || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F');
+    }
+
+    /**
+     * 是否为字符或数字，包括A~Z、a~z、0~9
+     *
+     * <pre>
+     *   CharUtil.isLetterOrNumber('a')  = true
+     *   CharUtil.isLetterOrNumber('A')  = true
+     *   CharUtil.isLetterOrNumber('3')  = true
+     *   CharUtil.isLetterOrNumber('-')  = false
+     *   CharUtil.isLetterOrNumber('\n') = false
+     *   CharUtil.isLetterOrNumber('&copy;') = false
+     * </pre>
+     *
+     * @param ch 被检查的字符
+     * @return true表示为字符或数字，包括A~Z、a~z、0~9
+     */
+    public static boolean isLetterOrNumber(final char ch) {
+        return isLetter(ch) || isNumber(ch);
+    }
+
+
+    /**
+     * 给定类名是否为字符类，字符类包括：
+     *
+     * <pre>
+     * Character.class
+     * char.class
+     * </pre>
+     *
+     * @param clazz 被检查的类
+     * @return true表示为字符类
+     */
+    public static boolean isCharClass(Class<?> clazz) {
+        return clazz == Character.class || clazz == char.class;
+    }
+
+    /**
+     * 给定对象对应的类是否为字符类，字符类包括：
+     *
+     * <pre>
+     * Character.class
+     * char.class
+     * </pre>
+     *
+     * @param value 被检查的对象
+     * @return true表示为字符类
+     */
+    public static boolean isChar(Object value) {
+        //noinspection ConstantConditions
+        return value instanceof Character || value.getClass() == char.class;
+    }
+
+    /**
+     * 是否空白符<br>
+     * 空白符包括空格、制表符、全角空格和不间断空格<br>
+     *
+     * @param c 字符
+     * @return 是否空白符
+     * @see Character#isWhitespace(int)
+     * @see Character#isSpaceChar(int)
+     * @since 4.0.10
+     */
+    public static boolean isBlankChar(char c) {
+        return isBlankChar((int) c);
+    }
+
+    /**
+     * 是否空白符<br>
+     * 空白符包括空格、制表符、全角空格和不间断空格<br>
+     *
+     * @param c 字符
+     * @return 是否空白符
+     * @see Character#isWhitespace(int)
+     * @see Character#isSpaceChar(int)
+     * @since 4.0.10
+     */
+    public static boolean isBlankChar(int c) {
+        return Character.isWhitespace(c) || Character.isSpaceChar(c) || c == '\ufeff' || c == '\u202a';
+    }
+
+    /**
+     * 判断是否为emoji表情符<br>
+     *
+     * @param c 字符
+     * @return 是否为emoji
+     * @since 4.0.8
+     */
+    public static boolean isEmoji(char c) {
+        //noinspection ConstantConditions
+        return false == ((c == 0x0) || //
+                (c == 0x9) || //
+                (c == 0xA) || //
+                (c == 0xD) || //
+                ((c >= 0x20) && (c <= 0xD7FF)) || //
+                ((c >= 0xE000) && (c <= 0xFFFD)) || //
+                ((c >= 0x100000) && (c <= 0x10FFFF)));
+    }
+
+    /**
+     * 是否为Windows或者Linux（Unix）文件分隔符<br>
+     * Windows平台下分隔符为\，Linux（Unix）为/
+     *
+     * @param c 字符
+     * @return 是否为Windows或者Linux（Unix）文件分隔符
+     * @since 4.1.11
+     */
+    public static boolean isFileSeparator(char c) {
+        return SLASH == c || BACKSLASH == c;
+    }
+
+    /**
+     * 比较两个字符是否相同
+     *
+     * @param c1         字符1
+     * @param c2         字符2
+     * @param ignoreCase 是否忽略大小写
+     * @return 是否相同
+     * @since 4.0.3
+     */
+    public static boolean equals(char c1, char c2, boolean ignoreCase) {
+        if (ignoreCase) {
+            return Character.toLowerCase(c1) == Character.toLowerCase(c2);
+        }
+        return c1 == c2;
+    }
+
+    /**
+     * 获取字符类型
+     *
+     * @param c 字符
+     * @return 字符类型
+     * @since 5.2.3
+     */
+    public static int getType(int c) {
+        return Character.getType(c);
+    }
+
+    /**
+     * 获取给定字符的16进制数值
+     *
+     * @param b 字符
+     * @return 16进制字符
+     * @since 5.3.1
+     */
+    public static int digit16(int b) {
+        return Character.digit(b, 16);
+    }
+}

src/main/java/com/winhc/max/compute/graph/util/CharsetUtil.java

@@ -0,0 +1,61 @@
+package com.winhc.max.compute.graph.util;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.charset.UnsupportedCharsetException;
+
+/**
+ * @author: XuJiakai
+ * 2023/3/1 17:23
+ */
+public class CharsetUtil {
+
+    /**
+     * ISO-8859-1
+     */
+    public static final String ISO_8859_1 = "ISO-8859-1";
+    /**
+     * UTF-8
+     */
+    public static final String UTF_8 = "UTF-8";
+    /**
+     * GBK
+     */
+    public static final String GBK = "GBK";
+
+    /**
+     * ISO-8859-1
+     */
+    public static final Charset CHARSET_ISO_8859_1 = StandardCharsets.ISO_8859_1;
+    /**
+     * UTF-8
+     */
+    public static final Charset CHARSET_UTF_8 = StandardCharsets.UTF_8;
+    /**
+     * GBK
+     */
+    public static final Charset CHARSET_GBK;
+
+    static {
+        //避免不支持GBK的系统中运行报错 issue#731
+        Charset _CHARSET_GBK = null;
+        try {
+            _CHARSET_GBK = Charset.forName(GBK);
+        } catch (UnsupportedCharsetException e) {
+            //ignore
+        }
+        CHARSET_GBK = _CHARSET_GBK;
+    }
+
+    /**
+     * 转换为Charset对象
+     *
+     * @param charsetName 字符集，为空则返回默认字符集
+     * @return Charset
+     * @throws UnsupportedCharsetException 编码不支持
+     */
+    public static Charset charset(String charsetName) throws UnsupportedCharsetException {
+        return StrUtil.isBlank(charsetName) ? Charset.defaultCharset() : Charset.forName(charsetName);
+    }
+
+}

src/main/java/com/winhc/max/compute/graph/util/DateUtils.java

@@ -0,0 +1,115 @@
+package com.winhc.max.compute.graph.util;
+
+import java.text.SimpleDateFormat;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.util.Date;
+import java.util.Locale;
+
+/**
+ * @author ZhangJi
+ * @since 2021-10-22 15:09
+ */
+public class DateUtils {
+    public static final DateTimeFormatter YYYYMMDD = DateTimeFormatter.ofPattern("yyyyMMdd");
+    public static final DateTimeFormatter YYYYMMDDHHMMSS = DateTimeFormatter.ofPattern("yyyyMMdd HH:mm:ss");
+    public static final DateTimeFormatter YYYY_MM_DDHHMMSS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+    public static final DateTimeFormatter YYYY_MM_DD = DateTimeFormatter.ofPattern("yyyy-MM-dd");
+    public static final String FORMAT_YYYY_MM_DDHHMMSS = "yyyy-MM-dd HH:mm:ss";
+
+    public static final String formatDate(Date date, String pattern) {
+        String v = null;
+        try {
+            if (date == null)
+                return null;
+            SimpleDateFormat dateFormat = new SimpleDateFormat(pattern);
+            v = dateFormat.format(date);
+        } catch (Exception e) {
+            // do nothing
+        }
+        return v;
+    }
+
+    public static String nowDate() {
+        return nowDate(null);
+    }
+
+    public static String nowDate(DateTimeFormatter pattern) {
+        if (pattern == null) {
+            pattern = DateTimeFormatter.ISO_DATE;
+        }
+        return LocalDateTime.now().format(pattern);
+    }
+
+    public static String nowDateTime() {
+        return nowDateTime(null);
+    }
+
+    public static String nowDateTime(DateTimeFormatter pattern) {
+        if (pattern == null) {
+            pattern = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
+        }
+        return LocalDateTime.now().format(pattern);
+    }
+
+    public static String getYesterday() {
+        return LocalDate.now().plusDays(-1).format(DateTimeFormatter.BASIC_ISO_DATE);
+    }
+
+    public static final String FORMAT_YYYY_MM_DD_HH_MM_SS = "yyyy-MM-dd HH:mm:ss";
+
+    public static final String formatDate_YYYY_MM_DD_HH_MM_SS(Date date) {
+        String v = null;
+        try {
+            if (date == null)
+                return null;
+            SimpleDateFormat dateFormat = new SimpleDateFormat(FORMAT_YYYY_MM_DD_HH_MM_SS);
+            v = dateFormat.format(date);
+        } catch (Exception e) {
+            // do nothing
+        }
+        return v;
+    }
+
+    public static final Date parseDate(String d) {
+        Date v = null;
+        try {
+            if (d == null)
+                return null;
+            SimpleDateFormat dateFormat = new SimpleDateFormat(FORMAT_YYYY_MM_DDHHMMSS);
+            v = dateFormat.parse(d);
+        } catch (Exception e) {
+            // do nothing
+        }
+        return v;
+    }
+
+    public static final String formatDate_YYYY_MM_DD_HH_MM_SS() {
+        return formatDate_YYYY_MM_DD_HH_MM_SS(new Date());
+    }
+    private static final DateTimeFormatter df = DateTimeFormatter.ofPattern(FORMAT_YYYY_MM_DDHHMMSS)
+            .withLocale(Locale.CHINA)
+            .withZone(ZoneId.systemDefault());
+
+    public static String getAnyTime(Integer i) {
+        Instant instant = Instant.now().minus(i, ChronoUnit.DAYS);
+        return df.format(instant);
+    }
+
+    public static void main(String[] args) {
+        System.out.println(getAnyTime(31));
+        System.out.println(parseDate("2014-01-22 00:00:00"));
+        System.out.println(parseDate("2014-01-22"));
+        System.out.println(nowDate(null));
+        System.out.println(nowDateTime(null));
+        System.out.println(getYesterday());
+        System.out.println(DateUtils.nowDate(DateTimeFormatter.BASIC_ISO_DATE));
+        System.out.println(nowDate(YYYYMMDD));
+        System.out.println(nowDate(YYYY_MM_DD));
+        System.out.println(nowDate(YYYY_MM_DDHHMMSS));
+    }
+}

src/main/java/com/winhc/max/compute/graph/util/StrUtil.java

@@ -0,0 +1,101 @@
+package com.winhc.max.compute.graph.util;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+
+/**
+ * @author: XuJiakai
+ * 2023/3/1 17:16
+ */
+public class StrUtil {
+    public static final String EMPTY_JSON = "{}";
+    public static final char DELIM_START = '{';
+    public static final char C_DELIM_START = DELIM_START;
+    public static final char BACKSLASH = '\\';
+    public static final char C_BACKSLASH = BACKSLASH;
+    /**
+     * 将对象转为字符串<br>
+     * 1、Byte数组和ByteBuffer会被转换为对应字符串的数组 2、对象数组会调用Arrays.toString方法
+     *
+     * @param obj 对象
+     * @return 字符串
+     */
+    public static String utf8Str(Object obj) {
+        return str(obj, CharsetUtil.CHARSET_UTF_8);
+    }
+
+    /**
+     * 字符串是否为空白 空白的定义如下： <br>
+     * 1、为null <br>
+     * 2、为不可见字符（如空格）<br>
+     * 3、""<br>
+     *
+     * @param str 被检测的字符串
+     * @return 是否为空
+     */
+    public static boolean isBlank(CharSequence str) {
+        int length;
+
+        if ((str == null) || ((length = str.length()) == 0)) {
+            return true;
+        }
+
+        for (int i = 0; i < length; i++) {
+            // 只要有一个非空字符即为非空字符串
+            if (false == CharUtil.isBlankChar(str.charAt(i))) {
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+
+    /**
+     * 将对象转为字符串
+     *
+     * <pre>
+     * 1、Byte数组和ByteBuffer会被转换为对应字符串的数组
+     * 2、对象数组会调用Arrays.toString方法
+     * </pre>
+     *
+     * @param obj         对象
+     * @param charsetName 字符集
+     * @return 字符串
+     */
+    public static String str(Object obj, String charsetName) {
+        return str(obj, Charset.forName(charsetName));
+    }
+
+    /**
+     * 将对象转为字符串
+     * <pre>
+     * 	 1、Byte数组和ByteBuffer会被转换为对应字符串的数组
+     * 	 2、对象数组会调用Arrays.toString方法
+     * </pre>
+     *
+     * @param obj     对象
+     * @param charset 字符集
+     * @return 字符串
+     */
+    public static String str(Object obj, Charset charset) {
+        if (null == obj) {
+            return null;
+        }
+
+        if (obj instanceof String) {
+            return (String) obj;
+        } else if (obj instanceof byte[]) {
+            return str((byte[]) obj, charset);
+        } else if (obj instanceof Byte[]) {
+            return str((Byte[]) obj, charset);
+        } else if (obj instanceof ByteBuffer) {
+            return str((ByteBuffer) obj, charset);
+        } else if (ArrayUtil.isArray(obj)) {
+            return ArrayUtil.toString(obj);
+        }
+
+        return obj.toString();
+    }
+
+}

src/main/java/com/winhc/max/compute/graph/util/WritableRecordExtensions.java

@@ -16,7 +16,7 @@ public class WritableRecordExtensions {
 
 
     @SneakyThrows
-    public static <T extends Writable> T getOrNull(WritableRecord writableRecord, String filedName, Class<T> type) throws IOException {
+    public static <T extends Writable> T getOrNullInstance(WritableRecord writableRecord, String filedName, Class<T> type) throws IOException {
         Writable writable = writableRecord.get(filedName);
         if (writable == null || writable.equals(NullWritable.get())) {
             return type.newInstance();
@@ -25,12 +25,18 @@ public class WritableRecordExtensions {
         }
     }
 
-    public static Text getTextOrNull(WritableRecord writableRecord, String filedName) throws IOException {
+    @SneakyThrows
+    public static <T extends Writable> T getOrNull(WritableRecord writableRecord, String filedName, Class<T> type) throws IOException {
         Writable writable = writableRecord.get(filedName);
-        if (writable.equals(NullWritable.get())) {
-            return new Text();
+        if (writable == null || writable.equals(NullWritable.get())) {
+            return null;
         } else {
-            return ((Text) writable);
+            return ((T) writable);
         }
     }
+
+
+    public static Text getTextOrNull(WritableRecord writableRecord, String filedName) throws IOException {
+        return getOrNullInstance(writableRecord, filedName, Text.class);
+    }
 }

src/main/java/com/winhc/max/compute/graph/util/WritableUtils.java

@@ -0,0 +1,30 @@
+package com.winhc.max.compute.graph.util;
+
+import com.aliyun.odps.io.Writable;
+import lombok.SneakyThrows;
+
+import java.io.DataOutput;
+
+/**
+ * @author: XuJiakai
+ * 2023/2/13 09:39
+ */
+public class WritableUtils {
+    @SneakyThrows
+    public static <T extends Writable> void write(DataOutput out, T... objects) {
+        for (T object : objects) {
+            if (object == null) {
+                Class<? extends Writable> aClass = object.getClass();
+                Writable o = aClass.newInstance();
+                o.write(out);
+            }else {
+                object.write(out);
+            }
+        }
+    }
+
+    public static void read() {
+
+    }
+
+}