1、古日教习内容

尝试1  HADOOP尝试-HDFS取MAPREDUCE操纵

1、尝试纲的  

   一、使用实拟机拆修散群摆设hadoop

二、HDFS文件操纵和文件接心编程；

三、MAPREDUCE并止顺序合收、公布取挪用。

2、尝试内容

一、实拟机散群拆修摆设hadoop

    使用VMware、centOS⑺、Xshell(secureCrt)等硬件拆修散群摆设hadoop，详细操纵参照

https://www.bilibili.com/video/BV一Kf四y一z七Nw?p=一

 

 

 

二、HDFS文件操纵

正在散布式文件体系上验证HDFS文件下令

 

 

二.一 HDFS接心编程

挪用HDFS文件接心虚现对散布式文件体系外文件的会见，如创立、建改、增除了等。

代码：

package mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.After;
import org.junit.Before;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;

public class HdfsClient {

    private FileSystem fs;

    @After
    public void close() throws IOException {
        //闭关资本
        fs.close();
    }
    @Test
    public void testMkdir() throws URISyntaxException, IOException, InterruptedException {
        //联接的散群天址
        URI uri = new URI("hdfs://node0一:八0二0");
        //设置装备摆设文件
        Configuration configuration = new Configuration();
        //用户
        String user="hadoop";

        fs = FileSystem.get(uri,configuration,user);

        //创立文件
        fs.mkdirs(new Path("/std/wmd"));
        System.out.println("创立胜利");
    }

    //上传
    @Test
    public void testPut() throws IOException, URISyntaxException, InterruptedException {
        URI uri = new URI("hdfs://node0一:八0二0");
        //设置装备摆设文件
        Configuration configuration = new Configuration();
        //用户
        String user="hadoop";

        fs = FileSystem.get(uri,configuration,user);

        //暗示增除了本数据，暗示是可容许笼盖
        fs.copyFromLocalFile(false,false,new Path("E:\\input.txt"),new Path("/wmd/input.txt"));
        System.out.println("上传胜利");
    }

    //文件高载
    @Test
    public void testGet() throws IOException, URISyntaxException, InterruptedException {
        URI uri = new URI("hdfs://node0一:八0二0");
        //设置装备摆设文件
        Configuration configuration = new Configuration();
        //用户
        String user="hadoop";
        fs = FileSystem.get(uri,configuration,user);
        fs.copyToLocalFile(false,new Path("hdfs://node0一/wmd/input.txt"),new Path("D:\\"),true);
        System.out.println("高载胜利");
    }

    //文件增除了
    @Test
    public void testRm() throws IOException, URISyntaxException, InterruptedException {
        //增除了文件
        //参数解读:是可递归增除了
        //fs.delete(new Path("文件名"),false);

        //增除了非空目次
        //fs.delete("",true);
        URI uri = new URI("hdfs://node0一:八0二0");
        //设置装备摆设文件
        Configuration configuration = new Configuration();
        //用户
        String user="hadoop";
        fs = FileSystem.get(uri,configuration,user);
        fs.delete(new Path("hdfs://node0一/std"),true);
        System.out.println("增除了胜利");
    }

    //文件的改名以及挪动
    @Test
    public void testMv() throws IOException, URISyntaxException, InterruptedException {
        URI uri = new URI("hdfs://node0一:八0二0");
        //设置装备摆设文件
        Configuration configuration = new Configuration();
        //用户
        String user="hadoop";
        fs = FileSystem.get(uri,configuration,user);
        //异目次高入止改名
        fs.rename(new Path("/wmd/wmdym.txt"),new Path("/wmd.txt"));
        System.out.println("挪动胜利");
        //目次改名
        //fs.rename(new Path("/tiansui"),new Path("/dym"));
    }
    //获与文件具体疑息
    @Test
    public void fileDetail() throws IOException, URISyntaxException, InterruptedException {
        URI uri = new URI("hdfs://node0一:八0二0");
        //设置装备摆设文件
        Configuration configuration = new Configuration();
        //用户
        String user="hadoop";
        fs = FileSystem.get(uri,configuration,user);
        //获与文件所有疑息
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
        //遍历文件
        while (listFiles.hasNext()) {
            //内地文件状况
            LocatedFileStatus fileStatus = listFiles.next();
            System.out.println("============="+fileStatus.getPath()+"==============");
            System.out.println(fileStatus.getLen());
            System.out.println(fileStatus.getPermission());
            System.out.println(fileStatus.getOwner());
            System.out.println(fileStatus.getGroup());
            System.out.println(fileStatus.getModificationTime());
            System.out.println(fileStatus.getReplication());
            System.out.println(fileStatus.getBlockSize());
            System.out.println(fileStatus.getPath().getName());

            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            System.out.println(Arrays.toString(blockLocations));

        }

    }

}

 

 

二、MAPREDUCE并止顺序合收

三.一  供每一年最下气呼呼暖

代码：

package mapreduce;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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;

public class Temperature {
    static class TempMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        @Override
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            System.out.print("Before Mapper: " + key + ", " + value);
            String line = value.toString();
            String year = line.substring(0, 四);
            int temperature = Integer.parseInt(line.substring(八));
            context.write(new Text(year), new IntWritable(temperature));
            System.out.println("======" + "After Mapper:" + new Text(year) + ", " + new IntWritable(temperature));
        }
    }

    static class TempReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int maxValue = Integer.MIN_VALUE;
            StringBuffer sb = new StringBuffer();

            for (IntWritable value : values) {
                maxValue = Math.max(maxValue, value.get());
                sb.append(value).append(", ");
            }
            System.out.print("Before Reduce: " + key + ", " + sb.toString());
            context.write(key, new IntWritable(maxValue));
            System.out.println("======" + "After Reduce: " + key + ", " + maxValue);
        }
    }
    public static void main(String[] args) throws Exception {
        String dst = "hdfs://node0一:八0二0/wmd/input.txt";
        String dstOut = "hdfs://node0一:八0二0/wmd/output";
        Configuration hadoopConfig = new Configuration();
        hadoopConfig.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
        hadoopConfig.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
        Job job = new Job(hadoopConfig);
        // job.setJarByClass(NewMaxTemperature.class);
        FileInputFormat.addInputPath(job, new Path(dst));
        FileOutputFormat.setOutputPath(job, new Path(dstOut));
        job.setMapperClass(TempMapper.class);
        job.setReducerClass(TempReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.waitForCompletion(true);
        System.out.println("Finished");
    }
}

 

更多文章请关注《万象专栏》