title: JanusGraph BulkLoading CSV with ScriptInputFormat
body: <p>I am trying to load a CSV file into a JanusGraph.
  As I understand it, I need to create my graph and the schema, then use the <code>BulkLoaderVertexProgram</code> with my own custom groovy script to parse a csv file.
  Doing that, it seems to work as I can see the vertices, but the edges are not created.</p>
  <p>My configuration seems to be pretty much the same as all the examples I can find, but there must be something I'm missing.</p>
  <p>Is it possible to bulk load edges from a CSV file?</p>
  <hr />
  <p><strong>Here is my setup:</strong></p>
  <p>I'm starting cassandra with the default bin/janusgraph.sh script</p>
  <p><strong>My gremlin commands:</strong></p>
  <pre><code>gremlin&gt; :load data/defineNCBIOSchema.groovy
  ==&gt;true
  gremlin&gt; graph = JanusGraphFactory.open('conf/gremlin-server/socket-janusgraph-apr-test.properties')
  ==&gt;standardjanusgraph[cassandrathrift:[127.0.0.1]]
  gremlin&gt; defineNCBIOSchema(graph)
  ==&gt;null
  gremlin&gt; graph.close()
  ==&gt;null
  
  gremlin&gt; graph = GraphFactory.open('conf/hadoop-graph/apr-test-hadoop-script.properties')
  ==&gt;hadoopgraph[scriptinputformat-&gt;graphsonoutputformat]
  gremlin&gt; blvp = BulkLoaderVertexProgram.build().bulkLoader(OneTimeBulkLoader).writeGraph('conf/gremlin-server/socket-janusgraph-apr-test.properties').create(graph)
  ==&gt;BulkLoaderVertexProgram[bulkLoader=IncrementalBulkLoader, vertexIdProperty=bulkLoader.vertex.id, userSuppliedIds=false, keepOriginalIds=true, batchSize=0]
  gremlin&gt; graph.compute(SparkGraphComputer).workers(1).program(blvp).submit().get()
  ==&gt;result[hadoopgraph[scriptinputformat-&gt;graphsonoutputformat],memory[size:0]]
  gremlin&gt; graph.close()
  ==&gt;null
  
  gremlin&gt; graph = GraphFactory.open('conf/hadoop-graph/apr-test-hadoop-load.properties')
  ==&gt;hadoopgraph[cassandrainputformat-&gt;gryooutputformat]
  gremlin&gt; g = graph.traversal().withComputer(SparkGraphComputer)
  ==&gt;graphtraversalsource[hadoopgraph[cassandrainputformat-&gt;gryooutputformat], sparkgraphcomputer]
  gremlin&gt; g.E() &lt;--- returns nothing
  </code></pre>
  <p><strong>My JanusGraph: (conf/gremlin-server/socket-janusgraph-apr-test.properties)</strong></p>
  <pre><code>gremlin.graph=org.janusgraph.core.JanusGraphFactory
  storage.backend=cassandrathrift
  storage.hostname=127.0.0.1
  cache.db-cache = true
  cache.db-cache-clean-wait = 20
  cache.db-cache-time = 180000
  cache.db-cache-size = 0.25
  index.search.backend=elasticsearch
  index.search.directory=/tmp/searchindex
  index.search.elasticsearch.client-only=false
  index.search.elasticsearch.local-mode=true
  index.search.hostname=127.0.0.1
  </code></pre>
  <p><strong>My graph for the bulkLoader: (conf/hadoop-graph/apr-test-hadoop-script.properties)</strong></p>
  <pre><code>gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph
  gremlin.hadoop.graphInputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.script.ScriptInputFormat
  gremlin.hadoop.graphOutputFormat=org.apache.hadoop.mapreduce.lib.output.NullOutputFormat
  gremlin.hadoop.jarsInDistributedCache=true
  
  gremlin.hadoop.inputLocation=data/apr-test-doc.csv
  gremlin.hadoop.scriptInputFormat.script=data/apr-test-CSVInputScript.groovy
  gremlin.hadoop.outputLocation=output
  
  query.fast-property=false
  
  spark.master=local[*]
  spark.executor.memory=1g
  spark.serializer=org.apache.spark.serializer.KryoSerializer
  </code></pre>
  <p><strong>The read graph: (conf/hadoop-graph/apr-test-hadoop-load.properties)</strong></p>
  <pre><code>gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph
  gremlin.hadoop.graphInputFormat=org.janusgraph.hadoop.formats.cassandra.CassandraInputFormat
  gremlin.hadoop.graphOutputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.gryo.GryoOutputFormat
  
  gremlin.hadoop.jarsInDistributedCache=true
  gremlin.hadoop.inputLocation=none
  gremlin.hadoop.outputLocation=output
  janusgraphmr.ioformat.conf.storage.backend=cassandra
  janusgraphmr.ioformat.conf.storage.hostname=localhost
  janusgraphmr.ioformat.conf.storage.port=9160
  janusgraphmr.ioformat.conf.storage.cassandra.keyspace=janusgraph
  cassandra.thrift.framed.size_mb=60
  cassandra.input.partitioner.class=org.apache.cassandra.dht.Murmur3Partitioner
  spark.master=local[*]
  spark.serializer=org.apache.spark.serializer.KryoSerializer
  </code></pre>
  <p><strong>My groovy script</strong></p>
  <pre><code>class Globals {
      static String[] h = [];
      static int lineNumber = 0;
  }
  
  def parse(line, factory) {
      def vertexType = 'Disease'
      def edgeLabel = 'parent'
      def parentsIndex = 2;
      
      Globals.lineNumber++
  
      // columns ignoring quoted ,
      def c = line.split(/,(?=(?:[^\&quot;]*\&quot;[^\&quot;]*\&quot;)*[^\&quot;]*$)/)
      
      //  if first column is Class ID ignore the line, it is the header line
      if (c[0] == /ClassID/) {
          Globals.h = c
          return null
      }
      
      def v1 = graph.addVertex(T.id, c[0], T.label, vertexType)
      
      for (i = 0; i &lt; c.length; ++i) {
          if (i != parentsIndex) { // Ignore parent
              def f = removeInvalidChar(c[i])
              if (f?.trim()) {
                  v1.property(Globals.h[i], f)
              }
          }
      }
      
      def parents = []    
      if (c.length &gt; parentsIndex) {
          parents = c[parentsIndex].split(/\|/)
      }
      
      for (i = 0; i &lt; parents.size(); ++i) {
          def v2 = graph.addVertex(T.id, parents[i], T.label, vertexType)
          v1.addInEdge(edgeLabel, v2)             
      }
  
      return v1
  }
  
  def removeInvalidChar(col) {
  
      def f = col.replaceAll(/^\&quot;|\&quot;$/, &quot;&quot;) // Remove quotes
      f = f.replaceAll(/\{/, /(/) // Remove {
      f = f.replaceAll(/\}/, /)/) // Remove }
      
      if (f == /label/) {
          f = /label2/
      }
  
      return f
  }
  </code></pre>
  <p><strong>The schema</strong></p>
  <pre><code>def defineNCBIOSchema(graph) {
      
      mgmt = graph.openManagement()
      
      // vertex labels
      vertexLabel = mgmt.makeVertexLabel('Disease').make()
      
      // edge labels
      parent = mgmt.makeEdgeLabel('parent').multiplicity(MULTI).make()
      
      // vertex and edge properties
      blid = mgmt.makePropertyKey('bulkLoader.vertex.id').dataType(String.class).make()
      classID = mgmt.makePropertyKey('ClassID').dataType(String.class).cardinality(Cardinality.SINGLE).make()
      preferedLabel = mgmt.makePropertyKey('PreferredLabel').dataType(String.class).cardinality(Cardinality.SINGLE).make()
      
      // global indices
      mgmt.buildIndex('ClassIDIndex', Vertex.class).addKey(classID).unique()
  
      mgmt.commit()
  }
  </code></pre>
  <p><strong>The CSV</strong></p>
  <pre><code>ClassID,PreferredLabel,Parents
  Vertex3,Prefered Label 3,
  Vertex2,Prefered Label 2,Vertex3
  Vertex1,Prefered Label 1,Vertex2|Vertex3
  </code></pre>