public static <M> JavaPairRDD<Object, VertexWritable> prepareFinalGraphRDD(
            final JavaPairRDD<Object, VertexWritable> graphRDD,
            final JavaPairRDD<Object, ViewIncomingPayload<M>> viewIncomingRDD,
            final Set<VertexComputeKey> vertexComputeKeys) {
        // the graphRDD and the viewRDD must have the same partitioner
        if (graphRDD.partitioner().isPresent())
            assert (graphRDD.partitioner().get().equals(viewIncomingRDD.partitioner().get()));
        final String[] vertexComputeKeysArray = VertexProgramHelper.vertexComputeKeysAsArray(vertexComputeKeys); // the compute keys as an array
        return graphRDD.leftOuterJoin(viewIncomingRDD)
                .mapValues(tuple -> {
                    final StarGraph.StarVertex vertex = tuple._1().get();
                    vertex.dropVertexProperties(vertexComputeKeysArray); // drop all existing compute keys
                    // attach the final computed view to the cached graph
                    final List<DetachedVertexProperty<Object>> view = tuple._2().isPresent() ? tuple._2().get().getView() : Collections.emptyList();
                    for (final DetachedVertexProperty<Object> property : view) {
                        if (!VertexProgramHelper.isTransientVertexComputeKey(property.key(), vertexComputeKeys))
                            property.attach(Attachable.Method.create(vertex));
                    }
                    return tuple._1();
                });
    }
@author Marko A. Rodriguez (http://markorodriguez.com)
spark-gremlin/src/main/java/org/apache/tinkerpop/gremlin/spark/process/computer/SparkExecutor.java