泾川县建设局网站,html5的优点,wordpress 京东客,营销方案的几个要素Yarn架构体系 主从架构
也是采用 master#xff08;Resource Manager#xff09;- slave #xff08;Node Manager#xff09;架构#xff0c;Resource Manager 整个集群只有一个#xff0c;一个可靠的节点。
1、 每个节点上可以负责该节点上的资源管理以及任务调度Resource Manager- slave Node Manager架构Resource Manager 整个集群只有一个一个可靠的节点。
1、 每个节点上可以负责该节点上的资源管理以及任务调度Node Manager 会定时向Resource Manager汇报本节点上 的资源使用情况和任务运行状态 2、 Resource Manager会通过心跳应答的机制向Node Manager下达命令或者分发新的任务 3、 Yarn 将某一资源分配给该应用程序后应用程序会启动一个Application Master 4、 Application Master为应用程序负责向Resource Manager申请资源申请资源之后再和申请到的节点进行通信运行内部任务。
Resource Manager
RM是一个全局的资源管理器负责整个系统的资源管理和分配。它主要由两个组件构成调度器Scheduler和应用程序管理器Applications ManagerASM。
Schedule 资源调度器是一个可插拔的组件用户可根据自己需要设计资新的源调度器YARN提供多个可直接使用的资源调度器。资源调度器将系统中的资源分配给正在运行的程序不负责监控或跟踪应用的执行状态不负责重启失败的任务。
Applications Manager 应用程序管理器负责管理整个系统中所有应用程序包括应用程序提交、与调度器协商资源以启动ApplicationMaster、监控ApplicationMaster运行状态并在失败时重新启动它等。
Node Manager
NM是每个节点上的资源和任务管理器一方面它会定时地向RM汇报本节点上的资源使用情况和各个Container的运行状态另一方面它接收并处理来自AM的Container启动/停止等各种请求。
ApplicationMaster
用户提交的每个应用程序均包含一个AM主要功能包括
1.与RM调度器协商以获取资源用Container表示
2.将得到的任务进一步分配给内部的任务
3.与NM通信以启动/停止任务
4.监控所有任务运行状态并在任务运行失败时重新为任务申请资源以重启任务。
注RM只负责监控AM在AM运行失败时候启动它RM并不负责AM内部任务的容错这由AM来完成。
Container
Container是YARN中的资源抽象它封装了某个节点上的多维度资源如内存、CPU、磁盘、网络等当AM向RM申请资源时RM为AM返回的资源便是用Container表示。YARN会为每个任务分配一个Container且该任务只能使用该Container中描述的资源。yarn的container容器是yarn虚拟出来的一个东西,属于虚拟化的,它是由memoryvcore组成,是专门用来运行任务的 Yarn的安装
etc/hadoop/目录下 yarn-site.xml文件 cd /opt/apps/hadoop-3.1.1/etc/hadoop/vi yarn-site.xml!-- resource,manager主节点所在机器 --
propertynameyarn.resourcemanager.hostname/namevaluelinux01/value
/property!-- 为mr程序提供shuffle服务 --
propertynameyarn.nodemanager.aux-services/namevaluemapreduce_shuffle/value
/property!-- 一台NodeManager的总可用内存资源 --
propertynameyarn.nodemanager.resource.memory-mb/namevalue4096/value
/property
!-- 一台NodeManager的总可用逻辑cpu核数 --
propertynameyarn.nodemanager.resource.cpu-vcores/namevalue4/value
/property!-- 是否检查容器的虚拟内存使用超标情况vmem为true 指的是默认检查虚拟内存,容器使用的虚拟内存不能超过我们设置的虚拟内存大小
--
propertynameyarn.nodemanager.vmem-check-enabled/namevaluefalse/value
/property!-- 容器的虚拟内存使用上限与物理内存的比率 --
propertynameyarn.nodemanager.vmem-pmem-ratio/namevalue2.1/value
/property将 yarn-site.xml 同步给其他Linux
scp yarn-site.xml linux02:$PWD
scp yarn-site.xml linux03:$PWD
配置一键启停
cd /opt/apps/hadoop-3.1.1/sbin
vi start-yarn.sh
vi stop-yarn.shYARN_RESOURCEMANAGER_USERroot
HADOOP_SECURE_DN_USERyarn
YARN_NODEMANAGER_USERrootstart-yarn.sh 一键启动 启动后可以访问 http://linux01:8088 查看页面解决linux连接部上 网卡出现问题
systemctl stop NetworkManager
systemctl diable NetworkManager
systemctl restart networkMR程序提交到Yarn上运行 使用idea提交程序
配置mapred-site.xml文件 添加到resources目录下
?xml version1.0?
?xml-stylesheet typetext/xsl hrefconfiguration.xsl?configurationpropertynameyarn.app.mapreduce.am.env/namevalueHADOOP_MAPRED_HOME/opt/apps/hadoop-3.1.1/value/propertypropertynamemapreduce.map.env/namevalueHADOOP_MAPRED_HOME/opt/apps/hadoop-3.1.1/value/propertypropertynamemapreduce.reduce.env/namevalueHADOOP_MAPRED_HOME/opt/apps/hadoop-3.1.1/value/property/configurationday05.com.doit.demo06;
修改提交任务的代码 maven打jar包的命令为 package
public class Test02 {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {System.setProperty(HADOOP_USER_NAME, root);Configuration conf new Configuration();//操作HDFS数据conf.set(fs.defaultFS, hdfs://linux01:8020);//设置运行模式conf.set(mapreduce.framework.name, yarn);//设置ResourceManager位置conf.set(yarn.resourcemanager.hostname, linux01);// 设置MapReduce程序运行在windows上的跨平台参数conf.set(mapreduce.app-submission.cross-platform,true);Job job Job.getInstance(conf, WordCount);//设置jar包路径job.setJar(D:\\IdeaProjects\\hadoop\\target\\test_yarn.jar);job.setMapperClass(WordCountMapper.class);job.setReducerClass(WordCountReduce.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//设置路径为HDFS路径FileInputFormat.setInputPaths(job,new Path(/wc/input/word.txt));FileOutputFormat.setOutputPath(job,new Path(/wc/out4));job.waitForCompletion(true);}
}
在linux上直接提交jar包
public class Test02 {public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {Configuration conf new Configuration();//设置运行模式conf.set(mapreduce.framework.name, yarn);//设置ResourceManager位置conf.set(yarn.resourcemanager.hostname, linux01);// 设置MapReduce程序运行在windows上的跨平台参数conf.set(mapreduce.app-submission.cross-platform,true);Job job Job.getInstance(conf, WordCount);//设置jar包路径//job.setJar(D:\\IdeaProjects\\hadoop\\target\\test_yarn.jar);job.setJarByClass(Test02.class);job.setMapperClass(WordCountMapper.class);job.setReducerClass(WordCountReduce.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//设置路径为HDFS路径FileInputFormat.setInputPaths(job,new Path(/wc/input/word.txt));FileOutputFormat.setOutputPath(job,new Path(/wc/out5));job.waitForCompletion(true);}
}
打成jar包后
linux上使用
需要查看 mapred-site.xml 如果没有配置 需要配置一下hadoop jar jar包名 运行的类
hadoop jar test_yarn.jar day03.com.doit.demo02.Test02
Map Join
Map端join是指数据达到map处理函数之前进行合并的,效率要远远高于Reduce端join因为Reduce端join是把所有的数据都经过Shuffle非常消耗资源。
order.txt
order011 u001
order012 u001
order033 u005
order034 u002
order055 u003
order066 u004
order077 u010user.txt
u001,hangge,18,male,angelababy
u002,huihui,58,female,ruhua
u003,guanyu,16,male,chunge
u004,laoduan,38,male,angelababy
u005,nana,24,femal,huangbo
u006,xingge,18,male,laoduan最终结果
u001,hangge,18,male,angelababy,order012
u001,hangge,18,male,angelababy,order011
u002,huihui,58,female,ruhua,order034
u003,guanyu,16,male,chunge,order055
u004,laoduan,38,male,angelababy,order066
u005,nana,24,femal,huangbo,order033
null,order077
一个用户可能会产生多个订单,可能user.txt中的用户非常少,但是订单数据又非常非常多,这时我们可以考虑使用Map端join.一个小文件,一个大文件时,可以使用Map端join,说的简单一些,就是不走reduce,通过Map直接得出结果.
原理:将小文件上传到分布式缓存,保证每个map都可以访问完整的小文件的数据,然后与大文件切分后的数据进行连接,得出最终结果. package hadoop06.com.doit.demo;import hadoop03.com.doit.demo02.WordCountMapper;
import hadoop03.com.doit.demo02.WordCountReducer;
import hadoop05.com.doit.demo05.Test;
import org.apache.commons.lang.ObjectUtils;
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.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;public class MapJoinDemo {public static class JoinMapper extends MapperLongWritable,Text,Text, NullWritable{//定义集合用来存储user.txt的数据 键是uid 值是这一行记录private MapString,String userMap new HashMap();private Text k2 new Text();Overrideprotected void setup(MapperLongWritable, Text, Text, NullWritable.Context context) throws IOException, InterruptedException {//读取本地user.txt文件 由于user.txt添加到了分布式缓存中,会将这个文件 缓存到执行maptask的计算机上//由于这个文件和class文件放在一起 可以直接读取BufferedReader br new BufferedReader(new FileReader(user.txt));String line null;while((line br.readLine())!null){//System.out.println(line);String uid line.split(,)[0];//将uid 和 user的一行记录放入到map中userMap.put(uid,line);}}Overrideprotected void map(LongWritable key, Text value, MapperLongWritable, Text, Text, NullWritable.Context context) throws IOException, InterruptedException {//得到order的一条记录String line value.toString();//获取order的 uidString uid line.split(\\s)[1];// u001//获取map中 当前uid的 用户信息String userInfo userMap.get(uid);//拼接字符串写出k2.set(userInfo,line.split(\\s)[0]);context.write(k2, NullWritable.get());}}public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException, URISyntaxException {System.setProperty(HADOOP_USER_NAME, root);Configuration con new Configuration();//配置到yarn上执行con.set(mapreduce.framework.name, yarn);//配置操作HDFS数据con.set(fs.defaultFS, hdfs://linux01:8020);//配置resourceManager位置con.set(yarn.resourcemanager.hostname, linux01);//配置mr程序运行在windows上的跨平台参数con.set(mapreduce.app-submission.cross-platform,true);Job job Job.getInstance(con,wordcount);//分布式缓存user.txt文件job.addCacheFile(new URI(hdfs://linux01:8020/user.txt));//设置jar包的路径job.setJar(D:\\IdeaProjects\\test_hadoop\\target\\test_hadoop-1.0-SNAPSHOT.jar);//设置Mapperjob.setMapperClass(JoinMapper.class);//设置最后结果的输出类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);//设置读取HDFS上的文件 的路径//设置读取文件的位置 可以是文件 也可以是文件夹FileInputFormat.setInputPaths(job,new Path(/join/order.txt));//设置输出文件的位置 指定一个文件夹 文件夹不已存在 会报错FileOutputFormat.setOutputPath(job,new Path(/join/out));//提交任务 并等待任务结束job.waitForCompletion(true);}}public class Test02 {public static void main(String[] args) {int[] arr {3,4,2,8,5,1,7,3};quickSort(arr,0,arr.length-1);System.out.println(Arrays.toString(arr));}public static void quickSort(int[] arr, int startIndex, int endIndex) {if (startIndex endIndex) {return;}// 核心算法部分分别介绍 双边指针交换法int pivotIndex doublePointerSwap(arr, startIndex, endIndex);// 用分界值下标区分出左右区间进行递归调用quickSort(arr, startIndex, pivotIndex - 1);quickSort(arr, pivotIndex 1, endIndex);}private static int doublePointerSwap(int[] arr, int startIndex, int endIndex) {int pivot arr[startIndex];int leftPoint startIndex;int rightPoint endIndex;while (leftPoint rightPoint) {// 从右向左找出比pivot小的数据while (leftPoint rightPoint arr[rightPoint] pivot) {rightPoint--;}// 从左向右找出比pivot大的数据while (leftPoint rightPoint arr[leftPoint] pivot) {leftPoint;}// System.out.println(leftPoint rightPoint);// 没有过界则交换if (leftPoint rightPoint) {int temp arr[leftPoint];arr[leftPoint] arr[rightPoint];arr[rightPoint] temp;}// System.out.println(Arrays.toString(arr));}// 最终将分界值与当前指针数据交换arr[startIndex] arr[rightPoint];arr[rightPoint] pivot;// 返回分界值所在下标return rightPoint;}}归并排序
package com.doit.demo;import java.util.Arrays;
import java.util.Random;public class Test03 {public static void main(String[] args) {int[] arr new int[100];//向数组中添加100的 0-1000的随机数for (int i 0; i arr.length; i) {arr[i] new Random().nextInt(1000);}//调用排序sort(arr,0,arr.length-1);System.out.println(Arrays.toString(arr));}public static void sort(int[] arr,int start,int end){//递归出口if(startend){return;}//计算中间索引 拆分数组int mid start(end- start)/2; // length 10 ; 09/2 4; 左0-4 右 5 - 9;
// //左边有序数组sort(arr,start,mid);//右边有序数组sort(arr,mid1,end);
////合并merge(arr,start,mid,end);}public static void merge(int[] arr ,int start,int mid, int end){//定义一个辅助数组int[] assist new int[arr.length] ;int i start;int leftIndex start;int rightIndex mid1;//循环比较while(leftIndexmid rightIndexend){//如果左边数组的元素比右边数组的元素小 则将左边数组的元素放入到辅助数组中if(arr[leftIndex] arr[rightIndex]){assist[i] arr[leftIndex];i;leftIndex;}else{//如果左边数组的元素比右边的元素大 则将右边的元素放入到辅助数组中assist[i] arr[rightIndex];i;rightIndex;}}//如果左边的数组没走完 将剩下的放入到辅助数组中while(leftIndexmid){assist[i] arr[leftIndex];i;leftIndex;}//如果右边的数组没走完 将剩下的放入到辅助数组中while(rightIndexend){assist[i] arr[rightIndex];i;rightIndex;}//将辅助数组的值 为 原本的数组赋值for(int index start;indexend;index){arr[index] assist[index];}}
}
