class JsonRecordReader extends AbstractEventRecordReader<BytesWritable> {

    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) throws IOException {
        
        super.initialize(split, context);
        
        if (!(split instanceof FileSplit)) {
            throw new IOException("Cannot handle split type " + split.getClass().getName());
        }
        
        FileSplit fsplit = (FileSplit) split;
        Path file = fsplit.getPath();
        rawFileName = file.getName();
        fileURI = file.toUri();
        FileSystem fs = file.getFileSystem(context.getConfiguration());
        InputStream is = fs.open(file);
        start = fsplit.getStart();
        end = start + fsplit.getLength();
        pos = start;
        
        String normURI = fileURI.getScheme() + "://" + fileURI.getPath();
        
        setupReader(is);
        
        if (logger.isInfoEnabled()) {
            logger.info("Reading Json records from " + normURI + " via " + is.getClass().getName());
        }
        
        jsonHelper = (JsonDataTypeHelper) createHelper(context.getConfiguration());
        this.parseHeaderOnly = !jsonHelper.processExtraFields();
        jsonFlattener = jsonHelper.newFlattener();
        
        if (logger.isInfoEnabled()) {
            logger.info("Json flattener mode: " + jsonFlattener.getFlattenMode().name());
        }
    }

}