支持“核心”参数

打包逻辑优化

支持“核心”参数
打包逻辑优化
eff193d3 · 王海鹰 · f32f595c · eff193d3 · eff193d3 · eff193d3
Commit eff193d3 authored Oct 22, 2020 by 王海鹰
6 changed files
--- a/pom.xml
+++ b/pom.xml
@@ -100,19 +100,16 @@ under the License.
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
-            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-shaded-hadoop-2</artifactId>
            <version>2.6.5-10.0</version>
-            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
-            <scope>provided</scope>
        </dependency>

        <dependency>

--- a/src/main/java/com/zorkdata/datamask/TransactionLogMask.java
+++ b/src/main/java/com/zorkdata/datamask/TransactionLogMask.java
@@ -2,8 +2,8 @@ package com.zorkdata.datamask;

 import com.alibaba.fastjson.JSON;
 import com.zorkdata.datamask.constant.ParamConstants;
-import com.zorkdata.datamask.hadoop.HadoopMask;
-import com.zorkdata.datamask.kafka.KafkaMask;
+import com.zorkdata.datamask.hadoop.HdfsLogMaskUtil;
+import com.zorkdata.datamask.kafka.KafkaMsgMaskUtil;
 import com.zorkdata.datamask.util.ZorkParameterUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -32,9 +32,9 @@ public class TransactionLogMask {
        String source = conf.get(ParamConstants.SOURCE);

        if (ParamConstants.HDFS.equals(source)) {
-            HadoopMask.maskHdfsLog(conf);
+            HdfsLogMaskUtil.maskHdfsLog(conf);
        } else if (ParamConstants.KAFKA.equals(source)) {
-            KafkaMask.maskKafkaMsg(conf);
+            KafkaMsgMaskUtil.maskKafkaMsg(conf);
        }
    }


--- a/src/main/java/com/zorkdata/datamask/hadoop/HadoopMask.java
+++ b/src/main/java/com/zorkdata/datamask/hadoop/HadoopMask.java
@@ -49,8 +49,8 @@ import java.util.Map;
 * @author: wanghaiying (<a href="wanghaiying@zorkdata.com.cn">wanghaiying@zorkdata.com.cn</a>)
 * Date: Create in 2020/9/23 9:30
 */
-public class HadoopMask {
-    private static final Logger logger = LoggerFactory.getLogger(HadoopMask.class);
+public class HdfsLogMaskUtil {
+    private static final Logger logger = LoggerFactory.getLogger(HdfsLogMaskUtil.class);

    /**
     * hdfs日志文件脱敏
@@ -73,6 +73,8 @@ public class HadoopMask {
        ArrayList< String> fieldsWhiteList = new ArrayList<String>(fieldsWhiteListArray.length);
        Collections.addAll(fieldsWhiteList, fieldsWhiteListArray);

+        String core = String.valueOf(conf.get(ParamConstants.CORE)).trim();
+
        List<String> logFiles = filterHdfsLogFiles(hdfsLogQueryParam.getHdfsSrc(), hdfsLogQueryParam.getDate(),
                hdfsLogQueryParam.getStartTime(), hdfsLogQueryParam.getEndTime());

@@ -96,7 +98,7 @@ public class HadoopMask {
                                    new TypeReference<LogData>() {
                                    });
                            //根据日志事件的核心信息做过滤
-                            if (null != hdfsLogQueryParam.getCore() && logData.getDimensions().get("hostname").indexOf("c9") > -1 ) {
+                            if (null != hdfsLogQueryParam.getCore() && logData.getDimensions().get("hostname").indexOf(core) > -1 ) {
                                //根据日志事件的timestamp做过滤
                                Long timestamp = DateUtils.utc2timestamp(logData.getTimestamp());
                                boolean flag = null != timestamp && timestamp > hdfsLogQueryParam.getStartTime()

--- a/src/main/java/com/zorkdata/datamask/kafka/KafkaMask.java
+++ b/src/main/java/com/zorkdata/datamask/kafka/KafkaMask.java
@@ -21,7 +21,7 @@ import java.util.Properties;
 * @Email xiesen310@163.com
 * @Date 2020/10/21 14:51
 */
-public class KafkaMask {
+public class KafkaMsgMaskUtil {
    /**
     * kafka消息数据脱敏
     *

--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -5,7 +5,7 @@ source: "hdfs"
 hdfs_src: "hdfs://cdh-2:8020/tmp/datawarehouse4/jzjy/kcbp_biz_log"

 # hdfs日志写入地址，非必传，默认写到hdfs-src目录下的output目录下
-hdfs_dest: "hdfs://cdh-2:8020/tmp/datawarehouse/jzjy/kcbp_biz_log/output5/"
+hdfs_dest: "hdfs://cdh-2:8020/tmp/datawarehouse/jzjy/kcbp_biz_log/output7/"

 # 脱敏结果下载到的本地路径
 download_path: "/tmp"
@@ -24,7 +24,7 @@ end_time: 1601348850000

 # 不做脱敏的字段白名单
 fieldsWhiteList: "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"
+orderamt,price,qty,bankcode,tacode,ofcode,transacc,taacc,indexTime,logchecktime,end_logtime,collecttime,deserializerTime,versioninfo,fmillsecond,smillsecond"

 # 脱敏用的正则表达式
 reg_exp:

--- a/src/main/resources/bin/start.sh
+++ b/src/main/resources/bin/start.sh
@@ -6,8 +6,7 @@ if [ ! -d "$deploy_path/logs" ]; then
  mkdir -p $deploy_path/logs
 fi

-FLINK_TASK_CONF=gmas-config.yaml
+FLINK_TASK_CONF=application.yml
 parallelism=1

-flink run -p $parallelism $deploy_path/lib/transactionLogMask-0.1.jar
-$$deploy_path/conf/$FLINK_TASK_CONF> $$deploy_path/logs/transactionLogMask-0.1.log &
\ No newline at end of file
+flink run  $deploy_path/lib/transactionLogMask-0.1.jar --conf $deploy_path/conf/$FLINK_TASK_CONF> $deploy_path/logs/transactionLogMask-0.1.log &
\ No newline at end of file