Hadoop

package org.example.wordcount2;
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 WordCountMapper extends Mapper<LongWritable , Text, Text, IntWritable>{
    private Text outk = new Text();
    private IntWritable outv = new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        // get a line ,text to string
        String line = value.toString();

        // 切割
        String[] words = line.split(" ");

        // loop write out
        for (String word:words){

            outk.set(word);
            context.write(outk,outv);
        }
    }
}

package org.example.wordcount2;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
public class WordCountReducer extends  Reducer{
    private IntWritable outV = new IntWritable();

    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {

        int sum = 0;

        for (IntWritable value : values) {
            sum += value.get();
        }

        outV.set(sum);

        // 写出
        context.write(key,outV);
    }

}

package org.example.wordcount2;
import org.apache.hadoop.conf.Configuration;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class WordCountDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 1 获取job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        // 2 设置jar包路径
        job.setJarByClass(WordCountDriver.class);

        // 3 关联mapper和reducer
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        // 4 设置map输出的kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 5 设置最终输出的kV类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 6 设置输入路径和输出路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 7 提交job
        boolean result = job.waitForCompletion(true);

        System.exit(result ? 0 : 1);
    }

}

hadoop jar MapReduceDemo-1.0-SNAPSHOT.jar org.example.wordcount2.WordCountDriver /input /uuu

// 5 设置最终输出的kV类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setInputFormatClass(CombineTextInputFormat.class);
CombineTextInputFormat.setMaxInputSplitSize(job,4194034);
// 6 设置输入路径和输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setInputFormatClass(CombineTextInputFormat.class);
        
        job.setNumReduceTasks(2);
        // 6 设置输入路径和输出路径
public class HashPartitioner<K, V> extendss Partitioner<K, V> {
    public int getPartition(K key, V value, int 1humReduceTasks)
         return (key.hashCode() & Integer.MAX_VALUE)៖ numReduceTasks;

root@hadoop100:~# yarn application -list 
2024-06-11 06:34:10,406 INFO client.DefaultNoHARMFailoverProxyProvider: Connecting to ResourceManager at hadoop101/192.168.13.191:8032
Total number of applications (application-types: [], states: [SUBMITTED, ACCEPTED, RUNNING] and tags: []):1
                Application-Id      Application-Name        Application-Type          User           Queue                   State           Final-State             Progress                        Tracking-URL
application_1717055365946_0004            word count               MAPREDUCE          root         default                 RUNNING             UNDEFINED                   0%              http://hadoop100:38735
root@hadoop100:~# yarn application -list 
2024-06-11 06:34:10,406 INFO client.DefaultNoHARMFailoverProxyProvider: Connecting to ResourceManager at hadoop101/192.168.13.191:8032
Total number of applications (application-types: [], states: [SUBMITTED, ACCEPTED, RUNNING] and tags: []):1
                Application-Id      Application-Name        Application-Type          User           Queue                   State           Final-State             Progress                        Tracking-URL
application_1717055365946_0004            word count               MAPREDUCE          root         default                 RUNNING             UNDEFINED                   0%              http://hadoop100:38735
root@hadoop100:~# yarn application -kill application_1717055365946_0004  
root@hadoop100:~# yarn logs -applicationId application_1717055365946_0004  
root@hadoop100:~# yarn logs -applicationId application_1717055365946_0004  -containerId 
# 查看尝试运行的任务
yarn applicationattempt -list application_1717055365946_0004 
# 
yarn applicationattempt -status application_1717055365946_0004 
# yarn 容器状态，只有在任务运行中可以查看
yarn container -list 
yarn container -status
yarn node -list -all
# 加载队列配置
yarn rmadmin -refreshQueues
# 查看队列
yarn queue -status default 

  vim capacity-scheduler.xml 
  <property>
    <name>yarn.scheduler.capacity.root.queues</name>
    <value>default,hive</value>
    <description>
      The queues at the this level (root is the root queue).
    </description>
  </property>
    <property>
    <name>yarn.scheduler.capacity.root.default.capacity</name>
    <value>40</value>
    <description>Default queue target capacity.</description>
  </property>
    <property>
    <name>yarn.scheduler.capacity.root.hive.capacity</name>
    <value>60</value>
    <description>Default queue target capacity.</description>
  </property>
    <property>
    <name>yarn.scheduler.capacity.root.default.user-limit-factor</name>
    <value>0.7</value>
    <description>
      Default queue user limit a percentage from 0.0 to 1.0.
    </description>
  </property>
    <property>
    <name>yarn.scheduler.capacity.root.hive.user-limit-factor</name>
    <value>1</value>
    <description>
      Default queue user limit a percentage from 0.0 to 1.0.
    </description>
  </property>
  
  
    <property>
    <name>yarn.scheduler.capacity.root.default.maximum-capacity</name>
    <value>60</value>
    <description>
      The maximum capacity of the default queue.
    </description>
  </property>
  
  
  <property>
    <name>yarn.scheduler.capacity.root.default.state</name>
    <value>RUNNING</value>
    <description>
      The state of the default queue. State can be one of RUNNING or STOPPED.
    </description>
  </property>

  <property>
    <name>yarn.scheduler.capacity.root.default.acl_submit_applications</name>
    <value>*</value>
    <description>
      The ACL of who can submit jobs to the default queue.
    </description>
  </property>
  
    <property>
    <name>yarn.scheduler.capacity.root.default.acl_administer_queue</name>
    <value>*</value>
    <description>
      The ACL of who can administer jobs on the default queue.
    </description>
  </property>
  
  <!-- 任务的超时时间设置：yarn application -appId appId -updateLifetimeTimeout
参考资料：https://blog.cloudera.com/enforcing-application-lifetime-slas-yarn/ -->
 
<!-- 如果application指定了超时时间，则提交到该队列的application能够指定的最大超时时间不能超过该值。 
-->
<property>
    <name>yarn.scheduler.capacity.root.hive.maximum-application-lifetime</name>
    <value>-1</value>
</property>
 
<!-- 如果application没指定超时时间，则用default-application-lifetime作为默认值 -->
<property>
    <name>yarn.scheduler.capacity.root.hive.default-application-lifetime</name>
    <value>-1</value>
</property>
# 提交任务到不同队列
hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-example-3.1.3.jar wordcount -D mapreduce.job.queuename=hive /input /outputttt

# 或者在driver中
Configure conf = new Configure();
conf.set("mapreduce.job.queuename","hive");

<property>
    <name>yarn.cluster.max-application-priority</name>
    <value>5</value>
</property>
hadoop jar hadoop-mapreduce-examples-3.1.3.jar pi -D mapreduce.job.priorit=5 5 20000

# 动态设置优先级
yarn application -appId xxx -upgradePriority 6

vim yarn-site.xml
<property>
    <name>yarn.resourcemanager.scheduler.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler</value>
    <description>配置使用公平调度器</description>
</property>
 
<property>
    <name>yarn.scheduler.fair.allocation.file</name>
    <value>/opt/module/hadoop-3.1.3/etc/hadoop/fair-scheduler.xml</value>
    <description>指明公平调度器队列分配配置文件</description>
</property>
 
<property>
    <name>yarn.scheduler.fair.preemption</name>
    <value>false</value>
    <description>禁止队列间资源抢占</description>
</property>
vim fair-scheduler.xml
<?xml version="1.0"?>
<allocations>
  <!-- 单个队列中Application Master占用资源的最大比例,取值0-1 ，企业一般配置0.1 -->
  <queueMaxAMShareDefault>0.5</queueMaxAMShareDefault>
  <!-- 单个队列最大资源的默认值 test atguigu default -->
  <queueMaxResourcesDefault>4096mb,4vcores</queueMaxResourcesDefault>
 
  <!-- 增加一个队列test -->
  <queue name="test">
    <!-- 队列最小资源 -->
    <minResources>2048mb,2vcores</minResources>
    <!-- 队列最大资源 -->
    <maxResources>4096mb,4vcores</maxResources>
    <!-- 队列中最多同时运行的应用数，默认50，根据线程数配置 -->
    <maxRunningApps>4</maxRunningApps>
    <!-- 队列中Application Master占用资源的最大比例 -->
    <!-- <maxAMShare>0.5</maxAMShare>-->
    <!-- 该队列资源权重,默认值为1.0 -->
    <weight>1.0</weight>
    <!-- 队列内部的资源分配策略 -->
    <schedulingPolicy>fair</schedulingPolicy>
  </queue>
  <!-- 增加一个队列atguigu -->
  <queue name="atguigu" type="parent">
    <!-- 队列最小资源 -->
    <minResources>2048mb,2vcores</minResources>
    <!-- 队列最大资源 -->
    <maxResources>4096mb,4vcores</maxResources>
    <!-- 队列中最多同时运行的应用数，默认50，根据线程数配置 -->
    <maxRunningApps>4</maxRunningApps>
    <!-- 队列中Application Master占用资源的最大比例 -->
    <!-- <maxAMShare>0.5</maxAMShare>-->
    <!-- 该队列资源权重,默认值为1.0 -->
    <weight>1.0</weight>
    <!-- 队列内部的资源分配策略 -->
    <schedulingPolicy>fair</schedulingPolicy>
  </queue>
 
  <!-- 任务队列分配策略,可配置多层规则,从第一个规则开始匹配,直到匹配成功 -->
  <queuePlacementPolicy>
    <!-- 提交任务时指定队列,如未指定提交队列,则继续匹配下一个规则; false表示：如果指定队列不存在,不允许自动创建-->
    <rule name="specified" create="false"/>
    <!-- 提交到root.group.username队列,若root.group不存在,不允许自动创建；若root.group.user不存在,允许自动创建 -->
    <rule name="nestedUserQueue" create="true">
        <rule name="primaryGroup" create="false"/>
    </rule>
    <!-- 最后一个规则必须为reject或者default。Reject表示拒绝创建提交失败，default表示把任务提交到default队列 -->
    <rule name="reject" />
  </queuePlacementPolicy>
</allocations>
# 指定用户和不指定用户
hadoop jar hadoop-mapreduce-examples-3.1.3.jar pi -D mapreduce.job.queuename=root.test 1 1 
hadoop jar hadoop-mapreduce-examples-3.1.3.jar pi 1 1 

创建类WordCount并实现Tool接口：
package com.atguigu.yarn;
 
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;
import org.apache.hadoop.util.Tool;
 
import java.io.IOException;
 
public class WordCount implements Tool {
 
    private Configuration conf;
 
    @Override
    public int run(String[] args) throws Exception {
 
        Job job = Job.getInstance(conf);
 
        job.setJarByClass(WordCountDriver.class);
 
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
 
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
 
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
 
        return job.waitForCompletion(true) ? 0 : 1;
    }
 
    @Override
    public void setConf(Configuration conf) {
        this.conf = conf;
    }
 
    @Override
    public Configuration getConf() {
        return conf;
    }
 
    public static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
 
        private Text outK = new Text();
        private IntWritable outV = new IntWritable(1);
 
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
 
            String line = value.toString();
            String[] words = line.split(" ");
 
            for (String word : words) {
                outK.set(word);
 
                context.write(outK, outV);
            }
        }
    }
 
    public static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable outV = new IntWritable();
 
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
 
            int sum = 0;
 
            for (IntWritable value : values) {
                sum += value.get();
            }
            outV.set(sum);
 
            context.write(key, outV);
        }
    }
    }