hadoop编程实践 – “清洗”(根据具体需求)

hadoop编程实践 - “清洗”(根据具体需求)
强烈推介IDEA2021.1.3破解激活,IntelliJ IDEA 注册码,2021.1.3IDEA 激活码  

大家好,我是架构君,一个会写代码吟诗的架构师。今天说一说hadoop编程实践 - “清洗”(根据具体需求),希望能够帮助大家进步!!!

项目文件:Github

需求一:

package test.dataclean;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/*
 * @ author:Kouch
 * 
 *  “清洗”思路:
 *      1 map: 获取的一行数据;判断一行字符串长度;
 *      2 reduce:
 * 
 *  注:结合具体需求;
 */

public class DataHandle1 {
    
    //map
    public static class Map extends Mapper<Object,Text,Text,Text>{
        
        private static Text line=new Text();
        public void map(Object key,Text value,Context context) throws IOException, InterruptedException {
            line=value;
            //测试
            System.out.println("内容:"+line);
            
            //一行字符串长度;
            String str=line.toString();
            //System.out.println("zhuan:"+str);
            
            if(str.length()>20) {
                context.write(line, new Text(""));
            }
        }
    }
    
    //reduce
    public static class Reduce extends Reducer<Text,Text,Text,Text>{
        public void reduce(Text key,Iterable<Text>values,Context context) throws IOException, InterruptedException {
            
            //测试
            //System.out.println("内容:"+key);
            context.write(key, new Text(""));
        }
    }
    
    
    //main
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        
        //配置类
        Configuration conf=new Configuration();
        conf.set("mapred.job.tracker", "localhost:9000");
        
        //获取传参
        //方式一:
        String[] ioArgs=new String[] {
   "input/dailydata1.txt","out"};
        String[] otherArgs=new GenericOptionsParser(conf,ioArgs).getRemainingArgs();
        if(otherArgs.length!=2) {
            System.err.println("Usage:Data Clean <in> <out> - path?");
            System.exit(2);
        }
        
        //判断输出文件是否存在;存在-删除;
        String url="hdfs://localhost:9000/user/kouch/"+ioArgs[1];
        FileSystem fs=FileSystem.get(URI.create(url), conf);
        if(fs.delete(new Path(url), true)) {
   //true:文件夹下所有文件;false:如果此文件存在其他文件就不删除
            System.out.println("删除"+url);
        }
        
        //Job设置
        Job job=Job.getInstance();
        job.setJarByClass(Deduplication.class);
        job.setMapperClass(Map.class);
        job.setCombinerClass(Reduce.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        //设置输入输出目录
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        
        //等待job完成之后再返回结果并退出程序
        System.exit(job.waitForCompletion(true)?0:1);
        
    }
    
    
}

 

需求二:

package test.dataclean;

import java.io.IOException;
import java.net.URI;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

/*
 * @ author:Kouch
 * 
 *  “清洗”思路:
 *      1 map: 获取的一行数据;去除错误数据;截取有效字段;输入context;
 *      2 reduce:
 * 
 *  注:结合具体需求;
 *  
 *  统计:get/post/head 请求;
 */

public class DataHandle2 {
    
    //map
    public static class Map extends Mapper<Object,Text,Text,IntWritable>{
        private static final IntWritable one = new IntWritable(1);
        private static Text line=new Text();
        public void map(Object key,Text value,Mapper<Object, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            line=value;
            //测试
            //System.out.println("内容:"+line);
            String str=line.toString();
            
            if(!(str.indexOf("%")>0)) {
                //System.out.println("内容:"+line);
                String[] strs=str.split("\"");
                //System.out.println("内容:"+strs[1]);
                String need=strs[1];
                
                if(need.startsWith("G")) {
                    //System.out.println("G");
                    context.write(new Text("Get"), one);
                }else if(need.startsWith("H")) {
                    //System.out.println("H");
                    context.write(new Text("Head"), one);
                }else if(need.startsWith("P")){
                    //System.out.println("P");
                    context.write(new Text("Post"), one);
                }else {
                    
                }
            }
            
        }
    }
    
    //reduce
    public static class Reduce extends Reducer<Text,IntWritable,Text,IntWritable>{
        
        private IntWritable result = new IntWritable();
        
        public void reduce(Text key,Iterable<IntWritable>values,Context context) throws IOException, InterruptedException {
            
            //测试
            //System.out.println("内容:"+key);
            
            int sum=0;
            
            //迭代累计频率;
            IntWritable val;
            for(Iterator i$ = values.iterator(); i$.hasNext(); sum += val.get()) {
                val = (IntWritable)i$.next();
            }

            this.result.set(sum);
            context.write(key, this.result);
        }
    }
    
    
    //main
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        
        //配置类
        Configuration conf=new Configuration();
        conf.set("mapred.job.tracker", "localhost:9000");
        
        //获取传参
        //方式一:
        String[] ioArgs=new String[] {
   "input/daya2.txt","out3"};
        String[] otherArgs=new GenericOptionsParser(conf,ioArgs).getRemainingArgs();
        if(otherArgs.length!=2) {
            System.err.println("Usage:Data Clean <in> <out> - path?");
            System.exit(2);
        }
        
        //判断输出文件是否存在;存在-删除;
        String url="hdfs://localhost:9000/user/kouch/"+ioArgs[1];
        FileSystem fs=FileSystem.get(URI.create(url), conf);
        if(fs.delete(new Path(url), true)) {
   //true:文件夹下所有文件;false:如果此文件存在其他文件就不删除
            System.out.println("删除"+url);
        }
        
        //Job设置
        Job job=Job.getInstance();
        job.setJarByClass(Deduplication.class);
        job.setMapperClass(Map.class);
        job.setCombinerClass(Reduce.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        //设置输入输出目录
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        
        //等待job完成之后再返回结果并退出程序
        System.exit(job.waitForCompletion(true)?0:1);
        
    }
    
    
}

 

转载于:https://www.cnblogs.com/floakss/p/11455797.html

本文来源weixin_30595035,由架构君转载发布,观点不代表Java架构师必看的立场,转载请标明来源出处:https://javajgs.com/archives/29340

发表评论