基于Hadoop的日志收集框架---Chukwa的源码分析(数据处理)

我就是我 2022-03-20 08:54 189阅读 0赞

1.工具类、接口简介  
  
(1)

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.     // 用于对数据进行分类   
2.     org.apache.hadoop.chukwa.extraction.demux.DemuxManager   
3.        
4.     // mapreduce程序的map处理接口   
5.     org.apache.hadoop.chukwa.extraction.demux.processor.mapper.MapProcessor   
6.        
7.     // MapProcessor的工厂类, 用HashMap存储了数据类型与其对应的MapProcessor  
8.     org.apache.hadoop.chukwa.extraction.demux.processor.mapper.MapProcessorFactory   
9.        
10.    // map处理接口的实现类, 实现了数据块(ChunkImpl)处理、逐行获取内容、创建ChukwaRecord对象,  
11.    // 但是将写入OutputCollector的实现放在了抽象方法parse()的实现类中  
12.    org.apache.hadoop.chukwa.extraction.demux.processor.mapper.AbstractProcessor   
13.   
14.    // AbstractProcessor中parse()方法的实现类    
15.    org.apache.hadoop.chukwa.extraction.demux.processor.mapper.DefaultProcessor   
16.   
17.    // mapreduce程序的reduce处理接口   
18.    org.apache.hadoop.chukwa.extraction.demux.processor.reducer.ReduceProcessor   
19.   
20.    // reduce处理接口的实现类, 实现了将key、value写入OutputCollector的过程  
21.    org.apache.hadoop.chukwa.extraction.demux.processor.reducer.IdentityReducer

// 用于对数据进行分类
       org.apache.hadoop.chukwa.extraction.demux.DemuxManager
       
       // mapreduce程序的map处理接口
       org.apache.hadoop.chukwa.extraction.demux.processor.mapper.MapProcessor
       
       // MapProcessor的工厂类, 用HashMap存储了数据类型与其对应的MapProcessor
       org.apache.hadoop.chukwa.extraction.demux.processor.mapper.MapProcessorFactory
       
       // map处理接口的实现类, 实现了数据块(ChunkImpl)处理、逐行获取内容、创建ChukwaRecord对象,
       // 但是将写入OutputCollector的实现放在了抽象方法parse()的实现类中
       org.apache.hadoop.chukwa.extraction.demux.processor.mapper.AbstractProcessor
    
       // AbstractProcessor中parse()方法的实现类
       org.apache.hadoop.chukwa.extraction.demux.processor.mapper.DefaultProcessor
    
       // mapreduce程序的reduce处理接口
       org.apache.hadoop.chukwa.extraction.demux.processor.reducer.ReduceProcessor
    
       // reduce处理接口的实现类, 实现了将key、value写入OutputCollector的过程
       org.apache.hadoop.chukwa.extraction.demux.processor.reducer.IdentityReducer

(2)

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.     // 用于对数据进行归档   
2.     org.apache.hadoop.chukwa.extraction.archive.ChukwaArchiveManager

// 用于对数据进行归档
       org.apache.hadoop.chukwa.extraction.archive.ChukwaArchiveManager

(3)

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  // 用于将分类后的数据转储备份   
2.  org.apache.hadoop.chukwa.extraction.demux.PostProcessorManager

// 用于将分类后的数据转储备份
       org.apache.hadoop.chukwa.extraction.demux.PostProcessorManager

(4)

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  // 用于将备份数据按小时进行合并   
2.  org.apache.hadoop.chukwa.extraction.demux.HourlyChukwaRecordRolling

// 用于将备份数据按小时进行合并
       org.apache.hadoop.chukwa.extraction.demux.HourlyChukwaRecordRolling

(5)

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  // 用于将备份数据按日期进行合并   
2.  org.apache.hadoop.chukwa.extraction.demux.DailyChukwaRecordRolling

// 用于将备份数据按日期进行合并
       org.apache.hadoop.chukwa.extraction.demux.DailyChukwaRecordRolling

2.处理流程  
  
(1)数据分类  
    org.apache.hadoop.chukwa.extraction.demux.DemuxManager

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 创建守护进程"DemuxManager",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程 
3.   \* 调用start方法启动循环处理过程  
4.   \*/  
5.  public static void main(String\[\] args)   
6.    
7.  /\*\*  
8.   \* 1. 初始化处理过程中用到的目录  
9.   \*    chukwaRootDir     /chukwa/  
10.  \*    demuxRootDir      /chukwa/demuxProcessing/   
11.  \*    demuxErrorDir     /chukwa/demuxProcessing/inError/ 
12.  \*    demuxInputDir     /chukwa/demuxProcessing/mrInput/ 
13.  \*    demuxOutputDir    /chukwa/demuxProcessing/mrOutput/ 
14.  \*    dataSinkDir       /chukwa/logs/  
15.  \*    postProcessDir    /chukwa/postProcess/   
16.  \*    archiveRootDir    /chukwa/dataSinkArchives/ 
17.  \* 2. 从系统配置文件中获取nagios的配置信息  
18.  \* 3. 循环处理  
19.  \*    统计异常次数，如果超过五次则退出  
20.  \*    检测demuxOutputDir是否存在，如果存在则将其删除  
21.  \*    检测demuxInputDir是否存在，如果存在则设置可以处理并将处理次数加1，如果已经尝试过3次则将则将demuxInputDir转移到demuxErrorDir 
22.  \*    如果以上两个目录都不存在，则：  
23.  \*       reprocess归0  
24.  \*       在dataSinkDir目录下检测是否有.done文件，如果存在则将其移动到demuxInputDir目录下（最多移动500个），移动完成后将demuxReady设为true 
25.  \*    如果demuxReady已经为true,则开始处理数据(processData)   
26.  \*    发送demux信息到nagios    
27.  \*/  
28. public void start()   
29.   
30. /\*\*  
31.  \* 1. 运行Demux,设置reduce数为demuxReducerCount,设置输入输出目录分别为demuxInputDir、demuxOutputDir 
32.  \* 2. 检测demuxOutputDir是否存在，存在则将demuxOutputDir转移到postProcessDir 
33.  \* 3. 将demuxInputDir转移到archiveDir(archiveRootDir) 
34.  \*/  
35. protected boolean processData(String dataSinkDir, String demuxInputDir, String demuxOutputDir, String postProcessDir, String archiveDir)   
36.   
37. /\*\*  
38.  \* 在main方法中使用ToolRunner启动Demux, 调用内部类MapClass、ReduceClass进行mapreduce处理 
39.  \* 在运行时设置输出的格式化类为ChukwaRecordOutputFormat   
40.  \*/  
41. public static void main(String\[\] args)

/**
     * 创建守护进程"DemuxManager",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程
     * 调用start方法启动循环处理过程
     */
    public static void main(String[] args)
    
    /**
     * 1. 初始化处理过程中用到的目录
     *    chukwaRootDir 	/chukwa/
     *	  demuxRootDir		/chukwa/demuxProcessing/
     *	  demuxErrorDir		/chukwa/demuxProcessing/inError/
     *	  demuxInputDir		/chukwa/demuxProcessing/mrInput/
     *	  demuxOutputDir	/chukwa/demuxProcessing/mrOutput/
     *	  dataSinkDir		/chukwa/logs/
     *	  postProcessDir	/chukwa/postProcess/
     *	  archiveRootDir	/chukwa/dataSinkArchives/
     * 2. 从系统配置文件中获取nagios的配置信息
     * 3. 循环处理
     *	  统计异常次数，如果超过五次则退出
     *	  检测demuxOutputDir是否存在，如果存在则将其删除
     *	  检测demuxInputDir是否存在，如果存在则设置可以处理并将处理次数加1，如果已经尝试过3次则将则将demuxInputDir转移到demuxErrorDir
     * 	  如果以上两个目录都不存在，则：
     *	  	 reprocess归0
     *		 在dataSinkDir目录下检测是否有.done文件，如果存在则将其移动到demuxInputDir目录下（最多移动500个），移动完成后将demuxReady设为true
     *    如果demuxReady已经为true,则开始处理数据(processData)
     *    发送demux信息到nagios 	
     */
    public void start()
    
    /**
     * 1. 运行Demux,设置reduce数为demuxReducerCount,设置输入输出目录分别为demuxInputDir、demuxOutputDir
     * 2. 检测demuxOutputDir是否存在，存在则将demuxOutputDir转移到postProcessDir
     * 3. 将demuxInputDir转移到archiveDir(archiveRootDir)
     */
    protected boolean processData(String dataSinkDir, String demuxInputDir, String demuxOutputDir, String postProcessDir, String archiveDir)
    
    /**
     * 在main方法中使用ToolRunner启动Demux, 调用内部类MapClass、ReduceClass进行mapreduce处理
     * 在运行时设置输出的格式化类为ChukwaRecordOutputFormat
     */
    public static void main(String[] args)

org.apache.hadoop.chukwa.extraction.demux.Demux  
    在Demux中实现了对数据分类的Mapper类和Reducer类  
  
    org.apache.hadoop.chukwa.extraction.demux.Demux$MapClass

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 根据ChunkImpl的datatype在$CHUKWA\_HOME/conf/chukwa-demux-conf.xml中查找是否配置了mapProcessorClass 
3.   \* 如果没有则默认使用DefaultProcessor处理  
4.   \* 将获取的mapProcessorClass添加到mapProcessor工厂类的hashmap中 
5.   \* 在AbstractProcessor中实现了process方法  
6.   \*    获取数据块的相关信息  
7.   \*    从数据块中逐行读取内容  
8.   \*    将处理行内容的parse方法实现放在了DefaultProcessor中, 在此方法中调用父类的buildGenericRecord方法完成 
9.   \*    ChukwaArchiveKey  ->   ChukwaRecordKey   
10.  \*    ChunkImpl         ->   ChukwaRecord   
11.  \*    的转换  
12.  \*    写入OutputCollector  
13.  \*/  
14. public void map(ChukwaArchiveKey key, ChunkImpl chunk,   
15.                 OutputCollector<ChukwaRecordKey, ChukwaRecord> output,   
16.                 Reporter reporter)

/**
     * 根据ChunkImpl的datatype在$CHUKWA_HOME/conf/chukwa-demux-conf.xml中查找是否配置了mapProcessorClass
     * 如果没有则默认使用DefaultProcessor处理
     * 将获取的mapProcessorClass添加到mapProcessor工厂类的hashmap中
     * 在AbstractProcessor中实现了process方法
     *	  获取数据块的相关信息
     *	  从数据块中逐行读取内容
     *	  将处理行内容的parse方法实现放在了DefaultProcessor中, 在此方法中调用父类的buildGenericRecord方法完成
     *	  ChukwaArchiveKey 	-> 	ChukwaRecordKey
     *	  ChunkImpl			-> 	ChukwaRecord
     *	  的转换
     *	  写入OutputCollector
     */
    public void map(ChukwaArchiveKey key, ChunkImpl chunk,
    				OutputCollector<ChukwaRecordKey, ChukwaRecord> output,
    				Reporter reporter)

org.apache.hadoop.chukwa.extraction.demux.Demux$ReduceClass

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 根据ChukwaRecordKey的reduceType来确定reduceProcessorClass 
3.   \* org.apache.hadoop.chukwa.extraction.demux.processor.reducer. + reduceType 
4.   \* 如果没有找到类则使用IdentityReducer  
5.   \* 在IdentityReducer的process方法中只是将key、value写入OutputCollector 
6.   \*/  
7.  public void reduce(ChukwaRecordKey key, Iterator<ChukwaRecord> values,   
8.                  OutputCollector<ChukwaRecordKey, ChukwaRecord> output,   
9.                  Reporter reporter)

/**
     * 根据ChukwaRecordKey的reduceType来确定reduceProcessorClass
     * org.apache.hadoop.chukwa.extraction.demux.processor.reducer. + reduceType
     * 如果没有找到类则使用IdentityReducer
     * 在IdentityReducer的process方法中只是将key、value写入OutputCollector
     */
    public void reduce(ChukwaRecordKey key, Iterator<ChukwaRecord> values,
    				OutputCollector<ChukwaRecordKey, ChukwaRecord> output,
    				Reporter reporter)

org.apache.hadoop.chukwa.extraction.demux.ChukwaRecordOutputFormat  
    继承自MultipleSequenceFileOutputFormat，用于格式化输出结果

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 使输出目录为clusterName/dataType/dataType+Util.generateTimeOutput(record.getTime()) 
3.   \*/  
4.  protected String generateFileNameForKeyValue(ChukwaRecordKey key,   
5.              ChukwaRecord record, String name)

/**
     * 使输出目录为clusterName/dataType/dataType+Util.generateTimeOutput(record.getTime())
     */
    protected String generateFileNameForKeyValue(ChukwaRecordKey key,
    			ChukwaRecord record, String name)

org.apache.hadoop.chukwa.extraction.demux.processor.Util

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 根据时间戳构造文件名  
3.   \* 从timestamp中获取workingDay、workingHour、minutes   
4.   \* 如果运行日期不等于当前日期，则输出文件为  
5.   \*    \[dataType\]\_\[yyyyMMdd\].D.evt  
6.   \* 否则如果运行时间（小时）不等于当前时间（小时），则输出文件为  
7.   \*    \[dataType\]\_\[yyyyMMdd\]\_\[HH\].H.evt  
8.   \* 否则输出结果文件为  
9.   \*    分钟的个位数小于5，  
10.  \*       \[dataType\]\_\[yyyyMMdd\]\_\[HH\]\_\[minutes/10\].0.R.evt 
11.  \*    分钟的个位数大于等于5  
12.  \*       \[dataType\]\_\[yyyyMMdd\]\_\[HH\]\_\[minutes/10\].5.R.evt 
13.  \*/  
14. public static String generateTimeOutput(long timestamp)

/**
     * 根据时间戳构造文件名
     * 从timestamp中获取workingDay、workingHour、minutes
     * 如果运行日期不等于当前日期，则输出文件为
     *	  [dataType]_[yyyyMMdd].D.evt
     * 否则如果运行时间（小时）不等于当前时间（小时），则输出文件为
     *	  [dataType]_[yyyyMMdd]_[HH].H.evt
     * 否则输出结果文件为
     *	  分钟的个位数小于5，
     *		 [dataType]_[yyyyMMdd]_[HH]_[minutes/10].0.R.evt
     *    分钟的个位数大于等于5
     *		 [dataType]_[yyyyMMdd]_[HH]_[minutes/10].5.R.evt
     */
    public static String generateTimeOutput(long timestamp)

(2)数据归档  
    org.apache.hadoop.chukwa.extraction.archive.ChukwaArchiveManager

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 创建守护进程"DemuxManager",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程 
3.   \* 调用start方法启动循环处理过程  
4.   \*/  
5.  public static void main(String\[\] args)   
6.    
7.  /\*\*  
8.   \* 1. 初始化处理过程中用到的目录  
9.   \*    chukwaRootDir                 /chukwa/   
10.  \*    archiveRootDir                /chukwa/dataSinkArchives/ 
11.  \*    archivesRootProcessingDir     /chukwa/archivesProcessing/ 
12.  \*    archivesMRInputDir            /chukwa/archivesProcessing/mrInput/ 
13.  \*    archivesMROutputDir           /chukwa/archivesProcessing/mrOutput/ 
14.  \*    finalArchiveOutput            /chukwa/finalArchives/ 
15.  \* 2. 开始循环处理  
16.  \*      errorCount记录错误次数，当超过四次时退出  
17.  \*      判断archivesMRInputDir目录是否存在(是否已经有archive job的输入目录存在)，存在则使用runArchive方法对当天数据进行归档 
18.  \*      获取archiveRootDir目录下的文件，  
19.  \*      如果没有文件则休眠一分钟  
20.  \*      如果只有一个日期目录，则判断是否是当天且当前时间是否与上次运行间隔了两个小时，如果还不到两个小时则休眠半个小时 
21.  \*      如果有多个日期目录，则使用processDay方法处理每个日期目录下的文件   
22.  \*/  
23. public void start()   
24.   
25. /\*\*  
26.  \* 处理指定日期的数据  
27.  \*     now              当前时间          
28.  \*     currentDay       当前日期  
29.  \*     workingDay       要处理的日期目录  
30.  \*     oneHourAgo       一个小时之前的时间  
31.  \* 如果要处理的日期目录下没有文件且日期在当天日期之前，则将目录删除后返回  
32.  \* 使用fileCount记录文件总数, 遍历日期目录  
33.  \*     如果文件修改时间不到一个小时  
34.  \*         记录日期目录下的文件数  
35.  \*         将文件转移到archivesMRInputDir目录  
36.  \*         如果文件数已经到达最大文件数(500)，则调用runArchive方法进行归档，并将fileCount重置为0 
37.  \*/  
38. public void processDay(FileStatus fsDay, String archivesMRInputDir,   
39.             String archivesMROutputDir, String finalArchiveOutput)   
40.   
41. /\*\*  
42.  \* 1. 构建job运行参数，从配置文件中加载配置项"archive.grouper"(归档分组类型) 
43.  \* 2. 检测mrOutput目录是否存在，存在则删除  
44.  \* 3. 使用ToolRunner运行ChukwaArchiveBuilder  
45.  \* 4. 使用mrInput中的日期在finalArchiveOutput中创建对应的目录   
46.  \* 5. 将mrOutput目录转储到上一步创建的目录下的"archive\_+当前时间"   
47.  \* 6. 删除mrInput目录  
48.  \*/  
49. public void runArchive(String archivesMRInputDir,   
50.             String archivesMROutputDir, String finalArchiveOutput)

/**
     * 创建守护进程"DemuxManager",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程
     * 调用start方法启动循环处理过程
     */
    public static void main(String[] args)
    
    /**
     * 1. 初始化处理过程中用到的目录
     *    chukwaRootDir 				/chukwa/
     *	  archiveRootDir				/chukwa/dataSinkArchives/
     *	  archivesRootProcessingDir		/chukwa/archivesProcessing/
     *	  archivesMRInputDir			/chukwa/archivesProcessing/mrInput/
     *	  archivesMROutputDir			/chukwa/archivesProcessing/mrOutput/
     *	  finalArchiveOutput			/chukwa/finalArchives/
     * 2. 开始循环处理
     *		errorCount记录错误次数，当超过四次时退出
     *		判断archivesMRInputDir目录是否存在(是否已经有archive job的输入目录存在)，存在则使用runArchive方法对当天数据进行归档
     *		获取archiveRootDir目录下的文件，
     *		如果没有文件则休眠一分钟
     *		如果只有一个日期目录，则判断是否是当天且当前时间是否与上次运行间隔了两个小时，如果还不到两个小时则休眠半个小时
     *		如果有多个日期目录，则使用processDay方法处理每个日期目录下的文件
     */
    public void start()
    
    /**
     * 处理指定日期的数据
     *     now				当前时间		
     *	   currentDay		当前日期
     *	   workingDay		要处理的日期目录
     *	   oneHourAgo		一个小时之前的时间
     * 如果要处理的日期目录下没有文件且日期在当天日期之前，则将目录删除后返回
     * 使用fileCount记录文件总数, 遍历日期目录
     * 	   如果文件修改时间不到一个小时
     *		   记录日期目录下的文件数
     *		   将文件转移到archivesMRInputDir目录
     *		   如果文件数已经到达最大文件数(500)，则调用runArchive方法进行归档，并将fileCount重置为0
     */
    public void processDay(FileStatus fsDay, String archivesMRInputDir,
    			String archivesMROutputDir, String finalArchiveOutput)
    
    /**
     * 1. 构建job运行参数，从配置文件中加载配置项"archive.grouper"(归档分组类型)
     * 2. 检测mrOutput目录是否存在，存在则删除
     * 3. 使用ToolRunner运行ChukwaArchiveBuilder
     * 4. 使用mrInput中的日期在finalArchiveOutput中创建对应的目录
     * 5. 将mrOutput目录转储到上一步创建的目录下的"archive_+当前时间"
     * 6. 删除mrInput目录
     */
    public void runArchive(String archivesMRInputDir,
    			String archivesMROutputDir, String finalArchiveOutput)

org.apache.hadoop.chukwa.extraction.archive.ChukwaArchiveBuilder  
  
    org.apache.hadoop.chukwa.extraction.archive.ChukwaArchiveBuilder$UniqueKeyReduce

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 统计每个key有多少个value  
3.   \*/    
4.  public void reduce(ChukwaArchiveKey key, Iterator<ChunkImpl> vals,   
5.                  OutputCollector<ChukwaArchiveKey, ChunkImpl> out, Reporter r)   
6.                                 
7.  /\*\*  
8.   \* 在main方法中使用ToolRunner启动ChukwaArchiveBuilder,   
9.   \* 在run方法中调用IdentityMapper、内部类UniqueKeyReduce进行mapreduce处理 
10.  \* 根据不同的参数设置不同的partitionerClass、outputFormat、jobName     
11.  \*/  
12. public static void main(String\[\] args)

/**
     * 统计每个key有多少个value
     */ 
    public void reduce(ChukwaArchiveKey key, Iterator<ChunkImpl> vals,
    				OutputCollector<ChukwaArchiveKey, ChunkImpl> out, Reporter r)
    							
    /**
     * 在main方法中使用ToolRunner启动ChukwaArchiveBuilder,
     * 在run方法中调用IdentityMapper、内部类UniqueKeyReduce进行mapreduce处理
     * 根据不同的参数设置不同的partitionerClass、outputFormat、jobName 	
     */
    public static void main(String[] args)

(3)数据转储  
    org.apache.hadoop.chukwa.extraction.demux.PostProcessorManager

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 创建守护进程"PostProcessorManager",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程 
3.   \* 调用start方法启动循环处理过程  
4.   \*/  
5.  public static void main(String\[\] args)   
6.    
7.  /\*\*  
8.   \* 1. 初始化处理过程中用到的目录  
9.   \*    chukwaRootDir                     /chukwa/   
10.  \*    postProcessDir                    /chukwa/postProcess/ 
11.  \*    chukwaRootReposDir                /chukwa/repos/ 
12.  \*    chukwaPostProcessInErrorDir       /chukwa/postProcessInError/ 
13.  \* 2. 开始循环处理  
14.  \*      errorCount记录了错误次数，如果已经超过四次则退出  
15.  \*      遍历demuxOutputDir,  
16.  \*      如果此目录下没有文件则休眠10秒  
17.  \*      如果有文件则将文件添加到directories(ArrayList<String>)并进行排序 
18.  \*      遍历directories  
19.  \*          对每一个文件使用processDemuxPigOutput方法进行处理，完成后调用 
20.  \*          movetoMainRepository方法移动文件，然后将文件删除，如果出现异常， 
21.  \*          则将文件转移到错误目录  
22.  \*/  
23. public void start()   
24.   
25. /\*\*  
26.  \* 从$CHUKWA\_HOME/conf/chukwa-demux-conf.xml中加载配置项 
27.  \*  chukwa.post.demux.data.loader，得到两个实现类   
28.  \*  org.apache.hadoop.chukwa.dataloader.MetricDataLoaderPool, 
29.  \*  org.apache.hadoop.chukwa.dataloader.FSMDataLoader 
30.  \*  遍历配置的实现类  
31.  \*      分别获取实例，构建查找对象  
32.  \*      (/chukwa/postProcess/cluster/dataType/\*.evt) 
33.  \*      将查找到的文件交由dataloader处理  
34.  \*/  
35. public boolean processDemuxPigOutput(String directory)   
36.   
37. /\*\*  
38.  \* 这里将移动到repository的功能整合到了org.apache.hadoop.chukwa.extraction.demux.MoveToRepository中 
39.  \*/  
40. public boolean movetoMainRepository(String sourceDirectory,   
41.             String repoRootDirectory)

/**
     * 创建守护进程"PostProcessorManager",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程
     * 调用start方法启动循环处理过程
     */
    public static void main(String[] args)
    
    /**
     * 1. 初始化处理过程中用到的目录
     *    chukwaRootDir 					/chukwa/
     *	  postProcessDir					/chukwa/postProcess/
     *	  chukwaRootReposDir				/chukwa/repos/
     *	  chukwaPostProcessInErrorDir		/chukwa/postProcessInError/
     * 2. 开始循环处理
     *		errorCount记录了错误次数，如果已经超过四次则退出
     *		遍历demuxOutputDir,
     *		如果此目录下没有文件则休眠10秒
     *		如果有文件则将文件添加到directories(ArrayList<String>)并进行排序
     *		遍历directories
     *			对每一个文件使用processDemuxPigOutput方法进行处理，完成后调用
     *			movetoMainRepository方法移动文件，然后将文件删除，如果出现异常，
     *			则将文件转移到错误目录
     */
    public void start()
    
    /**
     * 从$CHUKWA_HOME/conf/chukwa-demux-conf.xml中加载配置项
     *	chukwa.post.demux.data.loader，得到两个实现类
     *	org.apache.hadoop.chukwa.dataloader.MetricDataLoaderPool,
     *	org.apache.hadoop.chukwa.dataloader.FSMDataLoader
     *	遍历配置的实现类
     *		分别获取实例，构建查找对象
     *		(/chukwa/postProcess/cluster/dataType/*.evt)
     *		将查找到的文件交由dataloader处理
     */
    public boolean processDemuxPigOutput(String directory)
    
    /**
     * 这里将移动到repository的功能整合到了org.apache.hadoop.chukwa.extraction.demux.MoveToRepository中
     */
    public boolean movetoMainRepository(String sourceDirectory,
    			String repoRootDirectory)

org.apache.hadoop.chukwa.extraction.demux.MoveToRepository

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 遍历要转储的目录  
3.   \*    获取cluster名称，过滤掉"\_logs"、"\_temporary"   
4.   \*    使用processClusterDirectory方法处理每个cluster下的文件   
5.   \*    删除已经完成转储的cluster目录  
6.   \*/  
7.  public static void main(String\[\] args)   
8.    
9.  /\*\*  
10.  \* 遍历cluster目录  
11.  \*    获取dataType名称  
12.  \*    使用processDatasourceDirectory方法处理每个dataType下的文件 
13.  \*/  
14. static void processClusterDirectory(Path srcDir, String destDir)   
15.   
16. /\*\*  
17.  \* 遍历某个dataType下的evt文件  
18.  \*    如果是以".D.evt"结尾的文件  
19.  \*        获取日期  
20.  \*        调用writeRecordFile方法写入文件，其中  
21.  \*        输出目录为/chukwa/repos/<cluster>/<dataType>/<yyyyMMdd>/ 
22.  \*        输出文件为<dataType>\_<yyyyMMdd>  
23.  \*    如果是以".H.evt"结尾的文件  
24.  \*        获取日期、小时  
25.  \*        调用writeRecordFile方法写入文件，其中  
26.  \*        输出目录为/chukwa/repos/<cluster>/<dataType>/<yyyyMMdd>/<HH>/ 
27.  \*        输出文件为<dataType>\_<yyyyMMdd>\_<HH>   
28.  \*        调用addDirectory4Rolling方法对日期、小时做标记   
29.  \*    如果是以".R.evt"结尾的文件  
30.  \*        获取日期、小时、分钟  
31.  \*        调用writeRecordFile方法写入文件，其中  
32.  \*        输出目录为/chukwa/repos/<cluster>/<dataType>/<yyyyMMdd>/<HH>/<mm>/ 
33.  \*        输出文件为<dataType>\_<yyyyMMdd>\_<HH>\_<mm>   
34.  \*        调用addDirectory4Rolling方法对日期、小时做标记   
35.  \*/  
36. static void processDatasourceDirectory(String cluster, Path srcDir, String destDir)   
37.   
38. /\*\*  
39.  \* 将文件转储并改名  
40.  \*/  
41. static void writeRecordFile(String destDir, Path recordFile, String fileName)   
42.   
43. /\*\*  
44.  \* 在/chukwa/rolling/目录下生成相应数据类型的目录以便于进行数据合并   
45.  \* 如果不在当前日期，在/chukwa/rolling/下生成  
46.  \*    daily/<yyyyMMdd>/<cluster>/<dataType>   
47.  \* 否则，在/chukwa/rolling/下生成  
48.  \*    hourly/<yyyyMMdd>/<cluster>/<dataType>     
49.  \*/  
50. static void addDirectory4Rolling(boolean isDailyOnly, int day, int hour,   
51.             String cluster, String dataSource)

/**
     * 遍历要转储的目录
     *	  获取cluster名称，过滤掉"_logs"、"_temporary"
     *	  使用processClusterDirectory方法处理每个cluster下的文件
     *	  删除已经完成转储的cluster目录
     */
    public static void main(String[] args)
    
    /**
     * 遍历cluster目录
     *	  获取dataType名称
     *	  使用processDatasourceDirectory方法处理每个dataType下的文件
     */
    static void processClusterDirectory(Path srcDir, String destDir)
    
    /**
     * 遍历某个dataType下的evt文件
     *	  如果是以".D.evt"结尾的文件
     *		  获取日期
     *		  调用writeRecordFile方法写入文件，其中
     *		  输出目录为/chukwa/repos/<cluster>/<dataType>/<yyyyMMdd>/
     *	      输出文件为<dataType>_<yyyyMMdd>
     *	  如果是以".H.evt"结尾的文件
     *		  获取日期、小时
     *		  调用writeRecordFile方法写入文件，其中
     *		  输出目录为/chukwa/repos/<cluster>/<dataType>/<yyyyMMdd>/<HH>/
     *		  输出文件为<dataType>_<yyyyMMdd>_<HH>
     *		  调用addDirectory4Rolling方法对日期、小时做标记
     *	  如果是以".R.evt"结尾的文件
     *		  获取日期、小时、分钟
     *		  调用writeRecordFile方法写入文件，其中
     *		  输出目录为/chukwa/repos/<cluster>/<dataType>/<yyyyMMdd>/<HH>/<mm>/
     *		  输出文件为<dataType>_<yyyyMMdd>_<HH>_<mm>
     *		  调用addDirectory4Rolling方法对日期、小时做标记
     */
    static void processDatasourceDirectory(String cluster, Path srcDir, String destDir)
    
    /**
     * 将文件转储并改名
     */
    static void writeRecordFile(String destDir, Path recordFile, String fileName)
    
    /**
     * 在/chukwa/rolling/目录下生成相应数据类型的目录以便于进行数据合并
     * 如果不在当前日期，在/chukwa/rolling/下生成
     *	  daily/<yyyyMMdd>/<cluster>/<dataType>
     * 否则，在/chukwa/rolling/下生成
     *	  hourly/<yyyyMMdd>/<cluster>/<dataType>	
     */
    static void addDirectory4Rolling(boolean isDailyOnly, int day, int hour,
    			String cluster, String dataSource)

(4)数据合并  
    org.apache.hadoop.chukwa.extraction.demux.HourlyChukwaRecordRolling

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 1. 创建守护进程"PostProcessorManager",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程 
3.   \* 2. 初始化处理过程中用到的目录  
4.   \*    rollingFolder             /chukwa/rolling/   
5.   \*    chukwaMainRepository      /chukwa/repos/   
6.   \*    tempDir                   /chukwa/temp/hourlyRolling/  
7.   \* 3. 从执行参数中获取rollInSequence(是否顺序执行合并线程),deleteRawdata(是否删除原始数据)这两个参数设置的值 
8.   \* 4. 获取当前日期和小时  
9.   \* 5. 遍历 /chukwa/rolling/hourly/ 目录  
10.  \*       获取此目录下的每个日期，对其遍历  
11.  \*          获取此目录下的小时，如果此目录日期小于当前日期，或两者相同而且此小时小于当前小时(即只对当前小时之前的数据进行合并) 
12.  \*          调用buildHourlyFiles  
13.  \*/  
14. public static void main(String\[\] args)   
15.   
16. /\*\*  
17.  \* 获取处理目录 /chukwa/rolling/hourly/workingDay/workingHour 
18.  \* 遍历此目录下所有的cluster  
19.  \*    遍历每个cluster目录下的所有数据类型(dataSource)  
20.  \*       创建目录 /chukwa/repos/cluster/dataSource/workingDay/workingHour/rotateDone 
21.  \*       构建合并参数数组 mergeArgs  
22.  \*       mergeArgs\[0\] = /chukwa/repos/cluster/dataSource/workingDay/workingHour/\[0-5\]\* /\*.evt 
23.  \*       mergeArgs\[1\] = /chukwa/temp/hourlyRolling/cluster/dataSource/workingDay/workingHour\_currentTimeMillis 
24.  \*       mergeArgs\[2\] = /chukwa/repos/cluster/dataSource/workingDay/workingHour 
25.  \*       mergeArgs\[3\] = dataSource\_HourlyDone\_workingDay\_workingHour 
26.  \*       mergeArgs\[4\] = /chukwa/rolling/hourly/workingDay/workingHour/cluster/dataSource 
27.  \*       新建RecordMerger对象merge  
28.  \*       如果rollInSequence为true,则启动merge线程   
29.  \*       否则将merge添加到列表并启动线程，等待此线程运行完成  
30.  \*       删除dataSource目录  
31.  \*    删除cluster目录  
32.  \* 删除hour目录  
33.  \*/  
34. public static void buildHourlyFiles(String chukwaMainRepository,   
35.             String tempDir, String rollingFolder, int workingDay,   
36.             int workingHour)   
37.                
38. /\*\*  
39.  \* 使用IdentityMapper作为mapperClass, 使用IdentityReducer作为reducerClass, 只是简单的合并数据 
40.  \*/  
41. public int run(String\[\] args)

/**
     * 1. 创建守护进程"PostProcessorManager",将进程号写入相应的pid文件，以便于运行stop命令时可根据此pid文件杀死进程
     * 2. 初始化处理过程中用到的目录
     * 	  rollingFolder 			/chukwa/rolling/
     *	  chukwaMainRepository    	/chukwa/repos/
     * 	  tempDir                 	/chukwa/temp/hourlyRolling/	
     * 3. 从执行参数中获取rollInSequence(是否顺序执行合并线程),deleteRawdata(是否删除原始数据)这两个参数设置的值
     * 4. 获取当前日期和小时
     * 5. 遍历 /chukwa/rolling/hourly/ 目录
     *   	 获取此目录下的每个日期，对其遍历
     *       	获取此目录下的小时，如果此目录日期小于当前日期，或两者相同而且此小时小于当前小时(即只对当前小时之前的数据进行合并)
     *          调用buildHourlyFiles
     */
    public static void main(String[] args)
    
    /**
     * 获取处理目录 /chukwa/rolling/hourly/workingDay/workingHour
     * 遍历此目录下所有的cluster
     *    遍历每个cluster目录下的所有数据类型(dataSource)
     *       创建目录 /chukwa/repos/cluster/dataSource/workingDay/workingHour/rotateDone
     *       构建合并参数数组 mergeArgs
     *       mergeArgs[0] = /chukwa/repos/cluster/dataSource/workingDay/workingHour/[0-5]* /*.evt
     *       mergeArgs[1] = /chukwa/temp/hourlyRolling/cluster/dataSource/workingDay/workingHour_currentTimeMillis
     *       mergeArgs[2] = /chukwa/repos/cluster/dataSource/workingDay/workingHour
     *       mergeArgs[3] = dataSource_HourlyDone_workingDay_workingHour
     *       mergeArgs[4] = /chukwa/rolling/hourly/workingDay/workingHour/cluster/dataSource
     *       新建RecordMerger对象merge
     *       如果rollInSequence为true,则启动merge线程
     *       否则将merge添加到列表并启动线程，等待此线程运行完成
     *       删除dataSource目录
     *    删除cluster目录
     * 删除hour目录
     */
    public static void buildHourlyFiles(String chukwaMainRepository,
    			String tempDir, String rollingFolder, int workingDay,
    			int workingHour)
    			
    /**
     * 使用IdentityMapper作为mapperClass, 使用IdentityReducer作为reducerClass, 只是简单的合并数据
     */
    public int run(String[] args)

org.apache.hadoop.chukwa.extraction.demux.RecordMerger

Java代码 [ ![复制代码][icon_copy.gif]][_icon_copy.gif]  ![收藏代码][icon_star.png]![spinner.gif][]

1.  /\*\*  
2.   \* 调用ToolRunner.run运行tool, 即调用HourlyChukwaRecordRolling中的run方法，将 
3.   \* /chukwa/repos/cluster/dataSource/workingDay/workingHour/\[0-5\]\* /\*.evt 
4.   \* 合并到  
5.   \* /chukwa/temp/hourlyRolling/cluster/dataSource/workingDay/workingHour\_currentTimeMillis 
6.   \* 如果此任务处理成功  
7.   \*    调用 writeRecordFile()  
8.   \*    如果deleteRawData参数为true  
9.   \*       删除输入文件  
10.  \*       /chukwa/repos/cluster/dataSource/workingDay/workingHour/\[0-5\]\* /\*.evt 
11.  \*       遍历 /chukwa/repos/cluster/dataSource/workingDay/workingHour 
12.  \*          删除小时或分钟的目录  
13.  \*          /chukwa/repos/cluster/dataSource/workingDay/workingHour\[0-5\]\* / 
14.  \*    删除合并标记文件  
15.  \*    dataSource\_HourlyDone\_workingDay\_workingHour 
16.  \*    删除临时目录  
17.  \*    /chukwa/temp/hourlyRolling/cluster/dataSource/workingDay/workingHour\_currentTimeMillis 
18.  \*/  
19. public void run()   
20.   
21. /\*\*  
22.  \* 将mapreduce运行结果文件改名转储, 即将HourlyChukwaRecordRolling中的run方法合并后的文件转储 
23.  \* input        /chukwa/temp/hourlyRolling/cluster/dataSource/workingDay/workingHour\_currentTimeMillis/part-00000 
24.  \* outputDir    /chukwa/repos/cluster/dataSource/workingDay/workingHour 
25.  \* filename     dataSource\_HourlyDone\_workingDay\_workingHour 
26.  \* destFile     /chukwa/repos/cluster/dataSource/workingDay/workingHour/dataSource\_HourlyDone\_workingDay\_workingHour.1.evt 
27.  \*/  
28. void writeRecordFile(String input, String outputDir, String fileName)

[icon_copy.gif]: http://savagegarden.iteye.com/images/icon_copy.gif
[_icon_copy.gif]: http://savagegarden.iteye.com/blog/1441842#
[icon_star.png]: http://savagegarden.iteye.com/images/icon_star.png
[spinner.gif]: /images/20220320/24ac8f0fe43f4586ba5885a853297277.png