如何查看Hadoop运行过程中产生日志

用hadoop也算有一段时间了，一直没有注意过hadoop运行过程中，产生的数据日志，比如说System打印的日志，或者是log4j，slf4j等记录的日志，存放在哪里，日志信息的重要性，在这里散仙就不用多说了，调试任何程序基本上都得需要分析日志。

hadoop的日志主要是MapReduce程序，运行过程中，产生的一些数据日志，除了系统的日志外，还包含一些我们自己在测试时候，或者线上环境输出的日志，这部分日志通常会被放在userlogs这个文件夹下面，我们可以在mapred-sitexml里面配置运行日志的输出目录，散仙测试文件内容如下:

<xml version="10">

<xml-stylesheet type="text/xsl" href="configurationxsl">

<!-- Put site-specific property overrides in this file -->

<configuration>

<!-- jobtracker的master地址-->

<property>

<name>mapredjobtracker</name>

<value>19216875130:9001</value>

</property>

<property>

<!-- hadoop的日志输出指定目录-->

<name>mapredlocaldir</name>

<value>/root/hadoop12/mylogs</value>

</property>

</configuration>

配置好，日志目录后，我们就可以把这个配置文件，分发到各个节点上，然后启动hadoop。

下面我们看来下在eclipse环境中如何调试，散仙在setup，map和reduce方法中，分别使用System打印了一些数据，当我们使用local方式跑MR程序时候，日志并不会被记录下来，而是直接会在控制台打印，散仙的测试代码如下：

package comqintestdistributed;

import javaioFile;

import javaioFileReader;

import javaioIOException;

import javanetURI;

import javautilScanner;

import orgapachehadoopconfConfiguration;

import orgapachehadoopfilecacheDistributedCache;

import orgapachehadoopfsFSDataInputStream;

import orgapachehadoopfsFileSystem;

import orgapachehadoopfsPath;

import orgapachehadoopioIntWritable;

import orgapachehadoopioLongWritable;

import orgapachehadoopioText;

import orgapachehadoopmapredJobConf;

import orgapachehadoopmapreduceJob;

import orgapachehadoopmapreduceMapper;

import orgapachehadoopmapreduceReducer;

import orgapachehadoopmapreducelibdbDBConfiguration;

import orgapachehadoopmapreducelibinputFileInputFormat;

import orgapachehadoopmapreduceliboutputFileOutputFormat;

import orgapachelog4jpatternLogEvent;

import orgslf4jLogger;

import orgslf4jLoggerFactory;

import comqinoperadbWriteMapDB;

/

测试hadoop的全局共享文件

使用DistributedCached

大数据技术交流群： 37693216

@author qindongliang

/

public class TestDistributed {

private static Logger logger=LoggerFactorygetLogger(TestDistributedclass);

private static class FileMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

Path path[]=null;

/

Map函数前调用

/

@Override

protected void setup(Context context)

throws IOException, InterruptedException {

loggerinfo("开始启动setup了哈哈哈哈");

// Systemoutprintln("运行了");

Configuration conf=contextgetConfiguration();

path=DistributedCachegetLocalCacheFiles(conf);

Systemoutprintln("获取的路径是： "+path[0]toString());

// FileSystem fs = FileSystemget(conf);

FileSystem fsopen= FileSystemgetLocal(conf);

// FSDataInputStream in = fsopenopen(path[0]);

// Systemoutprintln(inreadLine());

// for(Path tmpRefPath : path) {

// if(tmpRefPathtoString()indexOf("refpng") != -1) {

// in = reffsopen(tmpRefPath);

// break;

// }

// }

// FileReader reader=new FileReader("file://"+path[0]toString());

// File f=new File("file://"+path[0]toString());

// FSDataInputStream in=fsopen(new Path(path[0]toString()));

// Scanner scan=new Scanner(in);

// while(scanhasNext()){

// Systemoutprintln(ThreadcurrentThread()getName()+"扫描的内容: "+scannext());

// }

// scanclose();

//

// Systemoutprintln("size: "+pathlength);

}

@Override

protected void map(LongWritable key, Text value,Context context)

throws IOException, InterruptedException {

// Systemoutprintln("map aaa");

//loggerinfo("Map里的任务");

Systemoutprintln("map里输出了");

// loggerinfo();

contextwrite(new Text(""), new IntWritable(0));

}

@Override

protected void cleanup(Context context)

throws IOException, InterruptedException {

loggerinfo("清空任务了。。。。。。");

}

}

private static class FileReduce extends Reducer<Object, Object, Object, Object>{

@Override

protected void reduce(Object arg0, Iterable<Object> arg1,

Context arg2)throws IOException, InterruptedException {

Systemoutprintln("我是reduce里面的东西");

}

}

public static void main(String[] args)throws Exception {

JobConf conf=new JobConf(TestDistributedclass);

//confset("mapredlocaldir", "/root/hadoop");

//Configuration conf=new Configuration();

// confset("mapredjobtracker","19216875130:9001");

//读取person中的数据字段

//confsetJar("ttjar");

//注意这行代码放在最前面，进行初始化，否则会报

String inputPath="hdfs://19216875130:9000/root/input";

String outputPath="hdfs://19216875130:9000/root/outputsort";

Job job=new Job(conf, "a");

DistributedCacheaddCacheFile(new URI("hdfs://19216875130:9000/root/input/f1txt"), jobgetConfiguration());

jobsetJarByClass(TestDistributedclass);

Systemoutprintln("运行模式： "+confget("mapredjobtracker"));

/设置输出表的的信息 第一个参数是job任务，第二个参数是表名，第三个参数字段项/

FileSystem fs=FileSystemget(jobgetConfiguration());

Path pout=new Path(outputPath);

if(fsexists(pout)){

fsdelete(pout, true);

Systemoutprintln("存在此路径, 已经删除");

}

/设置Map类/

// jobsetOutputKeyClass(Textclass);

//jobsetOutputKeyClass(IntWritableclass);

jobsetMapOutputKeyClass(Textclass);

jobsetMapOutputValueClass(IntWritableclass);

jobsetMapperClass(FileMapperclass);

jobsetReducerClass(FileReduceclass);

FileInputFormatsetInputPaths(job, new Path(inputPath)); //输入路径

FileOutputFormatsetOutputPath(job, new Path(outputPath));//输出路径

Systemexit(jobwaitForCompletion(true) 0 : 1);

}

}

Local模式下输出如下：

运行模式： local

存在此路径, 已经删除

WARN - NativeCodeLoader<clinit>(52) | Unable to load native-hadoop library for your platform using builtin-java classes where applicable

WARN - JobClientcopyAndConfigureFiles(746) | Use GenericOptionsParser for parsing the arguments Applications should implement Tool for the same

WARN - JobClientcopyAndConfigureFiles(870) | No job jar file set User classes may not be found See JobConf(Class) or JobConf#setJar(String)

INFO - FileInputFormatlistStatus(237) | Total input paths to process : 1

WARN - LoadSnappy<clinit>(46) | Snappy native library not loaded

INFO - TrackerDistributedCacheManagerdownloadCacheObject(423) | Creating f1txt in /root/hadoop12/hadooptmp/mapred/local/archive/9070031930820799196_1788685676_88844454/19216875130/root/input-work-186410214545932656 with rwxr-xr-x

INFO - TrackerDistributedCacheManagerdownloadCacheObject(463) | Cached hdfs://19216875130:9000/root/input/f1txt as /root/hadoop12/hadooptmp/mapred/local/archive/9070031930820799196_1788685676_88844454/19216875130/root/input/f1txt

INFO - TrackerDistributedCacheManagerlocalizePublicCacheObject(486) | Cached hdfs://19216875130:9000/root/input/f1txt as /root/hadoop12/hadooptmp/mapred/local/archive/9070031930820799196_1788685676_88844454/19216875130/root/input/f1txt

INFO - JobClientmonitorAndPrintJob(1380) | Running job: job_local479869714_0001

INFO - LocalJobRunner$Jobrun(340) | Waiting for map tasks

INFO - LocalJobRunner$Job$MapTaskRunnablerun(204) | Starting task: attempt_local479869714_0001_m_000000_0

INFO - Taskinitialize(534) | Using ResourceCalculatorPlugin : null

INFO - MapTaskrunNewMapper(729) | Processing split: hdfs://19216875130:9000/root/input/f1txt:0+31

INFO - MapTask$MapOutputBuffer<init>(949) | iosortmb = 100

INFO - MapTask$MapOutputBuffer<init>(961) | data buffer = 79691776/99614720

INFO - MapTask$MapOutputBuffer<init>(962) | record buffer = 262144/327680

INFO - TestDistributed$FileMappersetup(57) | 开始启动setup了哈哈哈哈

获取的路径是： /root/hadoop12/hadooptmp/mapred/local/archive/9070031930820799196_1788685676_88844454/19216875130/root/input/f1txt

map里输出了

map里输出了

INFO - TestDistributed$FileMappercleanup(107) | 清空任务了。。。。。。

INFO - MapTask$MapOutputBufferflush(1289) | Starting flush of map output

INFO - MapTask$MapOutputBuffersortAndSpill(1471) | Finished spill 0

INFO - Taskdone(858) | Task:attempt_local479869714_0001_m_000000_0 is done And is in the process of commiting

INFO - LocalJobRunner$JobstatusUpdate(466) |

INFO - TasksendDone(970) | Task 'attempt_local479869714_0001_m_000000_0' done

INFO - LocalJobRunner$Job$MapTaskRunnablerun(229) | Finishing task: attempt_local479869714_0001_m_000000_0

INFO - LocalJobRunner$Jobrun(348) | Map task executor complete

INFO - Taskinitialize(534) | Using ResourceCalculatorPlugin : null

INFO - LocalJobRunner$JobstatusUpdate(466) |

INFO - Merger$MergeQueuemerge(408) | Merging 1 sorted segments

INFO - Merger$MergeQueuemerge(491) | Down to the last merge-pass, with 1 segments left of total size: 16 bytes

INFO - LocalJobRunner$JobstatusUpdate(466) |

之前学习了一段时间的hadoop的相关知识 ，学习理论基础的时候要同时实际 *** 作才能对它更熟练，废话不多说来说说在hadoop上运行一个最简单的words count的程序

首先我先贴上这个程序的源代码 供大家参考 代码分为三个部分写的Run、 map阶段、 reduce阶段

Map：

[java] view plain copy

<span style="font-family:KaiTi_GB2312;font-size:18px;">package wordsCount;

import javaioIOException;

import javautilStringTokenizer;

import orgapachehadoopioIntWritable;

import orgapachehadoopioLongWritable;

import orgapachehadoopioText;

import orgapachehadoopmapreduceMapper;

public class WordsMapper extends Mapper<LongWritable, Text, Text, IntWritable>{

@Override

protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>Context context)

throws IOException, InterruptedException {

String line = valuetoString();

StringTokenizer st = new StringTokenizer(line);

while(sthasMoreTokens()){

String word = stnextToken();

contextwrite(new Text(word), new IntWritable(1));

}

}

}</span>

Reduce：

[java] view plain copy

<span style="font-family:KaiTi_GB2312;font-size:18px;">package wordsCount;

import javaioIOException;

import orgapachehadoopioIntWritable;

import orgapachehadoopioText;

import orgapachehadoopmapreduceReducer;

public class WordsReduce extends Reducer<Text, IntWritable, Text, IntWritable>{

@Override

protected void reduce(Text key, Iterable<IntWritable> iterator,

Reducer<Text, IntWritable, Text, IntWritable>Context context) throws IOException, InterruptedException {

// TODO 自动生成的方法存根

int sum = 0;

for(IntWritable i:iterator){

sum = sum + iget();

}

contextwrite(key, new IntWritable(sum));

}

}</span>

Run：

[java] view plain copy

<span style="font-family:KaiTi_GB2312;font-size:18px;">package wordsCount;

import orgapachehadoopconfConfiguration;

import orgapachehadoopfsPath;

import orgapachehadoopioIntWritable;

import orgapachehadoopioText;

import orgapachehadoopmapreduceJob;

import orgapachehadoopmapreducelibinputFileInputFormat;

import orgapachehadoopmapreducelibinputTextInputFormat;

import orgapachehadoopmapreduceliboutputFileOutputFormat;

import orgapachehadoopmapreduceliboutputTextOutputFormat;

public class Run {

public static void main(String[] args) throws Exception{

// TODO 自动生成的方法存根

Configuration configuration = new Configuration();

Job job = new Job(configuration);

jobsetJarByClass(Runclass);

jobsetJobName("words count!");

jobsetOutputKeyClass(Textclass);

jobsetOutputValueClass(IntWritableclass);

jobsetInputFormatClass(TextInputFormatclass);

jobsetOutputFormatClass(TextOutputFormatclass);

jobsetMapperClass(WordsMapperclass);

jobsetReducerClass(WordsReduceclass);

FileInputFormataddInputPath(job, new Path("hdfs://1921681111:9000/user/input/wc/"));

FileOutputFormatsetOutputPath(job,new Path("hdfs://1921681111:9000/user/result/"));

jobwaitForCompletion(true);

}

}</span>

Run里面的输入和输出路径根据自己的来修改

这个程序就不用去讲解了吧 到处都能找到

首先在hadoop上运行这个程序用两个方法

方法一：将自己的编译软件与hadoop相连（我用的是MyEclipse去链接hadoop），直接运行程序。MyEclipse连接hadoop的教程待会我会在文章结尾处给出一个链接供大家参考。

看到下面的信息就表示你成功了 然后你在再到你的输出文件夹里面就能查看运行的结果了

第二个文件里面的内容就是输出结果

第二种方法：将mapreduce程序打包成jar文件

这里简单的说一下打包的方法

然后下一步，完成就可以了

将打包好的jar文件传到你的装hadoop的机器上（我的hadoop集群是装在linux虚拟机中的）用SSH把jar传过去之后：

在你安装hadoop的目录下的bin目录下有一个hadoop的可执行文件，然后执行下面的 *** 作就可以了：

来解释下我的shell语句

/home/xiaohuihui/wordscountjar:打包之后的jar文件的所在位置（传到虚拟机中位置）

wordsCount/Run:这个位你的jar包中的主函数（这里的主函数就是Runclass）的名字 可以打开你的jar文件查看便知道

还可以在这个语句之后加上你的输入和输出的文件路径，但是这个我已经在我的程序中设置了

如果你运行上面的shell语句之后看到下面的输出，那恭喜你，成功了！！

查看结果你可通过在你的Eclipse连接好hadoop查看，还可以通过在hdfs文件系统的网页去查看（localhost：50070）。

还有一个很重要的一步就是，运行之前保证你的hadoop已经启动了，可以通过jps来查看你的进程中是否已经启动hadoop集群

Mapreduce简介

Hadoop MapReduce 源于Google发表的 MapReduce论文。Hadoop MapReduce 其实就是Google MapReduce的一个克隆版本。Hadoop 20即第二代Hadoop系统，其框架最核心的设计是HDFS、MapReduce和YARN。其中，HDFS为海量数据提供存储，MapReduce用于分布式计算，YARN用于进行资源管理。

其实，一次mapreduce过程就包括上图的6个步骤，input、splitting、mapping、shuffling、redecing、final redult。

文件要存储在HDFS中，每个文件被切分成多个一定大小的块也就是Block,(Hadoop10默认为64M,Hadoop20默认为128M),并且默认3个备份存储在多个的节点中。

MR通过Inputformat将数据文件从HDFS中读入取，读取完后会对数据进行split切片，切片的数量根据Block的大小所决定，然后每一个split的个数又决定map的个数，即一个split会分配一个maptask并行实例处理。

如何确定切分的文件大小？

数据进入到map函数中，然后开始按照一定的规则切分。其实这就是我们自定义的计算逻辑，我们编写mr程序的map函数的逻辑一般就在这个阶段执行。企业应用为了追求开发效率，一般都使用hive sql代替繁琐的mr程序了，这里附上一个经典的wordcount的map函数重温一下吧。

Shuffle是我们不需要编写的模块，但却是十分关键的模块。

在map中，每个 map 函数会输出一组 key/value对, Shuffle 阶段需要从所有 map主机上把相同的 key 的 key value对组合在一起，（也就是这里省去的Combiner阶段）组合后传给 reduce主机, 作为输入进入 reduce函数里。

Partitioner组件 负责计算哪些 key 应当被放到同一个 reduce 里

HashPartitioner类，它会把 key 放进一个 hash函数里，然后得到结果。如果两个 key 的哈希值 一样，他们的 key/value对 就被放到同一个 reduce 函数里。我们也把分配到同一个 reduce函数里的 key /value对 叫做一个reduce partition

我们看到 hash 函数最终产生多少不同的结果, 这个 Hadoop job 就会有多少个 reduce partition／reduce 函数，这些 reduce函数最终被JobTracker 分配到负责 reduce 的主机上，进行处理。

Map方法之后，数据首先进入到分区方法，把数据标记好分区，然后把数据发送到环形缓冲区；环形缓冲区默认大小100m，环形缓冲区达到80%时，进行溢写； 溢写前对数据进行排序 ，排序按照对key的索引进行字典顺序排序，排序的手段快排；溢写产生大量溢写文件，需要 对溢写文件进行归并排序 ；对溢写的文件也可以进行Combiner *** 作，前提是汇总 *** 作，求平均值不行。最后将文件按照分区存储到磁盘，等待Reduce端拉取。每个Reduce拉取Map端对应分区的数据。拉取数据后先存储到内存中，内存不够了，再存储到磁盘。拉取完所有数据后， 采用归并排序将内存和磁盘中的数据都进行排序 。在进入Reduce方法前，可以对数据进行分组 *** 作。值得注意的是， 整个shuffle *** 作是有3次排序的。

reduce() 函数以 key 及对应的 value 列表作为输入，按照用户自己的程序逻辑，经合并 key 相同的 value 值后，产 生另外一系列 key/value 对作为最终输出写入 HDFS。

将 hadoop执行job命令写到shell脚本中。类似 hadoop jar xjar ×××MainClassName inputPath outputPath这种命令。

hadoop客户机在本地，使用 Process执行shell脚本，java执行本地shell脚本的代码

1

2

3

4

Process process =null;

String command1 = " /testsh"；

process = RuntimegetRuntime()exec(command1);

processwaitFor();

3hadoop 在远程服务器上，java可以使用SSH连接远程服务器执行shell脚本。需要借助第三方jar包实现，常用的是ganymed-ssh2-build210jar，网上有很多例子。

大数据的时代， 到处张嘴闭嘴都是Hadoop, MapReduce, 不跟上时代怎么行？ 可是对一个hadoop的新手， 写一个属于自己的MapReduce程序还是小有点难度的， 需要建立一个maven项目， 还要搞清楚各种库的依赖， 再加上编译运行， 基本上头大两圈了吧。 这也使得很多只是想简单了解一下MapReduce的人望而却步。

本文会教你如何用最快最简单的方法编写和运行一个属于自己的MapReduce程序， let's go!

首先有两个前提：

1 有一个已经可以运行的hadoop 集群（也可以是伪分布系统)， 上面的hdfs和mapreduce工作正常 （这个真的是最基本的了， 不再累述， 不会的请参考 >

以上就是关于如何查看Hadoop运行过程中产生日志全部的内容，包括:如何查看Hadoop运行过程中产生日志、如何在hadoop环境下执行mapreduce任务、MapReduce工作流程最详细解释等相关内容解答，如果想了解更多相关内容，可以关注我们，你们的支持是我们更新的动力！