hadoop初步介绍：hdfs分布存储+ mr分布计算

朱雀 2022-05-11 13:00 112阅读 0赞

1.  hdfs 和RDBMS区别
2.  mr 和 网格计算，志愿计算

# 1，数据存储 #

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>磁盘存储</th> 
   <th>解决分布式问题</th> 
   <th>硬件需求</th> 
   <th>系统瓶颈</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>hdfs</td> 
   <td>磁盘阵列-集群</td> 
   <td>硬件故障，多数据源的数据准确性</td> 
   <td>普通机</td> 
   <td>数据传输：硬盘带宽</td> 
  </tr> 
  <tr> 
   <td>RDBMS</td> 
   <td>单磁盘</td> 
   <td></td> 
   <td>专业服务器</td> 
   <td>磁盘寻址：大量数据更新</td> 
  </tr> 
 </tbody> 
</table>

# 2，分析计算 #

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>适用场</th> 
   <th>特点</th> 
   <th>生态圈</th> 
   <th>结构特点</th> 
   <th>数据完整性</th> 
   <th>可扩展性</th> 
   <th>数据集结构化程度</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>mr</td> 
   <td>PB级数据：批处理</td> 
   <td>一写多读</td> 
   <td>yarn集成其他分布式程序,hive,saprk</td> 
   <td>读模式</td> 
   <td>低</td> 
   <td>高</td> 
   <td>半、非结构化</td> 
  </tr> 
  <tr> 
   <td>RDBMS</td> 
   <td>GB级数据：实时检索，更新</td> 
   <td>持续更新</td> 
   <td></td> 
   <td>写模式</td> 
   <td>高</td> 
   <td>低</td> 
   <td>结构化</td> 
  </tr> 
 </tbody> 
</table>

# 3，网格计算，志愿计算 #

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>特点</th> 
   <th>适用场景</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>网格计算</td> 
   <td>分散节点计算+ 网络共享文件系统</td> 
   <td>小规模数据：无网络传输瓶颈</td> 
  </tr> 
  <tr> 
   <td>网格计算</td> 
   <td>任务单元化+ 分散计算+ 校验结果</td> 
   <td>cup密集型：计算时间&gt;传输时间</td> 
  </tr> 
  <tr> 
   <td>mr</td> 
   <td>转移计算+ 数据本地化</td> 
   <td>作业周期短（小时计）,高速局域网内，高配硬件</td> 
  </tr> 
 </tbody> 
</table>

# 4，mr 对比linux:awk流处理 #

## 4.1，awk处理： 年度最高温度统计 ##

![在这里插入图片描述][70]

## 4. 2，mapreduce处理：每年最高温度统计 ##

![在这里插入图片描述][70 1]

### a, ruby写mapreduce: ###

![在这里插入图片描述][70 2]

### b, java 写mapreduce ====>idea +maven: 添加依赖 ###

<dependency>
                <groupId>org.apache.hadoop</groupId>
                <artifactId>hadoop-client</artifactId>
                <version>2.7.3</version>
            </dependency>

map方法

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    import java.io.IOException;
    
    public class Map1 extends Mapper<LongWritable, Text, IntWritable,IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //整理的数据输入：
            //1982,-8
            //1931,-4
            String str = value.toString();
            String[] arr = str.split(",");
            int year=0, tmp=Integer.MIN_VALUE;
    
            //数据转换
            try {
                 year= Integer.parseInt(arr[0]);
                 tmp= Integer.parseInt(arr[1]);
            }catch (Exception e){
                e.printStackTrace();
            }
            //输出：新数据
            context.write(new IntWritable(year),new IntWritable(tmp));
        }
    }

reduce方法

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.mapreduce.Reducer;
    import java.io.IOException;
    import java.util.Iterator;
    
    public class Reduce1 extends Reducer<IntWritable,IntWritable,IntWritable,IntWritable> {
        @Override
        protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            //输入数据：1931,【-4,23,4,35,6】
            //聚合数据： 求每组数据中的max(tmp)
            int max=Integer.MIN_VALUE;
            Iterator<IntWritable> it = values.iterator();
            while (it.hasNext()){
                IntWritable next = it.next();
                int tmp = next.get();
    
                max= (max >tmp) ? max:tmp;
            }
            //输出： 最高温度
            context.write(key, new IntWritable(max));
        }
    }

app类： 调度组织job

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import java.io.IOException;
    
    public class App1 {
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
            job.setJarByClass(App1.class);
            job.setJobName("maxTmp");
    
            //map,reduce
            job.setMapperClass(Map1.class);
            job.setReducerClass(Reduce1.class);
    
            job.setMapOutputKeyClass(IntWritable.class);
            job.setMapOutputValueClass(IntWritable.class);
            job.setOutputKeyClass(IntWritable.class);
            job.setOutputValueClass(IntWritable.class);
    
            //map端： 预聚合
            job.setCombinerClass(Reduce1.class);
            job.setNumReduceTasks(3);
            //输入输出
            FileInputFormat.addInputPath(job,new Path("/home/wang/txt/tmp.txt"));
            //删除已存在的目录,以防报错
            FileSystem fs = FileSystem.get(conf);
            Path outPath = new Path("/home/wang/tmp-out");
            if(fs.exists(outPath))
                fs.delete(outPath,true);
            FileOutputFormat.setOutputPath(job,outPath);
    
            //提交等待
            job.waitForCompletion(true);
        }
    }

[70]: /images/20220505/e7ce20a19a7c4823822a6f78b1cab3ee.png
[70 1]: /images/20220505/c9150b5201a34b088d89177ceac9a87e.png
[70 2]: /images/20220505/5600b1e90f23444f8e4118034a1aeeab.png