okram

        @Override
        public void cleanup(final Mapper<NullWritable, FaunusVertex, LongWritable, Holder<FaunusVertex>>.Context context) throws IOException, InterruptedException {
            if (this.graph instanceof TransactionalGraph) {
                try {
                    ((TransactionalGraph) this.graph).commit();
                    context.getCounter(Counters.SUCCESSFUL_TRANSACTIONS).increment(1l);
                } catch (Exception e) {
                    LOGGER.error("Could not commit transaction during Map.cleanup():", e);
                    ((TransactionalGraph) this.graph).rollback();
                    context.getCounter(Counters.FAILED_TRANSACTIONS).increment(1l);

okram

catch (final Exception e) {
                if (this.graph instanceof TransactionalGraph) {
                    ((TransactionalGraph) this.graph).rollback();
                    context.getCounter(Counters.FAILED_TRANSACTIONS).increment(1l);
                }
                throw new IOException(e.getMessage(), e);
            }

okram

package com.tinkerpop.gremlin;

import java.util.function.Function;

/**
 * @author Marko A. Rodriguez (http://markorodriguez.com)
 */
public class Monad<S, E> {

    private final Monad<?, S> start;

okram

/**
 * @author Marko A. Rodriguez (http://markorodriguez.com)
 */
        public class TinkerFactory {
            public static TinkerGraph createTinkerGraph() {

                TinkerGraph graph = new TinkerGraph();

                Vertex marko = graph.addVertex(TinkerProperty.of(Property.Key.ID, 1, "name", "marko", "age", 29));

okram

// MONAD DEFINITION

public abstract class Monad<S, E> {

    protected final Monad<?, S> start;

    public Monad(Monad<?, S> start) {
        this.start = start;
    }

okram

        TinkerGraph g = TinkerFactory.createClassic();
        ComputeResult result = g.computer().program(LambdaVertexProgram.create()
                .setup(gm -> {
                })
                .execute((v, gm) -> {
                    v.setProperty("name", "marko" + gm.getIteration());
                })
                .terminate(gm -> gm.getIteration() > 3)
                .computeKeys(VertexProgram.ofComputeKeys("name", VertexProgram.KeyType.VARIABLE))
                .build())

okram

gremlin> g.V.filter{it.name == 'marko'}.map
15:36:47 WARN mapreduce.FaunusCompiler: Using the developer Faunus job jar: target/faunus-0.4.0-SNAPSHOT-job.jar
15:36:47 INFO mapreduce.FaunusCompiler: Compiled to 1 MapReduce job(s)
15:36:47 INFO mapreduce.FaunusCompiler: Executing job 1 out of 1: MapSequence[com.thinkaurelius.faunus.mapreduce.transform.VerticesMap.Map, com.thinkaurelius.faunus.mapreduce.filter.FilterMap.Map, com.thinkaurelius.faunus.mapreduce.transform.PropertyMapMap.Map]
15:36:47 INFO mapreduce.FaunusCompiler: Job data location: output/job-0
15:36:47 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
15:36:48 INFO input.FileInputFormat: Total input paths to process : 1
15:36:48 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15:36:48 WARN snappy.LoadSnappy: Snappy native library not loaded
15:36:48 INFO mapred.JobClient: Running job: job_20130

okram

package com.thinkaurelius.faunus.mapreduce.util;

import groovy.lang.Closure;
import sun.misc.BASE64Decoder;
import sun.misc.BASE64Encoder;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;

okram

    public void testLambdaProgram() {
        //TinkerGraph g = TinkerFactory.createClassic();
        TinkerGraph g = new TinkerGraph();
        Stream.generate(g::addVertex).limit(500000).count();
        ComputeResult result = g.compute().program(LambdaVertexProgram.create()
                .setup(gm -> {
                })
                .execute((v, gm) -> {
                    v.setProperty("i", gm.getIteration());
                })

okram

    public ComputeResult submit() {
        final long time = System.currentTimeMillis();
        this.vertexMemory.setComputeKeys(this.vertexProgram.getComputeKeys());
        this.vertexProgram.setup(this.graphMemory);

        boolean done = false;
        while (!done) {

            final ExecutorService executor = Executors.newFixedThreadPool(this.threads);
            final Iterator<Vertex> vertices = this.graph.query().vertices().iterator();