@Test
    public void testComplexChain() throws Exception {
        Spark.create("local[4]");

        final String rddName = TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class, "testComplexChain", "graphRDD");
        final String rddName2 = TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class, "testComplexChain", "graphRDD2");
        final Configuration configuration = super.getBaseConfiguration();
        configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, GryoInputFormat.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, TestFiles.PATHS.get("tinkerpop-modern-v3.kryo"));
        configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, rddName);
        configuration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);

        assertFalse(Spark.hasRDD(Constants.getGraphLocation(rddName)));
        assertEquals(0, Spark.getContext().getPersistentRDDs().size());
        Graph graph = GraphFactory.open(configuration);
        graph = graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.EDGES).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
        GraphTraversalSource g = graph.traversal();
        assertEquals(6l, g.V().count().next().longValue());
        assertEquals(6l, g.E().count().next().longValue());
        assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
        ////
        assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName)));
        assertEquals(1, Spark.getContext().getPersistentRDDs().size());
        ////
        configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_READER, PersistedInputRDD.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_INPUT_LOCATION, rddName);
        configuration.setProperty(Constants.GREMLIN_HADOOP_GRAPH_WRITER, PersistedOutputRDD.class.getCanonicalName());
        configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, rddName2);
        ////
        graph = GraphFactory.open(configuration);
        graph = graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.EDGES).mapReduce(PageRankMapReduce.build().create()).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
        g = graph.traversal();
        assertEquals(6l, g.V().count().next().longValue());
        assertEquals(6l, g.E().count().next().longValue());
        assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
        ////
        assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName)));
        assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName2)));
        assertTrue(Spark.hasRDD(Constants.getMemoryLocation(rddName2, PageRankMapReduce.DEFAULT_MEMORY_KEY)));
        assertEquals(3, Spark.getContext().getPersistentRDDs().size());
        ////
        graph = GraphFactory.open(configuration);
        graph = graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.VERTEX_PROPERTIES).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
        g = graph.traversal();
        assertEquals(6l, g.V().count().next().longValue());
        assertEquals(0l, g.E().count().next().longValue());
        assertEquals(6l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
        ////
        assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName)));
        assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName2)));
        assertFalse(Spark.hasRDD(Constants.getMemoryLocation(rddName2, PageRankMapReduce.DEFAULT_MEMORY_KEY)));
        assertEquals(2, Spark.getContext().getPersistentRDDs().size());
        ////
        graph = GraphFactory.open(configuration);
        graph = graph.compute(SparkGraphComputer.class).persist(GraphComputer.Persist.NOTHING).program(PageRankVertexProgram.build().iterations(2).create(graph)).submit().get().graph();
        assertFalse(Spark.hasRDD(Constants.getGraphLocation(rddName2)));
        g = graph.traversal();
        assertEquals(0l, g.V().count().next().longValue());
        assertEquals(0l, g.E().count().next().longValue());
        assertEquals(0l, g.V().values(PageRankVertexProgram.PAGE_RANK).count().next().longValue());
        ////
        assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName)));
        assertFalse(Spark.hasRDD(Constants.getGraphLocation(rddName2)));
        assertFalse(Spark.hasRDD(Constants.getMemoryLocation(rddName2, PageRankMapReduce.DEFAULT_MEMORY_KEY)));
        assertEquals(1, Spark.getContext().getPersistentRDDs().size());
        Spark.close();
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/test/java/org/apache/tinkerpop/gremlin/spark/structure/io/PersistedInputOutputRDDIntegrateTest.java