基于Lucene和HDFS的PB级数据索引、搜索、存储系统.zip

package com.hansight.hadoopsearcher.util; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.mapreduce.*; import org.apache.hadoop.util.Progressable; import java.lang.reflect.Constructor; import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; /** * Created by zhachao on 15-9-15. */ public class HadoopCompat { private static final boolean useV21; private static final Constructor<?> JOB_CONTEXT_CONSTRUCTOR; private static final Constructor<?> TASK_CONTEXT_CONSTRUCTOR; private static final Constructor<?> MAP_CONTEXT_CONSTRUCTOR; private static final Constructor<?> GENERIC_COUNTER_CONSTRUCTOR; private static final Field READER_FIELD; private static final Field WRITER_FIELD; private static final Method GET_CONFIGURATION_METHOD; private static final Method SET_STATUS_METHOD; private static final Method GET_COUNTER_METHOD; private static final Method GET_COUNTER_ENUM_METHOD; private static final Method INCREMENT_COUNTER_METHOD; private static final Method GET_COUNTER_VALUE_METHOD; private static final Method GET_TASK_ATTEMPT_ID; private static final Method GET_JOB_ID_METHOD; private static final Method GET_JOB_NAME_METHOD; private static final Method GET_INPUT_SPLIT_METHOD; private static final Method GET_DEFAULT_BLOCK_SIZE_METHOD; private static final Method GET_DEFAULT_REPLICATION_METHOD; //TODO : private static final Method H2_IS_FILE_CLOSED_METHOD; static { boolean v21 = true; final String PACKAGE = "org.apache.hadoop.mapreduce"; try { Class.forName(PACKAGE + ".task.JobContextImpl"); } catch (ClassNotFoundException cnfe) { v21 = false; } useV21 = v21; Class<?> jobContextCls; Class<?> taskContextCls; Class<?> taskIOContextCls; Class<?> mapContextCls; Class<?> genericCounterCls; try { if (v21) { jobContextCls = Class.forName(PACKAGE+".task.JobContextImpl"); taskContextCls = Class.forName(PACKAGE+".task.TaskAttemptContextImpl"); taskIOContextCls = Class.forName(PACKAGE+".task.TaskInputOutputContextImpl"); mapContextCls = Class.forName(PACKAGE + ".task.MapContextImpl"); genericCounterCls = Class.forName(PACKAGE+".counters.GenericCounter"); } else { jobContextCls = Class.forName(PACKAGE+".JobContext"); taskContextCls = Class.forName(PACKAGE+".TaskAttemptContext"); taskIOContextCls = Class.forName(PACKAGE+".TaskInputOutputContext"); mapContextCls = Class.forName(PACKAGE + ".MapContext"); genericCounterCls = Class.forName("org.apache.hadoop.mapred.Counters$Counter"); } } catch (ClassNotFoundException e) { throw new IllegalArgumentException("Can't find class", e); } try { JOB_CONTEXT_CONSTRUCTOR = jobContextCls.getConstructor(Configuration.class, JobID.class); JOB_CONTEXT_CONSTRUCTOR.setAccessible(true); TASK_CONTEXT_CONSTRUCTOR = taskContextCls.getConstructor(Configuration.class, TaskAttemptID.class); TASK_CONTEXT_CONSTRUCTOR.setAccessible(true); GENERIC_COUNTER_CONSTRUCTOR = genericCounterCls.getDeclaredConstructor(String.class, String.class, Long.TYPE); GENERIC_COUNTER_CONSTRUCTOR.setAccessible(true); if (useV21) { MAP_CONTEXT_CONSTRUCTOR = mapContextCls.getDeclaredConstructor(Configuration.class, TaskAttemptID.class, RecordReader.class, RecordWriter.class, OutputCommitter.class, StatusReporter.class, InputSplit.class); Method get_counter; try { get_counter = TaskAttemptContext.class.getMethod("getCounter", String.class, String.class); } catch (Exception e) { get_counter = TaskInputOutputContext.class.getMethod("getCounter", String.class, String.class); } GET_COUNTER_METHOD = get_counter; GET_COUNTER_ENUM_METHOD = TaskAttemptContext.class.getMethod("getCounter", Enum.class); GET_DEFAULT_BLOCK_SIZE_METHOD = FileSystem.class.getMethod("getDefaultBlockSize", Path.class); GET_DEFAULT_REPLICATION_METHOD = FileSystem.class.getMethod("getDefaultReplication", Path.class); } else { MAP_CONTEXT_CONSTRUCTOR = mapContextCls.getConstructor(Configuration.class, TaskAttemptID.class, RecordReader.class, RecordWriter.class, OutputCommitter.class, StatusReporter.class, InputSplit.class); GET_COUNTER_METHOD = TaskInputOutputContext.class.getMethod("getCounter", String.class, String.class); GET_COUNTER_ENUM_METHOD = TaskInputOutputContext.class.getMethod("getCounter", Enum.class); GET_DEFAULT_BLOCK_SIZE_METHOD = FileSystem.class.getMethod("getDefaultBlockSize"); GET_DEFAULT_REPLICATION_METHOD = FileSystem.class.getMethod("getDefaultReplication"); } MAP_CONTEXT_CONSTRUCTOR.setAccessible(true); READER_FIELD = mapContextCls.getDeclaredField("reader"); READER_FIELD.setAccessible(true); WRITER_FIELD = taskIOContextCls.getDeclaredField("output"); WRITER_FIELD.setAccessible(true); GET_CONFIGURATION_METHOD = JobContext.class .getMethod("getConfiguration"); SET_STATUS_METHOD = TaskAttemptContext.class.getMethod("setStatus", String.class); GET_TASK_ATTEMPT_ID = TaskAttemptContext.class.getMethod("getTaskAttemptID"); INCREMENT_COUNTER_METHOD = Counter.class .getMethod("increment", Long.TYPE); GET_COUNTER_VALUE_METHOD = Counter.class .getMethod("getValue"); GET_JOB_ID_METHOD = JobContext.class .getMethod("getJobID"); GET_JOB_NAME_METHOD = JobContext.class .getMethod("getJobName"); GET_INPUT_SPLIT_METHOD = MapContext.class .getMethod("getInputSplit"); } catch (SecurityException e) { throw new IllegalArgumentException("Can't run constructor ", e); } catch (NoSuchMethodException e) { throw new IllegalArgumentException("Can't find constructor ", e); } catch (NoSuchFieldException e) { throw new IllegalArgumentException("Can't find field ", e); } } /** * True if runtime Hadoop version is 2.x, false otherwise. */ public static boolean isVersion2x() { return useV21; } private static Object newInstance(Constructor<?> constructor, Object...args) { try { return constructor.newInstance(args); } catch (InstantiationException | InvocationTargetException | IllegalAccessException e) { throw new IllegalArgumentException("Can't instantiate " + constructor, e); } } /** * C