<commit_before>package com.thinkaurelius.faunus.mapreduce.blueprints;

import com.thinkaurelius.faunus.FaunusVertex;
import com.thinkaurelius.faunus.Holder;
import com.thinkaurelius.titan.core.TitanFactory;
import com.tinkerpop.blueprints.Edge;
import com.tinkerpop.blueprints.Graph;
import com.tinkerpop.blueprints.TransactionalGraph;
import com.tinkerpop.blueprints.Vertex;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.commons.configuration.BaseConfiguration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;

import static com.tinkerpop.blueprints.Direction.IN;
import static com.tinkerpop.blueprints.Direction.OUT;

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

    public enum Counters {
        VERTICES_WRITTEN,
        VERTEX_PROPERTIES_WRITTEN,
        EDGES_WRITTEN,
        EDGE_PROPERTIES_WRITTEN
    }

    public static final String BLUEPRINTS_ID = "_blueprintsId";
    private static final long MUTATION_COMMITS = 5000;

    private static void commitGraph(final Graph graph, long mutations) {
        if (mutations % MUTATION_COMMITS == 0) {
            if (graph instanceof TransactionalGraph)
                ((TransactionalGraph) graph).stopTransaction(TransactionalGraph.Conclusion.SUCCESS);
        }
    }

    public static class Map extends Mapper<NullWritable, FaunusVertex, LongWritable, Holder<FaunusVertex>> {

        private Graph graph;
        long counter = 0;

        private final Holder<FaunusVertex> vertexHolder = new Holder<FaunusVertex>();
        private final LongWritable longWritable = new LongWritable();
        private final FaunusVertex shellVertex = new FaunusVertex();

        @Override
        public void setup(final Mapper.Context context) throws IOException, InterruptedException {
            Configuration config = context.getConfiguration();
            final BaseConfiguration titanconfig = new BaseConfiguration();
            titanconfig.setProperty("autotype", "blueprints");
            titanconfig.setProperty("storage.backend", "cassandra");
            titanconfig.setProperty("storage.hostname", ConfigHelper.getOutputInitialAddress(config));
            titanconfig.setProperty("storage.keyspace", ConfigHelper.getOutputKeyspace(config));
            titanconfig.setProperty("storage.port", ConfigHelper.getOutputRpcPort(config));
            this.graph = TitanFactory.open(titanconfig);
        }

        @Override
        public void map(final NullWritable key, final FaunusVertex value, final Mapper<NullWritable, FaunusVertex, LongWritable, Holder<FaunusVertex>>.Context context) throws IOException, InterruptedException {

            // Write FaunusVertex (and respective properties) to Titan
            final Vertex vertex = this.graph.addVertex(null);
            for (final String property : value.getPropertyKeys()) {
                vertex.setProperty(property, value.getProperty(property));
            }
            value.setProperty(BLUEPRINTS_ID, vertex.getId());

            // Propagate holders and ids
            for (Edge edge : value.getEdges(IN)) {
                this.longWritable.set((Long) edge.getVertex(OUT).getId());
                this.shellVertex.reuse(value.getIdAsLong());
                this.shellVertex.setProperty(BLUEPRINTS_ID, vertex.getId());
                context.write(this.longWritable, vertexHolder.set('s', this.shellVertex));
            }

            this.longWritable.set(value.getIdAsLong());
            context.write(this.longWritable, vertexHolder.set('v', value));

            WriteGraphMapReduce.commitGraph(this.graph, this.counter++);
        }

        @Override
        public void cleanup(final Mapper<NullWritable, FaunusVertex, LongWritable, Holder<FaunusVertex>>.Context context) throws IOException, InterruptedException {
            if (this.graph instanceof TransactionalGraph)
                ((TransactionalGraph) this.graph).stopTransaction(TransactionalGraph.Conclusion.SUCCESS);
        }
    }

    public static class Reduce extends Reducer<LongWritable, Holder<FaunusVertex>, NullWritable, FaunusVertex> {

        private Graph graph;
        long counter = 0;

        @Override
        public void setup(final Reduce.Context context) throws IOException, InterruptedException {
            Configuration config = context.getConfiguration();
            final BaseConfiguration titanconfig = new BaseConfiguration();
            titanconfig.setProperty("autotype", "blueprints");
            titanconfig.setProperty("storage.backend", "cassandra");
            titanconfig.setProperty("storage.hostname", ConfigHelper.getOutputInitialAddress(config));
            titanconfig.setProperty("storage.keyspace", ConfigHelper.getOutputKeyspace(config));
            titanconfig.setProperty("storage.port", ConfigHelper.getOutputRpcPort(config));
            this.graph = TitanFactory.open(titanconfig);
        }

        @Override
        public void reduce(final LongWritable key, final Iterable<Holder<FaunusVertex>> values, final Reducer<LongWritable, Holder<FaunusVertex>, NullWritable, FaunusVertex>.Context context) throws IOException, InterruptedException {
            final FaunusVertex vertex = new FaunusVertex();
            final java.util.Map<Long, Long> faunusTitanIdMap = new HashMap<Long, Long>();
            for (final Holder<FaunusVertex> holder : values) {
                if (holder.getTag() == 's') {
                    faunusTitanIdMap.put(holder.get().getIdAsLong(), (Long) holder.get().getProperty(BLUEPRINTS_ID));
                } else {
                    vertex.addAll(holder.get());
                }
            }

            final Vertex root = this.graph.getVertex(vertex.getProperty(BLUEPRINTS_ID));
            for (final Edge edge : vertex.getEdges(OUT)) {
                final Edge e = this.graph.addEdge(null, root, this.graph.getVertex(faunusTitanIdMap.get((Long) edge.getVertex(IN).getId())), edge.getLabel());
                for (final String property : edge.getPropertyKeys()) {
                    e.setProperty(property, edge.getProperty(property));
                }
                WriteGraphMapReduce.commitGraph(this.graph, this.counter++);
            }

            vertex.removeProperty(BLUEPRINTS_ID);
            context.write(NullWritable.get(), vertex);
        }

        @Override
        public void cleanup(final Reducer<LongWritable, Holder<FaunusVertex>, NullWritable, FaunusVertex>.Context context) throws IOException, InterruptedException {
            if (this.graph instanceof TransactionalGraph)
                ((TransactionalGraph) this.graph).stopTransaction(TransactionalGraph.Conclusion.SUCCESS);<commit_msg>new MapReduce based graph writer that doesn't pollute graph with indices and provide more flexibility. There is now a sideEffect writeGraph step.<commit_after>package com.thinkaurelius.faunus.mapreduce.blueprints;

import com.thinkaurelius.faunus.FaunusVertex;
import com.thinkaurelius.faunus.Holder;
import com.thinkaurelius.titan.core.TitanFactory;
import com.tinkerpop.blueprints.Edge;
import com.tinkerpop.blueprints.Graph;
import com.tinkerpop.blueprints.TransactionalGraph;
import com.tinkerpop.blueprints.Vertex;
import org.apache.cassandra.hadoop.ConfigHelper;
import org.apache.commons.configuration.BaseConfiguration;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;

import static com.tinkerpop.blueprints.Direction.IN;
import static com.tinkerpop.blueprints.Direction.OUT;

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

    public enum Counters {
        VERTICES_WRITTEN,
        VERTEX_PROPERTIES_WRITTEN,
        EDGES_WRITTEN,
        EDGE_PROPERTIES_WRITTEN
    }

    public static final String BLUEPRINTS_ID = "_blueprintsId";
    private static final long MUTATION_COMMITS = 5000;

    private static boolean commitGraph(final Graph graph, final long mutations) {
        if (mutations % MUTATION_COMMITS == 0) {
            if (graph instanceof TransactionalGraph) {
                ((TransactionalGraph) graph).stopTransaction(TransactionalGraph.Conclusion.SUCCESS);
                return true;
            }
        }
        return false;
    }

    public static class Map extends Mapper<NullWritable, FaunusVertex, LongWritable, Holder<FaunusVertex>> {

        private Graph graph;
        long counter = 0;

        private final Holder<FaunusVertex> vertexHolder = new Holder<FaunusVertex>();
        private final LongWritable longWritable = new LongWritable();
        private final FaunusVertex shellVertex = new FaunusVertex();

        @Override
        public void setup(final Mapper.Context context) throws IOException, InterruptedException {
            Configuration config = context.getConfiguration();
            final BaseConfiguration titanconfig = new BaseConfiguration();
            titanconfig.setProperty("autotype", "blueprints");
            titanconfig.setProperty("storage.backend", "cassandra");
            titanconfig.setProperty("storage.hostname", ConfigHelper.getOutputInitialAddress(config));
            titanconfig.setProperty("storage.keyspace", ConfigHelper.getOutputKeyspace(config));
            titanconfig.setProperty("storage.port", ConfigHelper.getOutputRpcPort(config));
            this.graph = TitanFactory.open(titanconfig);
        }

        @Override
        public void map(final NullWritable key, final FaunusVertex value, final Mapper<NullWritable, FaunusVertex, LongWritable, Holder<FaunusVertex>>.Context context) throws IOException, InterruptedException {

            // Write FaunusVertex (and respective properties) to Blueprints Graph
            final Vertex titanVertex = this.graph.addVertex(null);
            context.getCounter(Counters.VERTICES_WRITTEN).increment(1l);
            for (final String property : value.getPropertyKeys()) {
                titanVertex.setProperty(property, value.getProperty(property));
                context.getCounter(Counters.VERTEX_PROPERTIES_WRITTEN).increment(1l);
            }
            value.setProperty(BLUEPRINTS_ID, titanVertex.getId());

            // Propagate shell vertices with Blueprints ids
            this.shellVertex.reuse(value.getIdAsLong());
            this.shellVertex.setProperty(BLUEPRINTS_ID, titanVertex.getId());
            // TODO: Might need to be OUT for the sake of unidirectional edges in Titan
            for (final Edge faunusEdge : value.getEdges(IN)) {
                this.longWritable.set((Long) faunusEdge.getVertex(OUT).getId());
                context.write(this.longWritable, this.vertexHolder.set('s', this.shellVertex));
            }

            this.longWritable.set(value.getIdAsLong());
            context.write(this.longWritable, this.vertexHolder.set('v', value));

            // after so many mutations, successfully commit the transaction (if graph is transactional)
            WriteGraphMapReduce.commitGraph(this.graph, ++this.counter);
        }

        @Override
        public void cleanup(final Mapper<NullWritable, FaunusVertex, LongWritable, Holder<FaunusVertex>>.Context context) throws IOException, InterruptedException {
            if (this.graph instanceof TransactionalGraph)
                ((TransactionalGraph) this.graph).stopTransaction(TransactionalGraph.Conclusion.SUCCESS);
        }
    }

    public static class Reduce extends Reducer<LongWritable, Holder<FaunusVertex>, NullWritable, FaunusVertex> {

        private Graph graph;
        long counter = 0;

        @Override
        public void setup(final Reduce.Context context) throws IOException, InterruptedException {
            Configuration config = context.getConfiguration();
            final BaseConfiguration titanconfig = new BaseConfiguration();
            titanconfig.setProperty("autotype", "blueprints");
            titanconfig.setProperty("storage.backend", "cassandra");
            titanconfig.setProperty("storage.hostname", ConfigHelper.getOutputInitialAddress(config));
            titanconfig.setProperty("storage.keyspace", ConfigHelper.getOutputKeyspace(config));
            titanconfig.setProperty("storage.port", ConfigHelper.getOutputRpcPort(config));
            this.graph = TitanFactory.open(titanconfig);
        }

        @Override
        public void reduce(final LongWritable key, final Iterable<Holder<FaunusVertex>> values, final Reducer<LongWritable, Holder<FaunusVertex>, NullWritable, FaunusVertex>.Context context) throws IOException, InterruptedException {
            final FaunusVertex faunusVertex = new FaunusVertex();
            // generate a map of the faunus id with the blueprints id for all shell vertices (vertices incoming adjacent)
            final java.util.Map<Long, Long> faunusTitanIdMap = new HashMap<Long, Long>();
            for (final Holder<FaunusVertex> holder : values) {
                if (holder.getTag() == 's') {
                    faunusTitanIdMap.put(holder.get().getIdAsLong(), (Long) holder.get().getProperty(BLUEPRINTS_ID));
                } else {
                    faunusVertex.addAll(holder.get());
                }
            }

            Vertex titanVertex = this.graph.getVertex(faunusVertex.getProperty(BLUEPRINTS_ID));
            for (final Edge faunusEdge : faunusVertex.getEdges(OUT)) {
                final Edge titanEdge = this.graph.addEdge(null, titanVertex, this.graph.getVertex(faunusTitanIdMap.get((Long) faunusEdge.getVertex(IN).getId())), faunusEdge.getLabel());
                context.getCounter(Counters.EDGES_WRITTEN).increment(1l);
                for (final String property : faunusEdge.getPropertyKeys()) {
                    titanEdge.setProperty(property, faunusEdge.getProperty(property));
                    context.getCounter(Counters.EDGE_PROPERTIES_WRITTEN).increment(1l);
                }
                // after so many mutations, successfully commit the transaction (if graph is transactional)
                // for titan, if the transaction is committed, need to 'reget' the vertex
                if (WriteGraphMapReduce.commitGraph(this.graph, ++this.counter))
                    titanVertex = this.graph.getVertex(titanVertex);
            }

            // this is a sideEffect, thus remove the created blueprints id property
            faunusVertex.removeProperty(BLUEPRINTS_ID);
            context.write(NullWritable.get(), faunusVertex);
        }

        @Override
        public void cleanup(final Reducer<LongWritable, Holder<FaunusVertex>, NullWritable, FaunusVertex>.Context context) throws IOException, InterruptedException {
            if (this.graph instanceof TransactionalGraph)
                ((TransactionalGraph) this.graph).stopTransaction(TransactionalGraph.Conclusion.SUCCESS);
{'fraction_non_alphanumeric': 0.09382813528095803, 'fraction_numerical': 0.0011843663639952626, 'mean_word_length': 3.104509856872806, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '5016101', 'n_tokens_mistral': 4141, 'n_tokens_neox': 3991, 'n_words': 828}