将 Spark Streaming RDD 推送到 Neo4j -Scala答案

【问题标题】：Pushing Spark Streaming RDDs to Neo4j -Scala将 Spark Streaming RDD 推送到 Neo4j -Scala
【发布时间】：2015-06-26 04:50:51
【问题描述】：

我需要建立从 Spark Streaming 到 Neo4j 图数据库的连接。RDD 的类型为((is,I),(am,Hello)(sam,happy)....)。我需要在 Neo4j 中的每对单词之间建立一个边缘。

在我发现的 Spark Streaming 文档中

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 包装器。我可以用这些来完成工作吗？如果是这样，我该怎么做？如果没有，还有更好的选择吗？

谢谢大家......

【问题讨论】：

标签： scala neo4j apache-spark spark-streaming anormcypher

【解决方案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 个语句，那就更好了。

【讨论】：

感谢您的洞察力。我能够将 RDD 推送到 neo4j 以创建节点。我对使用 AnormCypher 在节点之间创建关系感到有点震惊。我浏览了 AnormCypher 的文档，但我没有发现如何创建关系。我的代码就像 - val sampleData = sc.parallelize(List("pandas","i like pandas","most pandas live in china")) sampleData.flatMap( _.split(" ")) .map( w = > Cypher("CREATE(:Word {text:{text}})") .on( "text" -> w ).execute() ).collect() - 如果我想在“panda”和“中国”我应该如何处理？

【解决方案2】：

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

【讨论】：

谢谢。我已经研究过了，但它只处理将图形处理作业从 Neo4j 推送到 Spark-GraphX。我需要将 RDD 推送到 Neo4j 中。