@Override
    public RecordReader<NullWritable, VertexWritable> createRecordReader(final InputSplit inputSplit, final TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        try {
            final org.apache.hadoop.conf.Configuration hadoopConfiguration = taskAttemptContext.getConfiguration();
            final SparkConf sparkConfiguration = new SparkConf();
            sparkConfiguration.setAppName(UUID.randomUUID().toString());
            hadoopConfiguration.forEach(entry -> sparkConfiguration.set(entry.getKey(), entry.getValue()));
            final InputRDD inputRDD = (InputRDD) Class.forName(sparkConfiguration.get(Constants.GREMLIN_HADOOP_GRAPH_READER)).newInstance();
            final JavaSparkContext javaSparkContext = new JavaSparkContext(Spark.create(sparkConfiguration));
            final Iterator<Tuple2<Object, VertexWritable>> iterator = inputRDD.readGraphRDD(ConfUtil.makeApacheConfiguration(taskAttemptContext.getConfiguration()), javaSparkContext).toLocalIterator();
            return new RecordReader<NullWritable, VertexWritable>() {
                @Override
                public void initialize(final InputSplit inputSplit, final TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {

                }

                @Override
                public boolean nextKeyValue() throws IOException, InterruptedException {
                    return iterator.hasNext();
                }

                @Override
                public NullWritable getCurrentKey() throws IOException, InterruptedException {
                    return NullWritable.get();
                }

                @Override
                public VertexWritable getCurrentValue() throws IOException, InterruptedException {
                    return iterator.next()._2();
                }

                @Override
                public float getProgress() throws IOException, InterruptedException {
                    return 1.0f; // TODO: make this dynamic (how? its an iterator.)
                }

                @Override
                public void close() throws IOException {

                }
            };
        } catch (final ClassNotFoundException | InstantiationException | IllegalAccessException e) {
            throw new IOException(e.getMessage(), e);
        }

    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/structure/io/InputRDDFormat.java