<dev>

1. 优化功能逻辑

src/assemblies/package.xml

@@ -28,7 +28,7 @@
             <includes>
                 <include>*.yml</include>
                 <include>*.properties</include>
-                <include>regular</include>
+<!--                <include>regular</include>-->
             </includes>
         </fileSet>
 

src/main/java/com/zorkdata/desensitization/function/DesensitizationFunction.java

@@ -73,10 +73,14 @@ public class DesensitizationFunction<T, R> extends RichFlatMapFunction<Tuple2<Ob
         }
         // 正常数据脱敏
         Map<String, String> normalFieldsMap = logData.getNormalFields();
-        // 1 脱敏普通列中key
-        desensitizationNormalFieldByKey(normalFieldsMap);
-        // 2 脱敏message
-        desensitizationNormalFieldMessage(normalFieldsMap);
+        try {
+            // 1 脱敏普通列中key
+            desensitizationNormalFieldByKey(normalFieldsMap);
+            // 2 脱敏message
+            desensitizationNormalFieldMessage(normalFieldsMap);
+        } catch (Exception e) {
+            log.error("数据：{};解析错误：异常信息：{}", JSON.toJSONString(logData), e);
+        }
         return logData;
     }
 
@@ -182,6 +186,9 @@ public class DesensitizationFunction<T, R> extends RichFlatMapFunction<Tuple2<Ob
                 int commaIndex = ansArr.indexOf(GeneralConstants.SEMICOLON);
                 String keysString = ansArr.substring(0, commaIndex);
                 String valuesString = ansArr.substring(commaIndex);
+                if (ansArr.length() > commaIndex) {
+                    valuesString = ansArr.substring(commaIndex + 1);
+                }
                 String[] keysArr = keysString.split(GeneralConstants.COMMA);
                 List<Integer> valueIndex = new ArrayList<>();
                 for (int i = 0; i < keysArr.length; i++) {
@@ -194,7 +201,7 @@ public class DesensitizationFunction<T, R> extends RichFlatMapFunction<Tuple2<Ob
                         .append(ansArrTemp[0])
                         .append(GeneralConstants.CODE_MSG)
                         .append(ansArrTemp[1].split(GeneralConstants.SEMICOLON)[0])
-                        .append(GeneralConstants.COMMA);
+                        .append(GeneralConstants.SEMICOLON);
                 String[] valueArr = valuesString.split(GeneralConstants.COMMA);
                 int valueSize = valueArr.length / keysArr.length;
                 List<Integer> ansValueIndex = getAnsValueIndex(valueIndex, valueSize, keysArr.length);

src/main/java/com/zorkdata/desensitization/hadoop/HdfsLogDesensitization.java

@@ -85,8 +85,8 @@ public class HdfsLogDesensitization implements Serializable {
                 return new Tuple2<>(key, NullWritable.get());
             }
         }).setParallelism(jobConfig.getTransformerParallelism())
-                .writeAsText("file:///lmt/output", org.apache.flink.core.fs.FileSystem.WriteMode.OVERWRITE)
-                // .output(hadoopOutputFormat)
+                // .writeAsText("file:///lmt/output", org.apache.flink.core.fs.FileSystem.WriteMode.OVERWRITE)
+                .output(hadoopOutputFormat)
                 .setParallelism(jobConfig.getSinkParallelism());
         try {
             env.execute(jobConfig.getJobName());

src/main/resources/application.yml

@@ -48,7 +48,7 @@ hdfs_user: "hdfs"
 # hdfs日志源文件地址，若source为hdfs，则该地址必传，必须以斜杠结尾
 hdfs_src: "/tmp/datawarehouse4/jzjy/kcbp_biz_log/"
 # hdfs日志写入地址，非必传，默认写到hdfs-src目录下的output目录下，必须以斜杠结尾
-hdfs_dest: "/tmp/datawarehouse/jzjy/kcbp_biz_log/output3/"
+hdfs_dest: "/tmp/datawarehouse/jzjy/kcbp_biz_log/output4/"
 
 # cdh下载配置
 # cdh能执行hdfs命令的机器的ip

src/test/java/com/zorkdata/desensitization/function/DesensitizationFunctionTest.java

@@ -134,4 +134,39 @@ public class DesensitizationFunctionTest {
         log.info(JSON.toJSONString(map));
     }
 
+    @Test
+    public void test4() {
+        Map<String, String> conf = new HashMap<>(8);
+        conf.put(ConfigConstants.JOB_NAME, "jobName");
+        conf.put(ConfigConstants.SOURCE_PARALLELISM, "1");
+        conf.put(ConfigConstants.TRANSFORMER_PARALLELISM, "1");
+        conf.put(ConfigConstants.SINK_PARALLELISM, "1");
+        conf.put(ConfigConstants.HDFS_URI, "1");
+        conf.put(ConfigConstants.HDFS_USER, "1");
+        conf.put(ConfigConstants.HDFS_SRC, "1");
+        conf.put(ConfigConstants.HDFS_DEST, "1");
+        conf.put(ConfigConstants.MATCH_HOSTNAME, "*");
+        conf.put(ConfigConstants.START_TIME, "2020-11-07 21:22:20");
+        conf.put(ConfigConstants.END_TIME, "2020-11-07 23:40:30");
+        conf.put(ConfigConstants.NAME_KEY, "custname,otherlinkman,longname,YXYHMC,deputyname");
+        conf.put(ConfigConstants.MOBILE_KEY, "netaddr,telno,mobileno,faxno,hometelno,worktelno,fax,mobil,call");
+        conf.put(ConfigConstants.PHONE_KEY, "netaddr,telno,mobileno,faxno,hometelno,worktelno,fax,mobil,call");
+        conf.put(ConfigConstants.EMAIL_KEY, "email");
+        conf.put(ConfigConstants.ID_KEY, "idno,deputyidno");
+        conf.put(ConfigConstants.BANK_CARD_KEY, "lmbankid");
+        conf.put(ConfigConstants.ADDRESS_KEY, "addr,workaddr,otheraddr,othertelno");
+        conf.put(ConfigConstants.IP_KEY, "printip,ipaddr,operipaddr,clientip");
+        conf.put(ConfigConstants.MAC_KEY, "netaddr,clientmac");
+        conf.put(ConfigConstants.PASSWORD_KEY, "trdpwd,newpwd,oldfundpwd,newfundpwd,bankpwd,dynamicpwd1,dynamicpwd2,dynamicpwd,fundpwd,newbankpwd");
+        conf.put(ConfigConstants.POSITION_KEY, "orderamt,orderqty,orderprice,fundeffect,stkeffect,profitcost,oddqty,sumqty,ordercount,orderqty,orderamt,fundbal,fundavl,marketvalue,fund,stkvalue,stkqty,stkbal,stkavl,mktval,matchamt,fundeffect,fee_yhs,fee_jsxf,fee_sxf,fee_ghf,fee_qsf,fee_jygf,feefront,fee_jsf,fee_zgf,fundbjhgavl,bb_matchclearamt,nHQSBSL,fundeffect,fundbal,fundeffect,bondbal,bondavl,maxdraw,marketvalue,matchqty,funddraw");
+        JobInitConfig jobInitConfig = new JobInitConfig(conf);
+        JobConfig jobConfig = new JobConfig(jobInitConfig);
+        DesensitizationFunction<Object, Object> desensitizationFunction = new DesensitizationFunction(jobConfig);
+        Map<String, String> map = new HashMap<>(8);
+        String message = "20201227-202920 Req: NodeId=6007, QueueId=304, MsgId=6000000604B2949123C20B24, Len=346,Buf=01280000000000000000100000000000000040F2E92B8E7AKCXP00  GV2gODkBbGg=       410570 00220000000  010_CA=2.3&_ENDIAN=0&funid=12345&trdpwd=&netaddr=12345678 Ans: NodeId=6007, QueueId=304, MsgId=6000000604B2949123C20E24, Len=1284, Buf=01300000000000000000306007060070600740F2E92B8E7AKCXP00 GV2gODkBbGg=5FE87E20006007000005013147410570  01158000000     010_CA=2.3&_SYSID=6007&_ENDIAN=0&_RS_1=MESSAGE;3;LEVEL,CODE,MSG;funid, custname, lmbankid, addr, orderamt;1234,廖鸣韬,464512324564564,123456789,fadfad,1234,廖鸣韬,464512324564564,123456789,fadfad&_EORS_2=2&_RC=2&_CC=48&_TL=3:1;48:1;";
+        map.put(GeneralConstants.MESSAGE, message);
+        desensitizationFunction.desensitizationNormalFieldMessage(map);
+        log.info(JSON.toJSONString(map));
+    }
+
 }
\ No newline at end of file

[sonarqube]

⚠ Remove this commented out code. 📘

[sonarqube]

⛔ Remove this hard-coded password. 📘

[sonarqube]

SonarQube analysis reported 180 issues

• ⛔ 2 blocker
• 🚫 24 critical
• ⚠ 131 major
• 🔽 20 minor
• ℹ 3 info

Watch the comments in this conversation to review them.

Top 30 extra issues

Note: The following issues were found on lines that were not modified in the commit. Because these issues can't be reported as line comments, they are summarized here:

1. ⛔ Remove this hard-coded password. 📘
2. 🚫 Define a constant instead of duplicating this literal " {\n" 11 times. 📘
3. 🚫 [Define a constant instead of duplicating this literal " "type": \n" 11 times. 📘
4. 🚫 Define a constant instead of duplicating this literal " "string",\n" 6 times. 📘
5. 🚫 Define a constant instead of duplicating this literal " "null"\n" 6 times. 📘
6. 🚫 [Define a constant instead of duplicating this literal " ]\n" 11 times.](https://git.zorkdata.com/liaomingtao/transaction-log-desensitization/blob/4ac51ac782f1aa64c5bb00279445640186789ac7/src/main/java/com/zorkdata/desensitization/avro/AvroSchemaDef.java#L23) 📘
7. 🚫 Define a constant instead of duplicating this literal " },\n" 9 times. 📘
8. 🚫 Define a constant instead of duplicating this literal " "null",\n" 5 times. 📘
9. 🚫 Define a constant instead of duplicating this literal " {\n" 5 times. 📘
10. 🚫 Define a constant instead of duplicating this literal " "type": "map",\n" 5 times. 📘
11. 🚫 Define a constant instead of duplicating this literal " "values": "string"\n" 3 times. 📘
12. 🚫 Define a constant instead of duplicating this literal " }\n" 5 times. 📘
13. 🚫 Define a constant instead of duplicating this literal "序列化失败" 13 times. 📘
14. 🚫 Refactor this method to reduce its Cognitive Complexity from 22 to the 15 allowed. 📘
15. 🚫 Refactor this method to reduce its Cognitive Complexity from 98 to the 15 allowed. 📘
16. 🚫 Refactor this method to reduce its Cognitive Complexity from 22 to the 15 allowed. 📘
17. 🚫 Refactor this method to reduce its Cognitive Complexity from 22 to the 15 allowed. 📘
18. 🚫 Refactor this method to reduce its Cognitive Complexity from 19 to the 15 allowed. 📘
19. 🚫 Change this "try" to a try-with-resources. (sonar.java.source not set. Assuming 7 or greater.) 📘
20. 🚫 Refactor this code to not throw exceptions in finally blocks. 📘
21. 🚫 Refactor this code to not throw exceptions in finally blocks. 📘
22. 🚫 Change this "try" to a try-with-resources. (sonar.java.source not set. Assuming 7 or greater.) 📘
23. 🚫 Define a constant instead of duplicating this literal "jobName" 4 times. 📘
24. 🚫 Define a constant instead of duplicating this literal "2020-11-07 21:22:20" 4 times. 📘
25. 🚫 Define a constant instead of duplicating this literal "2020-11-07 23:40:30" 4 times. 📘
26. ⚠ Rename "jsonObject" which hides the field declared at line 39. 📘
27. ⚠ Remove this expression which always evaluates to "true" 📘
28. ⚠ Remove this expression which always evaluates to "true" 📘
29. ⚠ This block of commented-out lines of code should be removed. 📘
30. ⚠ 及时清理不再使用的代码段或配置信息。 📘

• ... 148 more