ApacheHttpdLogfileInputFormat extends
        FileInputFormat<LongWritable, ParsedRecord> { public ApacheHttpdLogfileRecordReader createRecordReader() { try { return new ApacheHttpdLogfileRecordReader(getLogFormat(), getRequestedFields(), getTypeRemappings(), getAdditionalDissectors()); } catch (IOException e) { return null; } } ApacheHttpdLogfileInputFormat();  ApacheHttpdLogfileInputFormat(
            String logformat,
            Collection<String> requestedFields,
            Map<String, Set<String>> typeRemappings,
            List<Dissector> additionalDissectors); List<String> listPossibleFields(String logformat); static List<String> listPossibleFields(String logformat, Map<String, Set<String>> typeRemappings, List<Dissector> additionalDissectors); String getLogFormat(); Set<String> getRequestedFields(); Map<String, Set<String>> getTypeRemappings(); List<Dissector> getAdditionalDissectors(); ApacheHttpdLogfileRecordReader createRecordReader(); ApacheHttpdLogfileRecordReader getRecordReader(); @Override RecordReader<LongWritable, ParsedRecord> createRecordReader(
            final InputSplit split, final TaskAttemptContext context); void setTypeRemappings(Map<String, Set<String>> newTypeRemappings);  }
@Test public void checkInputFormat() throws IOException, InterruptedException { Configuration conf = new Configuration(false); conf.set("fs.default.name", "file: conf.set("nl.basjes.parse.apachehttpdlogline.format", logformat); conf.set("nl.basjes.parse.apachehttpdlogline.fields", "TIME.EPOCH:request.receive.time.epoch," + "HTTP.USERAGENT:request.user-agent"); File testFile = new File("src/test/resources/access.log"); Path path = new Path(testFile.getAbsoluteFile().toURI()); FileSplit split = new FileSplit(path, 0, testFile.length(), null); InputFormat inputFormat = ReflectionUtils.newInstance(ApacheHttpdLogfileInputFormat.class, conf); TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID()); RecordReader reader = inputFormat.createRecordReader(split, context); reader.initialize(split, context); assertTrue(reader.nextKeyValue()); Object value = reader.getCurrentValue(); if (value instanceof ParsedRecord) { assertEquals("1483272081000", ((ParsedRecord) value).getString("TIME.EPOCH:request.receive.time.epoch")); assertEquals("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.71 Safari/537.36", ((ParsedRecord) value).getString("HTTP.USERAGENT:request.user-agent")); } else { fail("Wrong return class type"); } }
@Test public void checkAllOutputTypes() throws IOException, InterruptedException { Configuration conf = new Configuration(false); conf.set("fs.default.name", "file: List<String> fields = Arrays.asList( "ANY:any", "ANY:any", "ANY:any", "STRING:string", "STRING:string", "STRING:string", "INT:int", "INT:int", "INT:int", "LONG:long", "LONG:long", "LONG:long", "FLOAT:float", "FLOAT:float", "FLOAT:float", "DOUBLE:double", "DOUBLE:double", "DOUBLE:double"); File testFile = new File("src/test/resources/access.log"); Path path = new Path(testFile.getAbsoluteFile().toURI()); FileSplit split = new FileSplit(path, 0, testFile.length(), null); Map<String, Set<String>> typeRemappings = new HashMap<>(); List<Dissector> dissectors = new ArrayList<>(); dissectors.add(new NormalValuesDissector(HttpdLogFormatDissector.INPUT_TYPE)); InputFormat inputFormat = new ApacheHttpdLogfileInputFormat( logformat, fields, typeRemappings, dissectors); TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID()); RecordReader reader = inputFormat.createRecordReader(split, context); reader.initialize(split, context); assertTrue(reader.nextKeyValue()); Object value = reader.getCurrentValue(); if (value instanceof ParsedRecord) { ParsedRecord record = (ParsedRecord)value; assertEquals("42", record.getString("ANY:any")); assertEquals(42L, record.getLong("ANY:any").longValue()); assertEquals(42D, record.getDouble("ANY:any"), 0.1D); assertEquals("FortyTwo", record.getString("STRING:string")); assertEquals(null, record.getLong("STRING:string")); assertEquals(null, record.getDouble("STRING:string")); assertEquals("42", record.getString("INT:int")); assertEquals(42L, record.getLong("INT:int").longValue()); assertEquals(null, record.getDouble("INT:int")); assertEquals("42", record.getString("LONG:long")); assertEquals(42L, record.getLong("LONG:long").longValue()); assertEquals(null, record.getDouble("LONG:long")); assertEquals("42.0", record.getString("FLOAT:float")); assertEquals(null, record.getLong("FLOAT:float")); assertEquals(42D, record.getDouble("FLOAT:float"), 0.1D); assertEquals("42.0", record.getString("DOUBLE:double")); assertEquals(null, record.getLong("DOUBLE:double")); assertEquals(42D, record.getDouble("DOUBLE:double"), 0.1D); } else { fail("Wrong return class type"); } }