使用Spark2与Neo4j Spark Connector向单节点Neo4j3导入数据的方法咨询

你完全不需要手动映射每个DataFrame并逐个编写查询执行——Neo4j Spark Connector提供了更简洁的方式来直接将Spark DataFrame的数据导入Neo4j，结合Databricks文档里的最佳实践，这里给你两种常用的实现方案：

方案一：使用Connector的save方法一键导入 #

这是最省心的方式，只需配置好连接参数和映射规则，就能快速完成数据导入，适合简单的节点或关系导入场景。

导入节点示例（Spark 2.x） #

import org.neo4j.spark._

// 假设你的DataFrame结构为：id: Int, name: String, age: Int
val userDF = spark.read.csv("your-input-data-path").toDF("id", "name", "age")

// 配置Neo4j连接（你已完成认证，这里补充导入相关配置即可）
val neo4jConnConfig = Neo4jConfig(
  "bolt://localhost:7687",
  ("user", "your-username"),
  ("password", "your-password")
)

// 将DataFrame数据导入为Neo4j的Person节点
userDF.write
  .format("org.neo4j.spark.DataSource")
  .mode("append") // 可选模式：append/overwrite/ignore
  .option("url", neo4jConnConfig.url)
  .option("authentication.basic.username", neo4jConnConfig.username)
  .option("authentication.basic.password", neo4jConnConfig.password)
  .option("labels", ":Person") // 给节点添加的标签
  .option("keys", "id,name,age") // 要映射为节点属性的DataFrame列名
  .save()

导入关系示例 #

如果需要导入节点间的关系，只需调整配置参数：

// 假设关系DataFrame结构：srcUserId: Int, dstUserId: Int, relation: String
val relationDF = spark.read.csv("your-relation-data-path").toDF("srcUserId", "dstUserId", "relation")

relationDF.write
  .format("org.neo4j.spark.DataSource")
  .mode("append")
  .option("url", neo4jConnConfig.url)
  .option("authentication.basic.username", neo4jConnConfig.username)
  .option("authentication.basic.password", neo4jConnConfig.password)
  .option("relationship", "FOLLOWS") // 关系类型
  .option("relationship.save.strategy", "keys")
  .option("relationship.source.labels", ":Person") // 起始节点的标签
  .option("relationship.source.node.keys", "srcUserId:id") // 起始节点匹配规则（DataFrame列:Neo4j节点属性）
  .option("relationship.target.labels", ":Person") // 目标节点的标签
  .option("relationship.target.node.keys", "dstUserId:id") // 目标节点匹配规则
  .save()

方案二：自定义Cypher查询实现复杂导入 #

如果你的导入逻辑需要更灵活的处理（比如合并重复节点、设置动态属性或关系类型），可以通过foreachBatch（Spark 2.3+支持）或foreachPartition来执行自定义Cypher语句，这种方式能满足个性化的业务需求。

批量导入示例（使用foreachBatch） #

import org.apache.spark.sql.ForeachWriter
import org.neo4j.driver.v1.{GraphDatabase, Values}

// 批处理场景下使用foreachBatch
userDF.write.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
  batchDF.foreachPartition { partition =>
    // 每个分区创建一个Neo4j连接，减少连接开销
    val driver = GraphDatabase.driver(neo4jConnConfig.url, org.neo4j.driver.v1.auth.basic(neo4jConnConfig.username, neo4jConnConfig.password))
    val session = driver.session()
    
    try {
      partition.foreach { row =>
        val userId = row.getAs[Int]("id")
        val userName = row.getAs[String]("name")
        val userAge = row.getAs[Int]("age")
        
        // 使用MERGE避免重复创建节点，同时更新属性
        val cypherQuery = """MERGE (p:Person {id: $userId}) 
                            |SET p.name = $userName, p.age = $userAge""".stripMargin
        session.run(cypherQuery, Values.parameters("userId", userId, "userName", userName, "userAge", userAge))
      }
    } finally {
      // 确保连接关闭
      session.close()
      driver.close()
    }
  }
}

额外提示 #

• 优先使用方案一处理简单导入场景，减少代码量和出错概率；
• 批量导入时，用foreachPartition代替foreach，能大幅降低Neo4j连接的创建次数，提升导入性能；
• 注意版本兼容性：Spark 2.x要搭配对应版本的Neo4j Spark Connector（比如Connector 2.4.x适配Spark 2.4.x和Neo4j 3.x）。

内容的提问来源于stack exchange，提问作者Gohmz