@Test
    public void shouldPersistRDDBasedOnStorageLevel() throws Exception {
        Spark.create("local[4]");
        int counter = 0;
        for (final String storageLevel : Arrays.asList("MEMORY_ONLY", "DISK_ONLY", "MEMORY_ONLY_SER", "MEMORY_AND_DISK_SER")) {
            assertEquals(counter, Spark.getRDDs().size());
            assertEquals(counter, Spark.getContext().getPersistentRDDs().size());
            counter++;
            final String rddName = TestHelper.makeTestDataDirectory(PersistedInputOutputRDDIntegrateTest.class, UUID.randomUUID().toString());
            final Configuration configuration = super.getBaseConfiguration();
            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_SPARK_PERSIST_STORAGE_LEVEL, storageLevel);
            configuration.setProperty(Constants.GREMLIN_HADOOP_OUTPUT_LOCATION, rddName);
            configuration.setProperty(Constants.GREMLIN_SPARK_PERSIST_CONTEXT, true);
            Graph graph = GraphFactory.open(configuration);
            graph.compute(SparkGraphComputer.class)
                    .result(GraphComputer.ResultGraph.NEW)
                    .persist(GraphComputer.Persist.EDGES)
                    .program(TraversalVertexProgram.build()
                            .traversal(graph.traversal().withComputer(SparkGraphComputer.class),
                                    "gremlin-groovy",
                                    "g.V().groupCount('m').by('name').out()").create(graph)).submit().get();
            ////////
            assertTrue(Spark.hasRDD(Constants.getGraphLocation(rddName)));
            assertEquals(StorageLevel.fromString(storageLevel), Spark.getRDD(Constants.getGraphLocation(rddName)).getStorageLevel());
            assertEquals(counter, Spark.getRDDs().size());
            assertEquals(counter, 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