Hadoop面试题汇总-20221031

#创建缓存组，默认为cache_data
hdfs cacheadmin -addPool cache_data -mode 0777

#生成一个1GB大小的文件
dd if=/dev/zero of=/tmp/test.zero bs=1M count=1024

#将文件上传到HDFS
hdfs dfs -put /tmp/test.zero /data

#生成缓存指令
hdfs cacheadmin -addDirective -path /data -pool cache_data -ttl 1d

#显示缓存池的信息
hdfs cacheadmin -listPools -stats cache_data

#统计信息，显示EXP Date
hdfs cacheadmin -listDirectives -path /data

#删除缓存指令
hdfs cacheadmin -removeDirectives -path /data

#删除缓存池
hdfs cacheadmin -removePool cache_data

# HDFS Shell帮助命令，help更加详细，usage简略些
hadoop fs -help [option]
hadoop fs -usage [option]

# 查看文件目录
hadoop fs -ls /
hadoop fs -ls -R /

# 创建文件
hadoop fs -touchz /edits.txt

# 创建目录
hadoop fs -mkdir /tmp_data

# 追加文件
hadoop fs -appendToFile edits.txt /edits.txt

# 查看文件内容
hadoop fs -cat /edits.txt

# 上传文件
hadoop fs -put /LocalPath /HDFSPath
hadoop fs -copyFromLocal /LocalPath /HDFSPath
hadoop fs -moveFromLocal /LocalPath /HDFSPath

# 下载文件
hadoop fs -get /HDFSPath /LocalPath
hadoop fs -copyToLocal /HDFSPath /LocalPath

# 删除文件
hadoop fs -rm /edits.txt
hadoop fs -rm -skipTrash /edits.txt

# 递归删除
hadoop fs -rm -r /shell

# 移动文件
hadoop fs -mv /originFile /newFile

# 拷贝文件
hadoop fs -copy /FilePath /dictPath

# 文件查找
hadoop fs -find / -name part-r-00000

  <!-- MapReduce HistoryServer地址 -->
  <property>
      <name>mapreduce.jobhistory.address</name>
      <value>node01:10020</value>
  </property>
   <!-- HistoryServer Web地址 -->
  <property>
      <name>mapreduce.jobhistory.webapp.address</name>
      <value>node01:19888</value>
  </property>
  <!-- MapReduce日志保存位置 -->
  <property>
      <name>mapreduce.jobhistory.intermediate-done-dir</name>
      <value>/mr-history/log</value>
  </property>
  <!-- HistoryServer日志保存位置 -->
  <property>
      <name>mapreduce.jobhistory.done-dir</name>
      <value>/mr-history/done</value>
  </property>

  <!-- 聚合日志开启后，保存到HDFS中 -->
  <property>
      <name>yarn.log-aggregation-enable</name>
      <value>true</value>
  </property>

<!-- 聚合日志保存时间 -->
<property>
    <name>yarn.log-aggregation.retain-seconds</name>
    <value>2592000</value><!--30 day-->
</property>
<!-- 聚合日志压缩类型 -->
<property>
    <name>yarn.nodemanager.log-aggregation.compression-type</name>
    <value>gz</value>
</property>
<!-- 不启用聚合日志，日志在本地保存的时间 -->
<property>
    <name>yarn.nodemanager.log.retain-seconds</name>
    <value>604800</value><!-- 7 day -->
</property>

<!--  nodemanager 本地文件存储目录  -->
<property>
    <name>yarn.nodemanager.local-dirs</name>
    <value>/opt/app/hadoop-2.7.7/hadoopDatas/yarn/local</value>
</property>
<!--  resourceManager 最多保存已完成任务个数  -->
<property>
    <name>yarn.resourcemanager.max-completed-applications</name>
    <value>1000</value>
</property>

$HADOOP_HOME/sbin/mr-jobhistory-daemon.sh start historyserver

<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/app/hadoop-2.7.7/etc/hadoop/fair-scheduler.xml</value>
	<description>公平调度策略配置文件目录</description>
</property>

<property>
	<name>yarn.scheduler.fair.preemption</name>
	<value>true</value>
	<description>开启资源抢占</description>
</property>
<property>
	<name>yarn.scheduler.fair.preemption.cluster-utilization-threshold</name>
	<value>0.8f</value>
	<description>当集群的整体资源利用率超过80%，则开始抢占</description>
</property>

<property>
	<name>yarn.scheduler.fair.user-as-default-queue</name>
	<value>true</value>
	<description>默认提交到default队列</description>
</property>

<property>
	<name>yarn.scheduler.fair.allow-undeclared-pools</name>
	<value>false</value>
	<description>任务无法提交到现有队列，是否允许新建一个队列</description>
</property>

<?xml version="1.0"?>
<allocations>
	<!-- 每个队列中的默认调度策略，默认值是fair -->
	<defaultQueueSchedulingPolicy>fair</defaultQueueSchedulingPolicy>

	<user name="hadoop">
		<!-- 用户hadoop最多运行的任务数 -->
		<maxRunningApps>30</maxRunningApps>
	</user>
	<!-- 默认运行任务数 -->
	<userMaxAppsDefault>10</userMaxAppsDefault>

	<!-- 队列划分  -->
	<!-- 共有4个队列：default、hadoop、develop、test  -->
	<!-- 权重weight分别是：1、2、1、1.5  -->
	<queue name="root">
		<minResources>512mb,4vcores</minResources>
		<maxResources>102400mb,100vcores</maxResources>
		<maxRunningApps>100</maxRunningApps>
		<weight>1.0</weight>
		<schedulingMode>fair</schedulingMode>
		<aclSubmitApps> </aclSubmitApps>
		<aclAdministerApps> </aclAdministerApps>

		<queue name="default">
			<minResources>512mb,4vcores</minResources>
			<maxResources>30720mb,30vcores</maxResources>
			<maxRunningApps>100</maxRunningApps>
			<schedulingMode>fair</schedulingMode>
			<weight>1.0</weight>
			<!--  允许提交任务的用户和组 -->
			<!--  格式为：用户名1,用户名2 用户名1所属组,用户名2所属组 -->
			<!--  * 表示接收所有用户的任务 -->
			<!--  默认情况下，当任务没有指定队列时，会提交到default中 -->
			<aclSubmitApps>*</aclSubmitApps>
		</queue>

		<queue name="hadoop">
			<minResources>512mb,4vcores</minResources>
			<maxResources>20480mb,20vcores</maxResources>
			<maxRunningApps>100</maxRunningApps>
			<schedulingMode>fair</schedulingMode>
			<weight>2.0</weight>
			<!--  允许提交任务的用户和组 -->
			<aclSubmitApps>hadoop hadoop</aclSubmitApps>
			<!--  允许管理任务的用户和组 -->
			<aclAdministerApps>hadoop hadoop</aclAdministerApps>
		</queue>

		<queue name="develop">
			<minResources>512mb,4vcores</minResources>
			<maxResources>20480mb,20vcores</maxResources>
			<maxRunningApps>100</maxRunningApps>
			<schedulingMode>fair</schedulingMode>
			<weight>1</weight>
			<aclSubmitApps>develop develop</aclSubmitApps>
			<aclAdministerApps>develop develop</aclAdministerApps>
		</queue>

		<queue name="test">
			<minResources>512mb,4vcores</minResources>
			<maxResources>20480mb,20vcores</maxResources>
			<maxRunningApps>100</maxRunningApps>
			<schedulingMode>fair</schedulingMode>
			<weight>1.5</weight>
			<aclSubmitApps>test,hadoop,develop test</aclSubmitApps>
			<aclAdministerApps>test group_businessC,supergroup</aclAdministerApps>
		</queue>
	</queue>

	<!-- 任务提交规则，由scheduler调度器决定提交的任务进入指定队列 -->
	<!-- 包含多个rule标签，rule标签默认情况下create参数=true，表示当前rule可以创建一个新队列来存放任务 -->
	<queuePlacementPolicy>
		<!-- 任务被提交到指定的队列；如果队列不存在，则不创建 -->
		<rule name="specified" create="false"/>
		<!-- 任务被提交到，以提交用户所属组名称所命名的队列；如果队列不存在，则不创建 -->
		<rule name="primaryGroup" create="false" />
		<!-- 当不满足以上规则，则提交到root.default队列（默认值） -->
		<rule name="default" queue="root.default"/>
	</queuePlacementPolicy>
</allocations>

root 
├── prod 
└── dev 
	├── spark 
	└── hdp

<property>
	<name>yarn.resourcemanager.scheduler.class</name>
	<value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler</value>
</property>

<?xml version="1.0" encoding="utf-8"?>

<configuration> 
  <property> 
    <name>yarn.scheduler.capacity.root.queues</name>  
    <value>prod,dev</value> 
  </property>  
  <property> 
    <name>yarn.scheduler.capacity.root.dev.queues</name>  
    <value>hdp,spark</value> 
  </property>  
  <property> 
    <name>yarn.scheduler.capacity.root.prod.capacity</name>  
    <value>40</value> 
  </property>  
  <property> 
    <name>yarn.scheduler.capacity.root.dev.capacity</name>  
    <value>60</value> 
  </property>  
  <property> 
    <name>yarn.scheduler.capacity.root.dev.maximum-capacity</name>  
    <value>75</value> 
  </property>  
  <property> 
    <name>yarn.scheduler.capacity.root.dev.hdp.capacity</name>  
    <value>50</value> 
  </property>  
  <property> 
    <name>yarn.scheduler.capacity.root.dev.spark.capacity</name>  
    <value>50</value> 
  </property> 
</configuration>

yarn rmadmin -refreshQueues

Configuration conf = new Configuration();
conf.set("mapreduce.job.queuename", "hadoop");

public class HashPartitioner<K, V> extends Partitioner<K, V> {

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

}

public class MyPartitioner extends Partitioner<Text, IntWritable> {

    @Override
    public int getPartition(Text key, IntWritable value, int numPartitions) {
        // TODO Auto-generated method stub
        return 0;
    }

}

job.setNumReduceTasks(5);
job.setPartitionerClass(MyPartitioner.class);

public interface WritableComparable<T> extends Writable, Comparable<T> 

public class Person implements Writable {

    private String name;
    private int age;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return "My name is " + this.name + ", and i'm " + this.age + " years old.";
    }
	
	// 序列化
    @Override
    public void write(DataOutput out) throws IOException {
        out.write(this.name.getBytes());
        out.writeInt(this.age);
    }

	// 反序列化
    @Override
    public void readFields(DataInput in) throws IOException {
        this.setName(in.readLine());
        this.setAge(in.readInt());
    }

}

Math.max(minSize, Math.min(maxSize, blockSize));   
mapreduce.input.fileinputformat.split.minsize=1 默认值为1  
mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默认值Long.MAXValue  
blockSize为128M 

mapreduce.input.fileinputformat.split.maxsize
mapreduce.input.fileinputformat.split.minsize

while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
    ...
    bytesRemaining -= splitSize;
}

CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m

// 初始化资源，一般用于打开IO流
// 常用IO流为FSDataInputStream，默认会定义在成员变量中：inputStream
public void initialize(InputSplit split,
                                  TaskAttemptContext context
                                  ) throws IOException, InterruptedException
{
	FileSplit fs = (FileSplit) split;
	Path path = fs.getPath();
	FileSystem fileSystem = path.getFileSystem(context.getConfiguration());
	inputStream = fileSystem.open(path);
}

// 关闭资源，一般用于关闭IO流
public void close() throws IOException {
	IOUtils.closeStream(inputStream);
}

// 类似于指针，如果要读取的数据存在，返回true，否则返回false
public boolean nextKeyValue() throws IOException, InterruptedException {
}

// 获取当前行的key
public KEYIN getCurrentKey() throws IOException, InterruptedException;
  
// 获取当前行的value
public VALUEIN getCurrentValue() throws IOException, InterruptedException;

// 返回数据读取进度，0-1
public float getProgress() throws IOException, InterruptedException;

job.setInputFormatClass();

// 如果返回正数，o在前面；返回负数,表示o在后面；相等，返回0
// 意味着如果降序，则o>this时，返回正数；o<this时返回负数。升序时，this>o返回正数；this<o返回负数
public int compareTo(T o) {
	return Long.compare(o.value,this.value);
};

public class OrderGroupingComparator extends WritableComparator {
	
	// 初始化空对象用于数据接收
	protected OrderGroupingComparator() {
		super(OrderBean.class, true);
	}

	@Override
	public int compare(WritableComparable a, WritableComparable b) {

		OrderBean aBean = (OrderBean) a;
		OrderBean bBean = (OrderBean) b;

		int result;
		if (aBean.getOrder_id() > bBean.getOrder_id()) {
			result = 1;
		} else if (aBean.getOrder_id() < bBean.getOrder_id()) {
			result = -1;
		} else {
			result = 0;
		}

		return result;
	}
}

public class MyRecordWriter extends RecordWriter<Text, IntWritable> {

    private FSDataOutputStream file;

    // 自定义方法，开流
    public void initialize(TaskAttemptContext job) throws IOException {
        String outdir = job.getConfiguration().get(FileOutputFormat.OUTDIR);
        FileSystem fileSystem = FileSystem.get(job.getConfiguration());
        file = fileSystem.create(new Path(outdir));
    }

    // 将数据写出
    @Override
    public void write(Text key, IntWritable value) throws IOException, InterruptedException {
        // TODO Auto-generated method stub
        file.write(value.hashCode());
    }

    // 关闭资源
    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(file);
    }
    
}

public class MyOutputFormat extends FileOutputFormat<Text, IntWritable> {

    @Override
    public RecordWriter<Text, IntWritable> getRecordWriter(TaskAttemptContext job)
            throws IOException, InterruptedException {
        MyRecordWriter mRecordWriter = new MyRecordWriter();
        mRecordWriter.initialize(job);
        return mRecordWriter;
    }

}

job.addCacheFile(new URI("path"));

URI[] cacheFiles = context.getCacheFiles();
String path = cacheFiles[0].toString();
BufferReader bufferReader = new BufferReader(new FileReader(path));
String line;
while (StringUtils.isNotEmpty(line = bufferReader.readLine())){
	//进行数据处理
}