使用Databricks PySpark连接器执行Neo4j约束创建Cypher查询失败

问题原因 #

你当前使用DataFrame的.write API执行创建约束的Cypher语句，而这个API的设计目的是将DataFrame中的数据写入Neo4j，仅支持数据写入类的Cypher语句（如CREATE/MERGE节点/关系且需关联DataFrame字段）。创建约束属于DDL操作，不符合.write API的要求，因此触发IllegalArgumentException: Please provide a valid WRITE query错误。

解决方案 #

不需要依赖DataFrame的写入逻辑，直接使用Neo4j Spark Connector提供的DDL执行方式即可，以下是两种可行方案：

方案1：使用Neo4jUtil执行DDL语句 #

通过Neo4jUtil工具类直接执行约束创建语句，无需多余的DataFrame：

from pyspark.sql import SparkSession
from org.neo4j.spark.util import Neo4jUtil

# 初始化Spark会话并配置Neo4j连接
spark = SparkSession.builder \
    .appName("Neo4j Integration") \
    .config("spark.neo4j.bolt.url", "bolt://localhost:7687") \
    .config("spark.neo4j.authentication.type", "basic") \
    .config("spark.neo4j.authentication.basic.username", "<username>") \
    .config("spark.neo4j.authentication.basic.password", "password") \
    .getOrCreate()

# 创建约束的Cypher语句
constraints_query = """
CREATE CONSTRAINT unique_person IF NOT EXISTS FOR (p:Person) REQUIRE p.name IS UNIQUE
"""

# 执行DDL语句
Neo4jUtil.executeQuery(spark.sparkContext, constraints_query, database="neo4j")

方案2：使用Spark Read API执行DDL语句 #

利用spark.read API执行DDL操作，无需额外导入工具类：

from pyspark.sql import SparkSession

# 初始化Spark会话
spark = SparkSession.builder \
    .appName("Neo4j Integration") \
    .getOrCreate()

# Neo4j连接配置
neo4j_config = {
    "url": "bolt://localhost:7687",
    "authentication.type": "basic",
    "authentication.basic.username": "<username>",
    "authentication.basic.password": "password"
}

# 创建约束的Cypher语句
constraints_query = """
CREATE CONSTRAINT unique_person IF NOT EXISTS FOR (p:Person) REQUIRE p.name IS UNIQUE
"""

# 执行DDL语句
spark.read.format("org.neo4j.spark.DataSource") \
    .options(**neo4j_config) \
    .option("query", constraints_query) \
    .option("database", "neo4j") \
    .load()

验证方式 #

执行完成后，可在Neo4j浏览器中执行SHOW CONSTRAINTS命令，确认unique_person约束已成功创建。

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