<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.tinkerpop.gremlin.hadoop.process.computer.spark;

import com.google.common.base.Optional;
import org.apache.commons.configuration.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritableIterator;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import scala.Tuple2;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.stream.Stream;

/**
 * @author <NAME> (http://markorodriguez.com)
 */
public final class SparkExecutor {

    private SparkExecutor() {
    }

    public static <M> JavaPairRDD<Object, SparkVertexPayload<M>> executeVertexProgramIteration(final JavaPairRDD<Object, SparkVertexPayload<M>> graphRDD, final SparkMemory memory, final Configuration apacheConfiguration) {
        // execute vertex program iteration
        final JavaPairRDD<Object, SparkVertexPayload<M>> verticesHoldingOutgoingMessages = graphRDD.mapPartitionsToPair(partitionIterator -> {     // each partition(Spark)/worker(TP3) has a local copy of the vertex program to reduce object creation
            final VertexProgram<M> workerVertexProgram = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration);
            final Set<String> elementComputeKeys = workerVertexProgram.getElementComputeKeys();
            workerVertexProgram.workerIterationStart(memory);
            return () -> IteratorUtils.map(partitionIterator, vertex -> {
                vertex._2().getOutgoingMessages().clear(); // there should be no outgoing messages at this point
                workerVertexProgram.execute(ComputerGraph.of(vertex._2().getVertex(), elementComputeKeys), vertex._2(), memory);
                if (!partitionIterator.hasNext())
                    workerVertexProgram.workerIterationEnd(memory);
                vertex._2().getMessages().clear(); // there should be no incoming messages at this point (only outgoing messages)
                return vertex;
            });
        });

        // emit messages by appending them to the graph as message payloads
        final JavaPairRDD<Object, SparkPayload<M>> verticesAndOutgoingMessages = verticesHoldingOutgoingMessages.flatMapToPair(vertex -> () ->
                IteratorUtils.<Tuple2<Object, SparkPayload<M>>>concat(
                        (Iterator) IteratorUtils.of(vertex),
                        (Iterator) IteratorUtils.map(vertex._2().detachOutgoingMessages(), // this removes all outgoing messages once they have been iterated by this step
                                message -> new Tuple2<>(message._1(), new SparkMessagePayload<>(message._2())))));

        // "message pass" by merging the message payloads with the vertex payloads
        final MessageCombiner<M> messageCombiner = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration).getMessageCombiner().orElse(null);
        final JavaPairRDD<Object, SparkVertexPayload<M>> verticesHoldingIncomingMessages = (JavaPairRDD) verticesAndOutgoingMessages
                .reduceByKey((payloadA, payloadB) -> {
                    if (payloadA.isVertex()) {
                        if (payloadB.isVertex())
                            throw new IllegalStateException("It should not be the case that two vertices reduce to the same key: " + payloadA.asVertexPayload().getVertex() + "==" + payloadB.asVertexPayload().getVertex());
                        else {
                            payloadA.addMessages(payloadB.getMessages(), messageCombiner);
                            return payloadA;
                        }
                    } else {
                        payloadB.addMessages(payloadA.getMessages(), messageCombiner);
                        return payloadB;
                    }
                })
                .filter(payload -> payload._2().isVertex());  // just in case there are messages sent to vertices that do not exist

        verticesHoldingIncomingMessages.foreachPartition(partitionIterator -> {
        }); // need to complete a task so its BSP.
        return verticesHoldingIncomingMessages;
    }
<commit_msg>a variation on the message passing algorithm where the message reduction is done separate from the graph data and then rejoined with the graph later.<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.tinkerpop.gremlin.hadoop.process.computer.spark;

import com.google.common.base.Optional;
import org.apache.commons.configuration.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.tinkerpop.gremlin.hadoop.Constants;
import org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritable;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.ObjectWritableIterator;
import org.apache.tinkerpop.gremlin.hadoop.structure.io.VertexWritable;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
import org.apache.tinkerpop.gremlin.process.computer.Memory;
import org.apache.tinkerpop.gremlin.process.computer.MessageCombiner;
import org.apache.tinkerpop.gremlin.process.computer.VertexProgram;
import org.apache.tinkerpop.gremlin.process.computer.util.ComputerGraph;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import org.apache.tinkerpop.gremlin.structure.VertexProperty;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedFactory;
import org.apache.tinkerpop.gremlin.structure.util.detached.DetachedVertexProperty;
import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
import scala.Tuple2;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Stream;

/**
 * @author <NAME> (http://markorodriguez.com)
 */
public final class SparkExecutor {

    private SparkExecutor() {
    }

    public static <M> JavaPairRDD<Object, SparkVertexPayload<M>> executeVertexProgramIteration(final JavaPairRDD<Object, SparkVertexPayload<M>> graphRDD, final SparkMemory memory, final Configuration apacheConfiguration) {
        // execute vertex program iteration
        final JavaPairRDD<Object, SparkVertexPayload<M>> verticesWithOutgoingMessages = graphRDD
                .mapPartitionsToPair(partitionIterator -> {     // each partition(Spark)/worker(TP3) has a local copy of the vertex program to reduce object creation
                    final VertexProgram<M> workerVertexProgram = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration);
                    final Set<String> elementComputeKeys = workerVertexProgram.getElementComputeKeys();
                    workerVertexProgram.workerIterationStart(memory);
                    return () -> IteratorUtils.map(partitionIterator, vertex -> {
                        vertex._2().getOutgoingMessages().clear(); // there should be no outgoing messages at this point
                        workerVertexProgram.execute(ComputerGraph.of(vertex._2().getVertex(), elementComputeKeys), vertex._2(), memory);
                        if (!partitionIterator.hasNext())
                            workerVertexProgram.workerIterationEnd(memory);
                        vertex._2().getMessages().clear(); // there should be no incoming messages at this point (only outgoing messages)
                        return vertex;
                    });
                }).cache();

        // "message pass" by reducing on the vertex object id of the message payloads
        final MessageCombiner<M> messageCombiner = VertexProgram.<VertexProgram<M>>createVertexProgram(apacheConfiguration).getMessageCombiner().orElse(null);
        final JavaPairRDD<Object, SparkMessagePayload<M>> incomingMessages = verticesWithOutgoingMessages
                .flatMapToPair(vertexPayload -> () ->
                        IteratorUtils.map(vertexPayload._2().detachOutgoingMessages(), // this removes all outgoing messages once they have been iterated by this step
                                message -> new Tuple2<>(message._1(), new SparkMessagePayload<>(message._2()))))
                .reduceByKey((messagePayloadA, messagePayloadB) -> {
                    messagePayloadA.addMessages(messagePayloadB.getMessages(), messageCombiner);
                    return messagePayloadA;
                });

        // join the incoming messages with the vertices
        final JavaPairRDD<Object, SparkVertexPayload<M>> verticesWithIncomingMessages = verticesWithOutgoingMessages
                .leftOuterJoin(incomingMessages)
                .mapValues(tuple -> {
                    final SparkVertexPayload<M> vertexPayload = tuple._1();
                    final SparkMessagePayload<M> messagePayload = tuple._2().orNull();
                    if (null != messagePayload) {
                        vertexPayload.getOutgoingMessages().clear();  // there should be no outgoing messages at this point (just to be safe)
                        vertexPayload.getMessages().clear();          // there should be no incoming messages at this point (just to be safe)
                        vertexPayload.getMessages().addAll(messagePayload.getMessages());
                    }
                    return vertexPayload;
                });


        verticesWithIncomingMessages.foreachPartition(partitionIterator -> {
        }); // need to complete a task so its BSP.
        return verticesWithIncomingMessages;
    }

{'fraction_non_alphanumeric': 0.09077809798270893, 'fraction_numerical': 0.002561639449247518, 'mean_word_length': 3.2813570938999312, 'pattern_counts': {'":': 0, '<': 50, '<?xml version=': 0, '>': 68, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2283043', 'n_tokens_mistral': 3232, 'n_tokens_neox': 3067, 'n_words': 796}