将星火流RDDs推送至Neo4j -Scala

Question

我需要建立从Neo4j流到database.The图形的连接，database.The RDD是类型的((is，I)，(am，Hello)(sam，database.The).)。我需要在Neo4j中的每对单词之间建立一个边。

在火花流文档中我找到了

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }
}

to the push to the data to an external database.

我是用Scala做的。我不太明白该怎么做？我找到了AnormCypher和Neo4jScala包装器。我能用这些来完成工作吗？如果是的话，我怎样才能做到呢？如果没有，还有其他更好的选择吗？

谢谢大家..。

Answer 1

我和AnormCypher做了个实验

如下所示：

implicit val connection = Neo4jREST.setServer("localhost", 7474, "/db/data/")
val conf = new SparkConf().setAppName("Simple Application")
val sc = new SparkContext(conf)
val logData = sc.textFile(FILE, 4).cache()
val count = logData
  .flatMap( _.split(" "))
  .map( w =>
    Cypher("CREATE(:Word {text:{text}})")
      .on( "text" -> w ).execute()
   ).filter( _ ).count()

Neo4j 2.2.x具有很好的并发写入性能，可以从Spark中使用。因此，您必须向Neo4j写入的并发线程越多，越好。如果您可以对每个请求分批处理100到1000次语句，那么就更好了。

Answer 2

看看MazeRunner (http://www.kennybastani.com/2014/11/using-apache-spark-and-neo4j-for-big.html)，因为它会给您一些想法。