@Test
    public void testSparkRDDPersistence() throws Exception {
        final String prefix = TestHelper.makeTestDataFile(SparkTest.class, "testSparkRDDPersistence", "graphRDD-");
        final Configuration configuration = new BaseConfiguration();
        configuration.setProperty("spark.master", "local[4]");
        Spark.create(configuration);

        configuration.setProperty("spark.serializer", GryoSerializer.class.getCanonicalName());
        configuration.setProperty(Graph.GRAPH, HadoopGraph.class.getName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, TestFiles.PATHS.get("tinkerpop-modern-v3.kryo"));
        configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_JARS_IN_DISTRIBUTED_CACHE, false);
        configuration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);

        for (int i = 0; i < 10; i++) {
            final String graphRDDName = Constants.getGraphLocation(prefix + i);
            assertEquals(i, Spark.getRDDs().size());
            configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, prefix + i);
            Graph graph = GraphFactory.open(configuration);
            graph.compute(SparkGraphComputer.class)
                .persist(GraphComputer.Persist.VERTEX_PROPERTIES)
                .vertexProperties(__.properties("dummy"))
                .program(PageRankVertexProgram.build().iterations(1).create(graph)).submit().get();
            assertNotNull(Spark.getRDD(graphRDDName));
            assertEquals(i + 1, Spark.getRDDs().size());
        }

        for (int i = 9; i >= 0; i--) {
            final String graphRDDName = Constants.getGraphLocation(prefix + i);
            assertEquals(i + 1, getPersistedRDDSize());
            assertEquals(i + 1, Spark.getRDDs().size());
            assertTrue(hasPersistedRDD(graphRDDName));
            Spark.removeRDD(graphRDDName);
            assertFalse(hasPersistedRDD(graphRDDName));
        }

        assertEquals(0, getPersistedRDDSize());
        assertEquals(0, Spark.getRDDs().size());
        Spark.close();
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/SparkTest.java