工作采坑札记:4. Hadoop获取InputSplit文件信息

1. 场景

  基于客户的数据处理需求,客户分发诸多小数据文件,文件每行代表一条记录信息,且每个文件以"类型_yyyyMMdd_批次号"命名。由于同一条记录可能存在于多个文件中,且处于多个文件中的相同记录最终只有时间最新的记录有效,但文件的每行记录并未提供时间信息,因此需要从每个文件名中提取时间信息作为文件每行记录信息。

  因此,考虑到小文件数量较多,且数据总量近千万级别,因此借助Hadoop工具,在MapReduce中获取处理该条记录所对应的拆分后的文件名信息。

2. 技术实现

  当Hadoop处理简单文本输入时,如job.setInputFormatClass(TextInputFormat.class);,mapper运行时,可以使用如下方法获取对应的filesplit,进而获取到文件路径信息、文件名信息等:

// 0.19 hadoop
(FileSplit) (reporter.getInputSplit());

// 0.20 hadoop
(FileSplit) (context.getInputSplit());

  但如果使用多输入文件时,如:MultipleInputs.addInputPath(job, new Path(path), SequenceFileInputFormat.class, ProfileMapper.class);,会出现如下异常信息:

java.lang.ClassCastException: org.apache.hadoop.mapreduce.lib.input.TaggedInputSplit cannot be cast to org.apache.hadoop.mapreduce.lib.input.FileSplit

  而实际需要的FileSplit是TaggedInputSplit中的成员变量inputSplit,但是TaggedInputSplit类在社区版的Hadoop中并非public,所以无法直接获取对应信息。  

  可以采用反射来获取TaggedInputSplit中的inputSplit,具体实现方法如下:

String getFileName(){
            InputSplit inputSplit = context.getInputSplit();
            Class<? extends InputSplit> splitClass = inputSplit.getClass();
            FileSplit fileSplit = null;
            if(splitClass.equals(FileSplit.class)){
                fileSplit = (FileSplit) inputSplit;
            }else if(splitClass.getName().equals("org.apache.hadoop.mapreduce.lib.input.TaggedInputSplit")){
                try {
                    Method getInputSplitMethod = splitClass.getDeclaredMethod("getInputSplit");
                    getInputSplitMethod.setAccessible(true);
                    fileSplit = (FileSplit) getInputSplitMethod.invoke(inputSplit);
                } catch (Exception e) {
                }
            }
            return fileSplit.getPath().getName();
        }

参考:

(1) https://blog.csdn.net/rabbitxl/article/details/8645428 

(2) https://stackoverflow.com/questions/11130145/hadoop-multipleinputs-fails-with-classcastexception

原文地址:https://www.cnblogs.com/mengrennwpu/p/9704082.html