<dev> 修改配置文件加载

5a759320 · zskycode · d81193c5 · 5a759320
Commit 5a759320 authored Apr 06, 2023 by zskycode
Show whitespace changes
Inline Side-by-side

Showing with 19 additions and 11 deletions

src/main/java/com/zorkdata/desensitization/hadoop/HdfsLogDesensitization.java ...rkdata/desensitization/hadoop/HdfsLogDesensitization.java +19 -11

No files found.
--- a/src/main/java/com/zorkdata/desensitization/hadoop/HdfsLogDesensitization.java
+++ b/src/main/java/com/zorkdata/desensitization/hadoop/HdfsLogDesensitization.java
@@ -55,24 +55,26 @@ public class HdfsLogDesensitization implements Serializable {
        desensitizationHdfsLog(this.jobConfig);
    }

+    public JobConf config(String jobName, String path) {
+        JobConf conf = new JobConf(HdfsLogDesensitization.class);
+        conf.setJobName(jobName);
+        conf.addResource("classpath:" + path + "/core-site.xml");
+        conf.addResource("classpath:" + path + "/hdfs-site.xml");
+///        conf.addResource("classpath:" + path + "/mapred-site.xml");
+        log.info("配置文件加载:"+"classpath:" + path + "/core-site.xml");
+        log.info("配置文件加载:"+"classpath:" + path + "/hdfs-site.xml");
+        return conf;
+    }
+
    public void desensitizationHdfsLog(JobConfig jobConfig) {
        // 初始化flink job env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        JobConf jobConfInput = new JobConf();
        if (!StringUtils.isEmpty(jobConfig.getHdfsSourcePath())) {
-            jobConfInput.addResource(new Path("file://" + jobConfig.getHdfsSourcePath() + "/core-site.xml"));
-            jobConfInput.addResource(new Path("file://" + jobConfig.getHdfsSourcePath() + "/hdfs-site.xml"));
-///        jobConfInput.addResource(new Path("file://" + jobConfig.getHdfsSourcePath() + "/mapred-site.xml"));
-        }
-
-        JobConf jobConfOutput = new JobConf();
-        if (!StringUtils.isEmpty(jobConfig.getHdfsSinkPath())) {
-            jobConfOutput.addResource(new Path("file://" + jobConfig.getHdfsSinkPath() + "/core-site.xml"));
-            jobConfOutput.addResource(new Path("file://" + jobConfig.getHdfsSinkPath() + "/hdfs-site.xml"));
-///        jobConfInput.addResource(new Path("file://" + jobConfig.getHdfsSinkPath() + "/mapred-site.xml"));
+            jobConfInput = config("jobConfInput", jobConfig.getHdfsSourcePath());
        }
        jobConfInput.set(AVRO_OUTPUT_SCHEMA, jobConfig.getAvroOutputSchema());
-        jobConfOutput.set(AVRO_OUTPUT_SCHEMA, jobConfig.getAvroOutputSchema());
+
        // source部分
        // 1、通过时间获取文件夹信息
        List<String> logFiles = filterHdfsLogFiles(jobConfig.getHdfsSrc(), jobConfig.getHdfsUri(), jobConfig.getHdfsUser());
@@ -88,6 +90,12 @@ public class HdfsLogDesensitization implements Serializable {
                hdfsLogInput.flatMap(new DesensitizationFunction<Tuple2<Object, Object>, LogData>(jobConfig));
        // sink部分
        // 获取目标hdfs的输出目录
+
+        JobConf jobConfOutput = new JobConf();
+        if (!StringUtils.isEmpty(jobConfig.getHdfsSinkPath())) {
+            jobConfOutput = config("jobConfOutput", jobConfig.getHdfsSinkPath());
+        }
+        jobConfOutput.set(AVRO_OUTPUT_SCHEMA, jobConfig.getAvroOutputSchema());
        String filePath = jobConfig.getHdfsDest();
        HadoopOutputFormat hadoopOutputFormat = new HadoopOutputFormat<>(new AvroOutputFormat(), jobConfOutput);
        FileOutputFormat.setOutputPath(jobConfOutput, new Path(filePath));