学习大数据：Hadoop

学习大数据：Hadoop

Hadoop的架构在其核心，Hadoop主要有两个层次，即：

加工/计算层(MapReduce)存储层(Hadoop分布式文件系统)除了上面提到的两个核心组件，Hadoop的框架还包括以下两个模块：

Hadoop通用：这是Java库和其他Hadoop组件所需的实用工具Hadoop YARN ：这是作业调度和集群资源管理的框架Hadoop Streaming 是一个实用程序，它允许用户使用任何可执行文件（例如shell实用程序）作为映射器和/或reducer创建和运行作业。

HDFS:HDFS遵循主从架构，它具有以下元素。

1、名称节点 -Namenode

名称节点是包含GNU/Linux操作系统和软件名称节点的普通硬件。它是一个可以在商品硬件上运行的软件。具有名称节点系统作为主服务器，它执行以下任务：

管理文件系统命名空间。 规范客户端对文件的访问。 它也执行文件系统操作，如重命名，关闭和打开的文件和目录。

2、数据节点 - Datanode

Datanode具有GNU/Linux操作系统和软件Datanode的普通硬件。对于集群中的每个节点(普通硬件/系统)，有一个数据节点。这些节点管理数据存储在它们的系统。

数据节点上的文件系统执行的读写操作，根据客户的请求。 还根据名称节点的指令执行操作，如块的创建，删除和复制。

3、块 -block

一般用户数据存储在HDFS文件。在一个文件系统中的文件将被划分为一个或多个段和/或存储在个人数据的节点。这些文件段被称为块。换句话说，数据的HDFS可以读取或写入的最小量被称为一个块。缺省的块大小为64MB，但它可以增加按需要在HDFS配置来改变

HDFS常用命令运行jar包：hadoop jar /Users/kexin/work/projects/Hadoop/target/hadoop-0.0.1-SNAPSHOT.jar com.kexin.hadoop.units.WordCount /test/test.txt /project/wordcount/output文件系统操作：hadoop fs -cat|ls|mkdir上传文件：hadoop dfs -put ./testdata.txt /testdata递归删除目录及文件：hadoop fs -rmr /testdata删除文件：hadoop fs -rm /testdata.txtMapReduce教程：

https://cnblogs.com/huxinga/p/6939896.html static class TokenizerMapper extends Mapper { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } public static class IntSumReducer extends Reducer { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } }

public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); }

在idea中远程调试程序

System.setProperty("hadoop.home.dir", "/Users/kexin/work/app/hadoop/hadoop-2.6.5");

Configuration conf = new Configuration(); String uri = "hdfs://localhost:9000"; Job job = null; try { job = Job.getInstance(conf); } catch (IOException e) { e.printStackTrace(); } job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileSystem fs = FileSystem.get(URI.create(uri), conf); try { FileInputFormat.addInputPath(job, new Path("hdfs://localhost:9000/test/test.txt")); Path outpath = new Path("hdfs://localhost:9000/project/wordcount/output"); if (fs.exists(outpath)) { fs.delete(outpath, true); } FileOutputFormat.setOutputPath(job, outpath); } catch (IllegalArgumentException | IOException e) { e.printStackTrace(); } try { job.submit(); } catch (ClassNotFoundException | IOException | InterruptedException e) { e.printStackTrace(); }

在idea中本地调试程序

System.setProperty("hadoop.home.dir", "/Users/kexin/work/app/hadoop/hadoop-2.6.5");

Configuration config = new Configuration(); try { FileSystem fs = FileSystem.get(config); Job job = Job.getInstance(config); job.setJarByClass(WordCount.class); job.setJobName("word count"); job.setMapperClass(TokenizerMapper.class); job.setReducerClass(IntSumReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path("/Users/kexin/work/projects/Hadoop/src/main/resources/input")); Path outpath = new Path("/Users/kexin/work/projects/Hadoop/src/main/resources/output"); if (fs.exists(outpath)) { fs.delete(outpath, true); } FileOutputFormat.setOutputPath(job, outpath); boolean f = job.waitForCompletion(true); if (f) { System.out.println("job任务执行成功"); } } catch (Exception e) { e.printStackTrace(); }

1、映射器映射器将输入k/v对映射到一组中间k/v对。转换后的中间记录不需要与输入记录的类型相同。给定的输入对可以映射到零个或多个输出对。通过调用context.write（WritableComparable，Writable）来收集输出对。

Hadoop MapReduce框架为作业的InputFormat生成的每个InputSplit生成一个map任务。

总的来说，映射器实现通过Job.setMapperClass（Class）方法传递给作业。然后，框架为InputSplit中的每个k/v对调用该任务的map。

映射的数量通常由输入的总大小驱动，即输入文件的块总数。也可以使用Configuration.set（MRJobConfig.NUM_MAPS，int）来设置映射数量。

随后将与给定输出键关联的所有中间值按框架分组，并传递给Reducer以确定最终输出。用户可以通过Job.setGroupingComparatorClass（Class）指定Comparator来控制分组。

对Mapper输出进行排序，然后根据Reducer进行分区。分区总数与作业的reduce任务数相同。用户可以通过实现自定义分区程序来控制哪些键（以及记录）转到哪个Reducer。

用户可以选择通过Job.setCombinerClass（Class）指定组合器来执行中间输出的本地聚合，比如合并重复的key，这有助于减少从Mapper传输到Reducer的数据量。

2、ReducerReducer有3个主要阶段：shuffle，sort和reduce

1、shuffle

Reducer的输入是映射器的排序输出。在此阶段，框架通过HTTP获取所有映射器的输出的相关分区

2、sort

框架在此阶段按键（因为不同的映射器可能输出相同的键）对Reducer输入进行分组。在获取map输出结果时，shuffle和sort阶段同时进行。

如果要求对中间密钥进行分组的等价规则与在减少之前对密钥进行分组的等价规则不同，则可以通过Job.setSortComparatorClass（Class）指定比较器。由于Job.setGroupingComparatorClass（Class）可用于控制中间键的分组方式，因此可以结合使用这些键来模拟值的二级排序。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。