<dev> 源和目标 hdfs ，两个集群的hdfs配置

d81193c5 · zskycode · b46fc271 · d81193c5 · d81193c5 · d81193c5
Commit d81193c5 authored Mar 27, 2023 by zskycode
5 changed files
--- a/src/main/java/com/zorkdata/desensitization/config/JobConfig.java
+++ b/src/main/java/com/zorkdata/desensitization/config/JobConfig.java
@@ -53,6 +53,8 @@ public class JobConfig implements Serializable {
        }
        keyList = keyList.stream().distinct().collect(Collectors.toList());
        this.setDesensitizationKeyList(keyList);
+        this.hdfsSourcePath = jobInitConfig.getHdfsSourcePath();
+        this.hdfsSinkPath = jobInitConfig.getHdfsSinkPath();
    }

    private String jobName;
@@ -72,4 +74,7 @@ public class JobConfig implements Serializable {
    private long endTimestamp;
    private List<String> desensitizationKeyList;

+    private String hdfsSourcePath;
+    private String hdfsSinkPath;
+
 }
--- a/src/main/java/com/zorkdata/desensitization/config/JobInitConfig.java
+++ b/src/main/java/com/zorkdata/desensitization/config/JobInitConfig.java
@@ -45,6 +45,8 @@ public class JobInitConfig implements Serializable {
        this.mac = MapUtils.getString(conf, ConfigConstants.MAC_KEY);
        this.position = MapUtils.getString(conf, ConfigConstants.POSITION_KEY);
        this.password = MapUtils.getString(conf, ConfigConstants.PASSWORD_KEY);
+        this.hdfsSourcePath =  MapUtils.getString(conf, ConfigConstants.HDFS_SOURCE_PATH);
+        this.hdfsSinkPath =  MapUtils.getString(conf, ConfigConstants.HDFS_SINK_PATH);
    }

    private String jobName;
@@ -61,6 +63,8 @@ public class JobInitConfig implements Serializable {
    private String endTime;
    private long startTimestamp;
    private long endTimestamp;
+    private String hdfsSourcePath;
+    private String hdfsSinkPath;


    private String name;

--- a/src/main/java/com/zorkdata/desensitization/constans/ConfigConstants.java
+++ b/src/main/java/com/zorkdata/desensitization/constans/ConfigConstants.java
@@ -37,6 +37,8 @@ public final class ConfigConstants {
    public static final String MAC_KEY = "mac";
    public static final String POSITION_KEY = "position";
    public static final String PASSWORD_KEY = "password";
+    public static final String HDFS_SOURCE_PATH = "hdfs_source_path";
+    public static final String HDFS_SINK_PATH = "hdfs_sink_path";

    public static final String SERVERS = "servers";
    public static final String ZOOKEEPER = "zookeeper";

--- a/src/main/java/com/zorkdata/desensitization/hadoop/HdfsLogDesensitization.java
+++ b/src/main/java/com/zorkdata/desensitization/hadoop/HdfsLogDesensitization.java
@@ -11,6 +11,7 @@ import org.apache.avro.mapred.AvroInputFormat;
 import org.apache.avro.mapred.AvroKey;
 import org.apache.avro.mapred.AvroOutputFormat;
 import org.apache.avro.mapred.AvroWrapper;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.hadoop.mapred.HadoopInputFormat;
@@ -57,14 +58,27 @@ public class HdfsLogDesensitization implements Serializable {
    public void desensitizationHdfsLog(JobConfig jobConfig) {
        // 初始化flink job env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-        JobConf jobConf = new JobConf();
-        jobConf.set(AVRO_OUTPUT_SCHEMA, jobConfig.getAvroOutputSchema());
+        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.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());
        String logFileListString = list2String(logFiles);
        HadoopInputFormat<Object, Object> hadoopInputFormat = new HadoopInputFormat<>
-                (new AvroInputFormat(), Object.class, Object.class, jobConf);
+                (new AvroInputFormat(), Object.class, Object.class, jobConfInput);
        AvroInputFormat.addInputPaths(hadoopInputFormat.getJobConf(), logFileListString);
        // 2、创建datasource
        DataSource<Tuple2<Object, Object>> hdfsLogInput = env
@@ -75,16 +89,16 @@ public class HdfsLogDesensitization implements Serializable {
        // sink部分
        // 获取目标hdfs的输出目录
        String filePath = jobConfig.getHdfsDest();
-        HadoopOutputFormat hadoopOutputFormat = new HadoopOutputFormat<>(new AvroOutputFormat(), jobConf);
-        FileOutputFormat.setOutputPath(jobConf, new Path(filePath));
+        HadoopOutputFormat hadoopOutputFormat = new HadoopOutputFormat<>(new AvroOutputFormat(), jobConfOutput);
+        FileOutputFormat.setOutputPath(jobConfOutput, new Path(filePath));
        // avro序列化算子(.writeAsText("file:///lmt/output"); 本地写入)
        flatMapOperator.map(new MapFunction<LogData, Tuple2<AvroWrapper<LogData>, NullWritable>>() {
-            @Override
-            public Tuple2<AvroWrapper<LogData>, NullWritable> map(LogData value) throws Exception {
-                AvroKey<LogData> key = new AvroKey<>(value);
-                return new Tuple2<>(key, NullWritable.get());
-            }
-        }).setParallelism(jobConfig.getTransformerParallelism())
+                    @Override
+                    public Tuple2<AvroWrapper<LogData>, NullWritable> map(LogData value) throws Exception {
+                        AvroKey<LogData> key = new AvroKey<>(value);
+                        return new Tuple2<>(key, NullWritable.get());
+                    }
+                }).setParallelism(jobConfig.getTransformerParallelism())
                // .writeAsText("file:///lmt/output", org.apache.flink.core.fs.FileSystem.WriteMode.OVERWRITE)
                .output(hadoopOutputFormat)
                .setParallelism(jobConfig.getSinkParallelism());

--- a/src/main/resources/application.yml
+++ b/src/main/resources/application.yml
@@ -50,6 +50,10 @@ hdfs_src: "/tmp/datawarehouse/net23/"
 # hdfs日志写入地址，非必传，默认写到hdfs-src目录下的output目录下，必须以斜杠结尾
 hdfs_dest: "/tmp/datawarehouse/net23/output14/"

+
+hdfs_source_path: "/root/hadoop/hadoop-conf"
+hdfs_sink_path: "/root/hadoop/ambari"
+
 # cdh下载配置
 # cdh能执行hdfs命令的机器的ip
 cdh_host_ip: "192.168.70.2"