<dev>

1. 优化脚本

package/scripts/start_and_download.sh

 #!/usr/bin/env bash
 
-FLINK_TASK_CONF=application.yml
-REGULAR_TASK_NAME=regular
+# 必须修改
 CDH_HOST_IP=192.168.70.2
 CDH_HOST_USER=root
-CDH_HOST_PASSWORD=NuqUtwbJUBRmUwgh
+CDH_HOST_PASSWORD=password
+
+# 与配置文件保持一致
 HDFS_DEST=/tmp/datawarehouse/jzjy/kcbp_biz_log/output1/
 DOWNLOAD_PATH=/tmp/
 JOB_NAME="国泰交易日志脱敏job"
+
+FLINK_TASK_CONF=application.yml
+REGULAR_TASK_NAME=regular
 SCP_PATH=$DOWNLOAD_PATH$(echo $HDFS_DEST|rev |cut -d '/' -f 2 | rev)/
 LOCAL_IP=$(ip a |grep inet| grep -v inet6 | grep -v 127 | cut -d '/' -f1 | cut -d ' ' -f6)
 
@@ -17,6 +21,44 @@ if [ ! -d "$DEPLOY_PATH/logs" ]; then
   mkdir -p $DEPLOY_PATH/logs
 fi
 
+# 取值配置文件
+IFS=': '
+while read k v
+do
+  if [[ "$k" == "hdfs_dest" ]]
+  then
+    HDFS_DEST=$(echo "$v" | sed -r 's/.*"(.+)".*/\1/')
+  fi
+  if [[ "$k" == "download_path" ]]
+  then
+    DOWNLOAD_PATH=$(echo "$v" | sed -r 's/.*"(.+)".*/\1/')
+  fi
+  if [[ "$k" == "job_name" ]]
+  then
+    JOB_NAME=$(echo "$v" | sed -r 's/.*"(.+)".*/\1/')
+  fi
+  if [[ "$k" == "cdh_host_ip" ]]
+  then
+    CDH_HOST_IP=$(echo "$v" | sed -r 's/.*"(.+)".*/\1/')
+  fi
+  if [[ "$k" == "cdh_host_user" ]]
+  then
+    CDH_HOST_USER=$(echo "$v" | sed -r 's/.*"(.+)".*/\1/')
+  fi
+  if [[ "$k" == "cdh_host_password" ]]
+  then
+    CDH_HOST_PASSWORD=$(echo "$v" | sed -r 's/.*"(.+)".*/\1/')
+  fi
+done < $DEPLOY_PATH/conf/$FLINK_TASK_CONF
+
+echo $HDFS_DEST
+echo $DOWNLOAD_PATH
+echo $JOB_NAME
+echo $CDH_HOST_IP
+echo $CDH_HOST_USER
+echo $CDH_HOST_PASSWORD
+
+
 flink run -d -c com.zorkdata.desensitization.TransactionLogDesensitization $DEPLOY_PATH/lib/transaction-log-desensitization-0.1.jar --conf $DEPLOY_PATH/conf/$FLINK_TASK_CONF --regular $DEPLOY_PATH/conf/$REGULAR_TASK_NAME > $DEPLOY_PATH/logs/submit.log &
 
 sleep 10

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

@@ -94,8 +94,8 @@ public class HdfsLogDesensitization implements Serializable {
         this.avroOutputSchema = new Schema.Parser().parse(AvroSchemaDef.ZORK_LOG_SCHEMA).toString(true);
         this.hdfsUri = String.valueOf(conf.get(ConfigConstants.HDFS_URI)).trim();
         this.hdfsUser = String.valueOf(conf.get(ConfigConstants.HDFS_USER)).trim();
-        this.hdfsSrc = String.valueOf(conf.get(ConfigConstants.HDFS_SRC)).trim();
-        this.hdfsDest = String.valueOf(conf.get(ConfigConstants.HDFS_DEST)).trim();
+        this.hdfsSrc = hdfsUri + String.valueOf(conf.get(ConfigConstants.HDFS_SRC)).trim();
+        this.hdfsDest = hdfsUri +  String.valueOf(conf.get(ConfigConstants.HDFS_DEST)).trim();
         this.core = String.valueOf(conf.get(ConfigConstants.CORE)).trim();
         this.startTime = String.valueOf(conf.get(ConfigConstants.START_TIME));
         this.endTime = String.valueOf(conf.get(ConfigConstants.END_TIME));

src/main/resources/application.yml

@@ -5,7 +5,7 @@ source.parallelism: "4"
 transformer.parallelism: "4"
 sink.parallelism: "4"
 
-# 数据来源，支持hdfs和kafka，必传
+# 数据来源，支持hdfs和kafka，必传，暂不支持kafka
 source: "hdfs"
 
 # 交易日志的“核心”信息，值以c开头、后面是数字序号，非必传
@@ -22,10 +22,17 @@ hdfs_uri: "hdfs://cdh-2:8020/"
 # hdfs 用户名
 hdfs_user: "hdfs"
 # hdfs日志源文件地址，若source为hdfs，则该地址必传
-hdfs_src: "hdfs://cdh-2:8020/tmp/datawarehouse4/jzjy/kcbp_biz_log/"
+hdfs_src: "/tmp/datawarehouse4/jzjy/kcbp_biz_log/"
 # hdfs日志写入地址，非必传，默认写到hdfs-src目录下的output目录下
-hdfs_dest: "hdfs://cdh-2:8020/tmp/datawarehouse/jzjy/kcbp_biz_log/output1/"
+hdfs_dest: "/tmp/datawarehouse/jzjy/kcbp_biz_log/output1/"
 
 # 不做脱敏的字段白名单
-fields_white_list: "messid,fundid,custid,orgid,brhid,secuid,bankcode,market,ordersno,ordergroup,count,poststr,stkcode,bsflag,orderamt,price,qty,bankcode,tacode,ofcode,transacc,taacc,indexTime,logchecktime,end_logtime,collecttime,deserializerTime,versioninfo,fmillsecond,smillsecond"
+fields_white_list: "funcid,count1,count2,count3,count4"
 
+# chd下载配置
+cdh_host_ip: "192.168.70.2"
+cdh_host_user: "root"
+cdh_host_password: "NuqUtwbJUBRmUwgh"
+
+# 与配置文件保持一致
+download_path: "/tmp/"

[sonarqube]

SonarQube analysis reported 109 issues

• 🚫 24 critical
• ⚠ 59 major
• 🔽 25 minor
• ℹ 1 info

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. 🚫 Add a default case to this switch. 📘
2. 🚫 switch中每个case需要通过break/return等来终止 📘
3. 🚫 switch块缺少default语句 📘
4. 🚫 Define a constant instead of duplicating this literal " {\n" 11 times. 📘
5. 🚫 [Define a constant instead of duplicating this literal " "type": \n" 11 times. 📘
6. 🚫 Define a constant instead of duplicating this literal " "string",\n" 6 times. 📘
7. 🚫 Define a constant instead of duplicating this literal " "null"\n" 6 times. 📘
8. 🚫 [Define a constant instead of duplicating this literal " ]\n" 11 times.](https://git.zorkdata.com/liaomingtao/transaction_log_desensitization/blob/a4f15ffc1d21175d0353a5bc4088f96f4d7b03d9/src/main/java/com/zorkdata/desensitization/avro/AvroSchemaDef.java#L23) 📘
9. 🚫 Define a constant instead of duplicating this literal " },\n" 9 times. 📘
10. 🚫 Define a constant instead of duplicating this literal " "null",\n" 5 times. 📘
11. 🚫 Define a constant instead of duplicating this literal " {\n" 5 times. 📘
12. 🚫 Define a constant instead of duplicating this literal " "type": "map",\n" 5 times. 📘
13. 🚫 Define a constant instead of duplicating this literal " "values": "string"\n" 3 times. 📘
14. 🚫 Define a constant instead of duplicating this literal " }\n" 5 times. 📘
15. 🚫 Define a constant instead of duplicating this literal "序列化失败" 13 times. 📘
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 22 to the 15 allowed. 📘
19. 🚫 常量【dataFormats】命名应全部大写并以下划线分隔 📘
20. 🚫 Add a nested comment explaining why this method is empty, throw an UnsupportedOperationException or complete the implementation. 📘
21. 🚫 Change this "try" to a try-with-resources. (sonar.java.source not set. Assuming 7 or greater.) 📘
22. 🚫 Refactor this code to not throw exceptions in finally blocks. 📘
23. 🚫 Refactor this code to not throw exceptions in finally blocks. 📘
24. 🚫 Change this "try" to a try-with-resources. (sonar.java.source not set. Assuming 7 or greater.) 📘
25. ⚠ This block of commented-out lines of code should be removed. 📘
26. ⚠ 及时清理不再使用的代码段或配置信息。 📘
27. ⚠ Replace this use of System.out or System.err by a logger. 📘
28. ⚠ Replace this use of System.out or System.err by a logger. 📘
29. ⚠ String contains no format specifiers. 📘
30. ⚠ Replace this use of System.out or System.err by a logger. 📘

• ... 79 more