工作采坑札記：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