Spark(19)Yarn-tool接口
（一）需求
让自己编写的程序也可以支持动态参数。编写Yarn的Tool接口。
（二）Tool接口定义
org.apache.hadoop.util.Tool 是 Apache Hadoop 框架里的一个接口，其用途是协助开发可通过命令行运行的 Hadoop 应用程序。该接口能够让程序解析命令行参数，并且以一致的方式和 Hadoop 配置系统交互。
接口定义如下：
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool;
public interface Tool {
int run(String[] args) throws Exception;
void setConf(Configuration conf);
Configuration getConf();
}
接口方法解释
- run(String[] args)：此方法为应用程序的主要执行逻辑，接收命令行参数，返回一个整数代表执行结果（通常 0 表示成功，非 0 表示失败）。
- setConf(Configuration conf)：用来设置 Hadoop 的配置对象。
- getConf()：用于获取当前的 Hadoop 配置对象。
（三）使用Tool接口步骤
步骤 1：实现 Tool 接口
你需要创建一个类来实现 Tool 接口，同时实现其中的run, setConf, getConf方法。
步骤 2：使用 ToolRunner 运行应用程序
ToolRunner 是 Hadoop 提供的一个实用类，可用于运行实现了 Tool 接口的应用程序。可以可以main方法中通过ToolRunner.run()来运行程序。
（四）示例代码
以下是一个简单示例，展示了如何使用 Tool 接口：
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
// 实现 Tool 接口
public class MyTool implements Tool {
private Configuration conf;
@Override
public int run(String[] args) throws Exception {
// 打印命令行参数
for (String arg : args) {
System.out.println("Argument: " + arg);
}
// 这里可以添加你的主要业务逻辑
return 0; // 返回 0 表示执行成功
}
@Override
public void setConf(Configuration conf) {
this.conf = conf;
}
@Override
public Configuration getConf() {
return conf;
}
public static void main(String[] args) throws Exception {
// 使用 ToolRunner 运行应用程序
int exitCode = ToolRunner.run(new Configuration(), new MyTool(), args);
System.exit(exitCode);
}
}
代码解释
MyTool 类实现了 Tool 接口，并且实现了 run、setConf 和 getConf 方法。
run 方法会打印出接收到的命令行参数，你可以在此添加自己的业务逻辑。
main 方法使用 ToolRunner.run 方法来运行应用程序，并根据返回结果调用 System.exit 退出程序。
6. 运行示例
假设你已经将上述代码保存为 MyTool.java，可以按照以下步骤编译和运行：
（五）具体步骤
（1）新建Maven项目YarnDemo，并设置pom.xml如下：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.atguigu.hadoop</groupId>
<artifactId>yarn_tool_test</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.1.3</version>
</dependency>
</dependencies>
</project>
（2）新建一个包，名字为com.yarn。
（3）在这个包下创建类WordCountDriver并实现Tool接口：
package com.example.mapreduce;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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 org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.IOException;
import java.util.Arrays;
//mapreduce: 分布式计算框架
// 提交任务给hadoop集群执行
// 它要做7件事 hadoop jar mc8.js com.example.mapreduce.WordCountDriver /要处理的文件夹 /结果路径
public class WordCountDriverTool implements Tool {
private Configuration conf;
public static void main(String[] args) throws Exception {
switch(args[0]) {
case "wordcount":
int run = ToolRunner.run(new Configuration(), new WordCountDriverTool(), Arrays.copyOfRange(args,1, args.length));
System.exit(run);
break;
default:
throw new RuntimeException("no such tool" +args[0]);
}
}
@Override
public int run(String[] args) throws Exception {
// 连接到hadoop集群
// conf.set("fs.defaultFS", "hdfs://hadoop100:8020");
Job job = Job.getInstance(conf);
// 2. 关联本地的jar包
job.setJarByClass(WordCountDriverTool.class);
// 3. 关联Mapper和Reducer
job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);
// 4. 设置Map的键值对泛型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(LongWritable.class);
// 5. 设置Reduce的键值对泛型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(LongWritable.class);
// 6. 设置输入路径（D:\vm\wcinput）和输出路径（D:\vm\output01）
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 7. 提交job,根据返回值设置程序退出code
return job.waitForCompletion(true) ? 0 : 1;
}
@Override
public void setConf(Configuration configuration) {
this.conf = configuration;
}
@Override
public Configuration getConf() {
return conf;
}
}
重新打包生成jar，假设名称为MapReduceDemo1-1.0-SNAPSHOT.jar
上传到集群的节点上。
运行jar
hadoop jar MapReduceDemo1-1.0-SNAPSHOT.jar com.example.mapreduce.WordCountDriverTool wordcount /wcinput /a1
不使用-D参数，测试运行。
接下来，使用-D参数运行效果。
hadoop jar MapReduceDemo1-1.0-SNAPSHOT.jar com.example.mapreduce.WordCountDriverTool wordcount -Dmapreduce.job.queue.name=test /wcinput /a2
运行结果如下
