Hadoop实践

1. 客户端（client）：编写mapreduce程序，配置作业，提交作业，这就是程序员完成的工作；
2. JobTracker：初始化作业，分配作业，与TaskTracker通信，协调整个作业的执行；
3. TaskTracker：保持与JobTracker的通信，在分配的数据片段上执行Map或Reduce任务，TaskTracker和JobTracker的不同有个很重要的方面，就是在执行任务时候TaskTracker可以有n多个，JobTracker则只会有一个（JobTracker只能有一个就和hdfs里namenode一样存在单点故障，我会在后面的mapreduce的相关问题里讲到这个问题的）
4. Hdfs：保存作业的数据、配置信息等等，最后的结果也是保存在hdfs上面

# vi etc/hadoop/core-site.xml
<configuration>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://172.16.10.150:9000</value>
    </property>
</configuration>
# vi etc/hadoop/hdfs-site.xml
<configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
</configuration>

#yarn-site.xml <!-- 开启日志聚合 --><property><name>yarn.log-aggregation-enable</name><value>true</value></property> #mapred-site.xml <!-- 设置jobhistoryserver 没有配置的话 history入口不可用 --> <property> <name>mapreduce.jobhistory.address</name> <value>172.16.10.150:10020</value> </property> <!-- 配置web端口 --> <property> <name>mapreduce.jobhistory.webapp.address</name> <value>172.16.10.150:19888</value> </property> <!-- 配置正在运行中的日志在hdfs上的存放路径 --> <property> <name>mapreduce.jobhistory.intermediate-done-dir</name> <value>/history/done_intermediate</value> </property> <!-- 配置运行过的日志存放在hdfs上的存放路径 --> <property> <name>mapreduce.jobhistory.done-dir</name> <value>/history/done</value> </property>

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test;

import java.net.URI;
import java.util.Iterator;
import java.util.Map.Entry;
/**
*
* 客户端去操作hdfs时，是有一个用户身份的
* 默认情况下，hdfs客户端api会从jvm中获取一个参数来作为自己的用户身份：-DHADOOP_USER_NAME=hadoop
*
* 也可以在构造客户端fs对象时，通过参数传递进去
* @author
*
*/
public class HdfsClientDemo {
FileSystem fs = null;
Configuration conf = null;
static String ip = "172.16.10.150";

@Before
public void init() throws Exception{
conf = new Configuration();
//conf.set("fs.defaultFS", "hdfs://"+ip+":9000");
//conf.set("hadoop.home.dir", "D:/Developer/hadoop-2.8.4");
//拿到一个文件系统操作的客户端实例对象
/*fs = FileSystem.get(conf);*/
//可以直接传入 uri和用户身份
fs = FileSystem.get(new URI("hdfs://"+ip+":9000"),conf, "root"); //最后一个参数为用户名
}

@Test
public void testUpload() throws Exception {
Thread.sleep(2000);
fs.copyFromLocalFile(new Path("F:\\工作目录\\users.txt"), new Path("/input/users.txt"));
fs.close();
}

@Test
public void listTest() throws Exception{

FileStatus[] listStatus = fs.listStatus(new Path("/"));
for (FileStatus fileStatus : listStatus) {
System.err.println(fileStatus.getPath()+"================="+fileStatus.toString());
}
//会递归找到所有的文件
RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/output/"), true);
while(listFiles.hasNext()){
LocatedFileStatus next = listFiles.next();
String name = next.getPath().getName();
Path path = next.getPath();
System.out.println(name + "---" + path.toString());
}
}
@Test
public void testDownload() throws Exception {
fs.copyToLocalFile(new Path("/user/test/input/users.txt"), new Path("d:/users.txt"));
fs.close();
}
@Test
public void testConf(){
Iterator<Entry<String, String>> iterator = conf.iterator();
while (iterator.hasNext()) {
Entry<String, String> entry = iterator.next();
System.out.println(entry.getValue() + "--" + entry.getValue());//conf加载的内容
}
}
/**
* 创建目录
*/
@Test
public void makdirTest() throws Exception {
boolean mkdirs = fs.mkdirs(new Path("/aaa/bbb"));
System.out.println(mkdirs);
}
/**
* 删除
*/
public static void deleteDir(FileSystem fs, String dir) throws Exception{
boolean delete = fs.delete(new Path("/"+dir), true);//true， 递归删除
System.out.println(delete);
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://"+ip+":9000");
//拿到一个文件系统操作的客户端实例对象
FileSystem fs = FileSystem.get(conf);
fs.copyFromLocalFile(new Path("D:/unLockPass.txt"), new Path("/input/unLockPass.txt"));
fs.close();
}
}

import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.LongWritable;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
public class WordCount {
static String ip = "172.16.10.150";

public static class WordCountMap extends Mapper<LongWritable, Text, Text, IntWritable> {
private final IntWritable one = new IntWritable(1);
private Text word = new Text();
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
StringTokenizer token = new StringTokenizer(line);
while (token.hasMoreTokens()) {
word.set(token.nextToken());
context.write(word, one);
}
}
}
public static class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {
public void reduce(Text key, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
int sum = 0;
for (IntWritable val : values) {
sum += val.get();
}
context.write(key, new IntWritable(sum));
}
}

public static void printDSFile(FileSystem fs, String dir){
BufferedReader in = null;
String line;
try {
System.out.println("uri:" + fs.getUri());
RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"+dir+"/"), true);
while(listFiles.hasNext()){
LocatedFileStatus next = listFiles.next();
String name = next.getPath().getName();
Path path = next.getPath();
FSDataInputStream result = fs.open(path);
in = new BufferedReader(new InputStreamReader(result, "UTF-8"));
String out="";
while ((line = in.readLine()) != null) {
out+=line+"\r\n";
}
System.out.println(name + ":\r\n" + out);
}
}catch(IOException ex){
ex.printStackTrace();
}finally {
if (in != null) {
try {
in.close();
}catch(IOException ex){
ex.printStackTrace();
}
}
}
}

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://"+ip+":9000");
FileSystem fs = FileSystem.get(new URI("hdfs://"+ip+":9000"),conf, "root");
HdfsClientDemo.deleteDir(fs, "output");
Job job = new Job(conf);
job.setJarByClass(WordCount.class);
job.setJobName("wordcount");
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setMapperClass(WordCountMap.class);
job.setReducerClass(WordCountReduce.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);

FileInputFormat.addInputPath(job, new Path(fs.getUri()+"/input/"));
FileOutputFormat.setOutputPath(job, new Path(fs.getUri()+"/output/"));
job.waitForCompletion(true);

printDSFile(fs, "output");
/*
fs.getHomeDirectory() - xxxx/user/root
hadoop fs -rmdir /output
hadoop fs -cat /output/*
*/
}
}