@Override
    public void initialize(final InputSplit genericSplit, final TaskAttemptContext context) throws IOException {
        final FileSplit split = (FileSplit) genericSplit;
        final Configuration configuration = context.getConfiguration();
        if (configuration.get(Constants.GREMLIN_HADOOP_GRAPH_FILTER, null) != null)
            this.graphFilter = VertexProgramHelper.deserialize(ConfUtil.makeApacheConfiguration(configuration), Constants.GREMLIN_HADOOP_GRAPH_FILTER);
        this.gryoReader = GryoReader.build().mapper(
                GryoMapper.build().addRegistries(IoRegistryHelper.createRegistries(ConfUtil.makeApacheConfiguration(configuration))).create()).create();
        long start = split.getStart();
        final Path file = split.getPath();
        if (null != new CompressionCodecFactory(configuration).getCodec(file)) {
            throw new IllegalStateException("Compression is not supported for the (binary) Gryo format");
        }
        // open the file and seek to the start of the split
        this.inputStream = file.getFileSystem(configuration).open(split.getPath());
        this.splitLength = split.getLength();
        if (this.splitLength > 0) this.splitLength -= (seekToHeader(this.inputStream, start) - start);
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
hadoop-gremlin/src/main/java/org/apache/tinkerpop/gremlin/hadoop/structure/io/gryo/GryoRecordReader.java