hadoop学习笔记--找到执行hadoop的入口

参与个hadoop项目，之前没搞过，赶紧学习：

照葫芦画瓢，得到代码是hdfs2local.sh脚本和LiaoNingFilter.jar包，迫不及待用jd-gui打开jar包，搜索到main（在MANIFEST.MF中没有找到main，只能search，其实在hdfs2local.sh脚本中写明了main所在的package）。

package cn.com.dtmobile.hadoop.biz.LiaoNingFilter.job;

import cn.com.dtmobile.hadoop.biz.LiaoNingFilter.mr.DataMap;
import cn.com.dtmobile.hadoop.biz.LiaoNingFilter.mr.DataReduce;
import java.net.URI;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
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.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class DataJob
  extends Configured
  implements Tool
{
  public int run(String[] args)
    throws Exception
  {
    Configuration conf = getConf();
    conf.setBoolean("mapred.output.compress", true);
    conf.setClass("mapred.output.compression.codec", GzipCodec.class, CompressionCodec.class);
    
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    
    String sysHour = otherArgs[10];
    
    DistributedCache.addCacheFile(new URI(otherArgs[9]), conf);
    Job job = new Job(conf, "LiaoNingFilter");
    job.getConfiguration().setStrings("sys_hour", new String[] { sysHour });
    job.setJarByClass(DataJob.class);
    job.setMapperClass(DataMap.class);
    job.setReducerClass(DataReduce.class);
    job.setNumReduceTasks(100);
    
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(Text.class);
    
    job.setOutputKeyClass(NullWritable.class);
    job.setOutputValueClass(Text.class);
    
    MultipleOutputs.addNamedOutput(job, "volterx", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "s1mmeorgn", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "voltesv", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "sgsorgn", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "s1uhttporgn", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "volteorgn", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "sharevolterx", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "shares1mme", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "sharehttp", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "sharevolteorgn", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "voltemossession", TextOutputFormat.class, NullWritable.class, Text.class);
    
    MultipleOutputs.addNamedOutput(job, "voltemosslice", TextOutputFormat.class, NullWritable.class, Text.class);
    
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    
    FileInputFormat.addInputPath(job, new Path(otherArgs[1]));
    
    FileInputFormat.addInputPath(job, new Path(otherArgs[2]));
    
    FileInputFormat.addInputPath(job, new Path(otherArgs[3]));
    
    FileInputFormat.addInputPath(job, new Path(otherArgs[4]));
    
    FileInputFormat.addInputPath(job, new Path(otherArgs[5]));
    
    FileInputFormat.addInputPath(job, new Path(otherArgs[6]));
    
    FileInputFormat.addInputPath(job, new Path(otherArgs[7]));
    
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[8]));
    
    return job.waitForCompletion(true) ? 0 : 1;
  }
  
  public static void main(String[] args)
    throws Exception
  {
    try
    {
      int returnCode = ToolRunner.run(new DataJob(), args);
      System.exit(returnCode);
    }
    catch (Exception e)
    {
      e.printStackTrace();
    }
  }

class DataJob76~92行：

这是传递参数给hadoop，但args到底是什么，otherArgs从run(String[] args)来的，run(String[] args)从main(String[] args)来的，main(String[] args)需要回到调用jar包的hdfs2local.sh脚本中找：

 #!/bin/bash
# 
 export HADOOP_CONF_DIR=/etc/hadoop/conf
 
#ANALY_DATE=20180619
#ANALY_HOUR=13
ANALY_DATE=`date +%Y%m%d`
ANALY_HOUR="`date -d ' -1 hour' +%H`"
CUR_DATE=`date +%Y%m%d`
CUR_HOUR="`date +%H`"
#CUR_DATE=20180619
#CUR_HOUR=13
if [ $CUR_HOUR = 00 ]
then
   ANALY_DATE="`date -d ' -1 day' +%Y%m%d`"
else
   ANALY_DATE=$CUR_DATE
fi
 mypath="$(cd "$(dirname "$0")"; pwd)"
 cd $mypath
 #SOURCE_SVR="hdfs://nameservice1:8020/datang"
 SOURCE_SVR="hdfs://nameservice1:8020/ws/detail"
 
 JAR_PACKAGK="/cup/d6/datang/bin/LiaoNingFilter.jar"
 JAR_MAIN="cn.com.dtmobile.hadoop.biz.LiaoNingFilter.job.DataJob"
 
 HIVE_TABLES="s1mme_orgn sgs_orgn volte_rx volte_sv s1u_http_orgn volte_mos_session volte_mos_slice"
 
 GX=${SOURCE_SVR}/volte_rx/p1_day=${ANALY_DATE}/*${ANALY_DATE}${ANALY_HOUR}*.ok1
 S1=${SOURCE_SVR}/s1mme_orgn/p1_day=${ANALY_DATE}/*${ANALY_DATE}${ANALY_HOUR}*.txt
 SV=${SOURCE_SVR}/volte_sv/p1_day=${ANALY_DATE}/*${ANALY_DATE}${ANALY_HOUR}*.ok1
 SGS=${SOURCE_SVR}/sgs_orgn/p1_day=${ANALY_DATE}/*${ANALY_DATE}${ANALY_HOUR}*.ok1
 MW=${SOURCE_SVR}/volte_sip/p1_day=${ANALY_DATE}/*${ANALY_DATE}${ANALY_HOUR}*.ok1
 HTTP=${SOURCE_SVR}/s1u_http_orgn/p1_day=${ANALY_DATE}/*${ANALY_DATE}${ANALY_HOUR}*.txt
 SESSION=${SOURCE_SVR}/volte_mos_session/p1_day=${ANALY_DATE}/*${ANALY_DATE}${ANALY_HOUR}*.ok1
 SLICE=${SOURCE_SVR}/volte_mos_slice/p1_day=${ANALY_DATE}/*${ANALY_DATE}${ANALY_HOUR}*.ok1
 OUTPUT=/datang/FILTER/${ANALY_DATE}${ANALY_HOUR}
 NODATA=/datang/nodata
 hdfs dfs -test -e ${GX}
 if [ $? -ne 0 ];then
 GX=${NODATA}
 fi
 hdfs dfs -test -e ${S1}
 if [ $? -ne 0 ];then
 S1=${NODATA}
 fi
 hdfs dfs -test -e ${SV}
 if [ $? -ne 0 ];then
 SV=${NODATA}
 fi
 hdfs dfs -test -e ${SGS}
 if [ $? -ne 0 ]; then
 SGS=${NODATA}
 fi
 hdfs dfs -test -e ${MW}
 if [ $? -ne 0 ]; then
 MW=${NODATA}
 fi
 hdfs dfs -test -e ${HTTP}
 if [ $? -ne 0 ]; then
 HTTP=${NODATA}
 fi
 hdfs dfs -test -e ${SESSION}
 if [ $? -ne 0 ];then
 SESSION=${NODATA}
 fi
 hdfs dfs -test -e ${SLICE}
 if [ $? -ne 0 ];then
 SLICE=${NODATA}
 fi

 T_PROCESS=/datang/t_process/t_process.csv
 echo "`date '+/%y/%m/%d %H:%M:%S'` INFO begni,exit..."
 HADOOP=`which hadoop`
 #${HADOOP} fs -rm -R ${OUTPUT}
 ${HADOOP} jar ${JAR_PACKAGK} ${JAR_MAIN} 
 ${GX} 
 ${S1} 
 ${SV} 
 ${SGS} 
 ${MW} 
 ${HTTP} 
 ${SESSION} 
 ${SLICE} 
 ${OUTPUT} 
 ${T_PROCESS} 
 ${ANALY_HOUR}
 
 #exit
 
 TEMP_DIR=/cup/d6/datang/TEMP
 echo "get data begin!------------------------------------------"
 
 for tableName in ${HIVE_TABLES}
  do
  
  file_name=`echo ${tableName}|sed 's/_//g'`
  
  rm -rf ${TEMP_DIR}/${tableName}/$ANALY_DATE/$ANALY_HOUR/*
  mkdir -p ${TEMP_DIR}/${tableName}/$ANALY_DATE/$ANALY_HOUR
  
  hdfs dfs -get ${OUTPUT}/${file_name}* ${TEMP_DIR}/${tableName}/$ANALY_DATE/$ANALY_HOUR/
  echo "hdfs dfs -get ${OUTPUT}/${file_name}* ${TEMP_DIR}/${tableName}/$ANALY_DATE/$ANALY_HOUR/"
 done
 
 orgn=volteorgn
 rm -rf $TEMP_DIR/volte_orgn/$ANALY_DATE/$ANALY_HOUR/*
 mkdir -p ${TEMP_DIR}/volte_orgn/$ANALY_DATE/$ANALY_HOUR
 hdfs dfs -get ${OUTPUT}/${orgn}* ${TEMP_DIR}/volte_orgn/$ANALY_DATE/$ANALY_HOUR
 rm -rf $TEMP_DIR/share_volte_orgn/$ANALY_DATE/$ANALY_HOUR/*
 mkdir -p ${TEMP_DIR}/share_volte_orgn/$ANALY_DATE/$ANALY_HOUR
 hdfs dfs -get ${OUTPUT}/sharevolteorgn* ${TEMP_DIR}/share_volte_orgn/$ANALY_DATE/$ANALY_HOUR
 
 rm -rf $TEMP_DIR/share_s1mme/$ANALY_DATE/$ANALY_HOUR/*
 mkdir -p ${TEMP_DIR}/share_s1mme/$ANALY_DATE/$ANALY_HOUR
 hdfs dfs -get ${OUTPUT}/shares1mme* ${TEMP_DIR}/share_s1mme/$ANALY_DATE/$ANALY_HOUR

 rm -rf $TEMP_DIR/share_volterx/$ANALY_DATE/$ANALY_HOUR/*
 mkdir -p ${TEMP_DIR}/share_volterx/$ANALY_DATE/$ANALY_HOUR
 hdfs dfs -get ${OUTPUT}/sharevolterx* ${TEMP_DIR}/share_volterx/$ANALY_DATE/$ANALY_HOUR

 DEL_HOUR="`date -d ' -6 hour' +%H`"
 if [ $ANALY_HOUR = 00 ]
 then
    ANALY_DATE="`date -d ' -1 day' +%Y%m%d`"
 elif [ $ANALY_HOUR = 01 ]; then
    ANALY_DATE="`date -d ' -1 day' +%Y%m%d`"
 elif [ $ANALY_HOUR = 02 ]; then
    ANALY_DATE="`date -d ' -1 day' +%Y%m%d`"
 elif [ $ANALY_HOUR = 03 ]; then
    ANALY_DATE="`date -d ' -1 day' +%Y%m%d`"
 elif [ $ANALY_HOUR = 04 ]; then
    ANALY_DATE="`date -d ' -1 day' +%Y%m%d`"
 elif [ $ANALY_HOUR = 23 ]; then
    ANALY_DATE="`date -d ' -1 day' +%Y%m%d`"
 else
    ANALY_DATE=$1
 fi
 hdfs dfs -rm -R -skipTrash /datang/FILTER/${ANALY_DATE}${DEL_HOUR}
 echo "get data end!------------------------------------------"
 echo "`date '+/%y/%m/%d %H:%M:%S'` INFO done ,exit..."
 exit
 

hdfs2local.sh脚本76~87行：

这是hadoop命令调用jar包的命令，格式为：

jar

运行jar文件。用户可以把他们的Map Reduce代码捆绑到jar文件中，使用这个命令执行。

用法：hadoop jar <jar> [mainClass] args...

http://hadoop.apache.org/docs/r1.0.4/cn/commands_manual.html

问题来了，取到的参数是哪些？

${HADOOP} jar ${JAR_PACKAGK} ${JAR_MAIN}  ${GX}  ${S1}  ${SV}  ${SGS}  ${MW}  ${HTTP}  ${SESSION}  ${SLICE}  ${OUTPUT}  ${T_PROCESS}  ${ANALY_HOUR}

按理说${HADOOP} jar后面的都是参数，但是数量与class DataJob76~92行的参数数量对不上，问题出在哪？分析GenericOptionsParser类有蹊跷：

34     String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

 找到官方文档：http://hadoop.apache.org/docs/r1.0.4/api/org/apache/hadoop/util/GenericOptionsParser.html

String[]

getRemainingArgs()            Returns an array of Strings containing only application-specific arguments.

还需要说明的是，hadoop的参数分两类：通用参数和命令参数

bin/hadoop command [genericOptions] [commandOptions]

通用参数是：

常规选项

下面的选项被 dfsadmin, fs, fsck和 job支持。 应用程序要实现 Tool来支持 常规选项。

GENERIC_OPTION	描述
-conf <configuration file>	指定应用程序的配置文件。
-D <property=value>	为指定property指定值value。
-fs <local|namenode:port>	指定namenode。
-jt <local|jobtracker:port>	指定job tracker。只适用于job。
-files <逗号分隔的文件列表>	指定要拷贝到map reduce集群的文件的逗号分隔的列表。 只适用于job。
-libjars <逗号分隔的jar列表>	指定要包含到classpath中的jar文件的逗号分隔的列表。 只适用于job。
-archives <逗号分隔的archive列表>	指定要被解压到计算节点上的档案文件的逗号分割的列表。 只适用于job。

除了generic options 通用参数以外的，都是命令的参数（或者叫application-specific arguments命令专属参数）

所以执行代码String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs()后，otherArgs 中剔除了${JAR_PACKAGK} ${JAR_MAIN}，只保留剩余的参数：

${GX}  ${S1}  ${SV}  ${SGS}  ${MW}  ${HTTP}  ${SESSION}  ${SLICE}  ${OUTPUT}  ${T_PROCESS}  ${ANALY_HOUR}

这就与class DataJob76~92行对应上了。


相关参考材料：

Hadoop 常见指令

https://blog.csdn.net/u011414200/article/details/50465433

MapReduce处理输出多文件格式(MultipleOutputs)

https://blog.csdn.net/shujuboke/article/details/77199840

Hadoop系列之ToolRunner与GenericOptionsParser用法

https://blog.csdn.net/u011734144/article/details/60769745

Class GenericOptionsParser

http://hadoop.apache.org/docs/r1.0.4/api/org/apache/hadoop/util/GenericOptionsParser.html

Hadoop 0.18 命令手册

http://hadoop.apache.org/docs/r1.0.4/cn/commands_manual.html