hadoop编程实践 - “清洗”（根据具体需求）

Posted 2022-09-28 floakss

项目文件：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);