写入节点

本页中的所有示例均假设 SparkSession 已使用适当的身份验证选项进行了初始化。有关更多详细信息，请参阅快速入门示例。

使用 labels 选项，连接器将 DataFrame 写入 Neo4j 数据库，作为具有给定标签的一组节点。

连接器会构建一个 CREATE 或 MERGE Cypher^® 查询（取决于保存模式），该查询使用 UNWIND 子句写入一批行（由 batch.size 选项定义大小的 events 列表）。

来自示例的代码创建了带有 :Person 标签的新节点。

示例

case class Person(name: String, surname: String, age: Int)

val peopleDF = List(
    Person("John", "Doe", 42),
    Person("Jane", "Doe", 40)
).toDF()

peopleDF.write
    .format("org.neo4j.spark.DataSource")
    .mode(SaveMode.Append)
    .option("labels", ":Person")
    .save()

示例

# Create example DataFrame
peopleDF = spark.createDataFrame(
    [
        {"name": "John", "surname": "Doe", "age": 42},
        {"name": "Jane", "surname": "Doe", "age": 40},
    ]
)

(
    peopleDF.write.format("org.neo4j.spark.DataSource")
    .mode("Append")
    .option("labels", ":Person")
    .save()
)

等效的 Cypher 查询

UNWIND $events AS event
CREATE (n:Person)
SET n += event.properties

可以使用冒号作为分隔符为节点写入多个标签。第一个标签前的冒号是可选的。

示例

peopleDF.write
    .format("org.neo4j.spark.DataSource")
    .mode(SaveMode.Append)
    // ":Person:Employee" and "Person:Employee"
    // are equivalent
    .option("labels", ":Person:Employee")
    .save()

示例

(
    peopleDF.write.format("org.neo4j.spark.DataSource")
    .mode("Append")
    .option("labels", ":Person")
    # ":Person:Employee" and "Person:Employee"
    # are equivalent
    .option("labels", ":Person:Employee")
    .save()
)

节点键

在 Overwrite 模式下，必须指定用于匹配节点的 DataFrame 列作为键。node.keys 选项接受逗号分隔的 key:value 对列表，其中键是 DataFrame 列名，值是节点属性名。

如果 key 与 value 相同，可以省略 value。例如，"name:name,surname:surname" 等同于 "name,surname"。

在使用 Overwrite 保存模式时的相同代码

覆盖示例

df.write
  .format("org.neo4j.spark.DataSource")
  .mode(SaveMode.Overwrite)
  .option("labels", ":Person")
  .option("node.keys", "name,surname")
  .save()

等效的 Cypher 查询

UNWIND $events AS event
MERGE (n:Person {
  name: event.keys.name,
  surname: event.keys.surname
})
SET n += event.properties

由于 Spark 任务的并发性，在使用 Overwrite 模式时，应使用属性唯一性约束来保证节点的唯一性。