Neo4J生产库与灾备库数据复制方案咨询（仅读生产权限）

刚好经手过类似的Neo4j只读权限下的跨实例同步需求，结合Hadoop/Spark技术栈给你一套可落地的周末同步方案，拆解成几个关键步骤：

因为你只有生产库的只读权限，没法用Neo4j自带的集群复制、增量同步（这些需要生产库的配置修改或写权限），所以核心思路是：

1. 用Spark通过Neo4j官方连接器读取生产库全量/增量数据
2. 可选借助HDFS做中间数据缓冲（应对超大数据集、断点续传场景）
3. 用Spark将数据写入灾备库（需要灾备库的读写权限）
4. 用调度工具（比如Airflow）配置周末定时执行

1. 环境准备 #

• 部署Spark 3.x+：确保能引入Neo4j Spark Connector（提交任务时可通过--packages org.neo4j:neo4j-spark-connector_2.12:5.2.0加载）
• 可选部署HDFS：用于存储抽取的中间数据，避免重复读取生产库
• 调度工具：推荐用Apache Airflow，配置周末定时任务
• 权限确认：生产库只读账号（需能执行MATCH查询）、灾备库读写账号（需能执行CREATE/MERGE/DELETE操作）

2. 生产库数据抽取（Spark读取） #

用PySpark编写抽取脚本，通过Cypher查询全量节点和关系数据，示例如下：

from pyspark.sql import SparkSession

# 初始化Spark会话，连接生产Neo4j
spark = SparkSession.builder \
    .appName("Neo4j-Prod-Extract") \
    .config("spark.neo4j.bolt.url", "bolt://prod-neo4j-host:7687") \
    .config("spark.neo4j.authentication.username", "readonly-user") \
    .config("spark.neo4j.authentication.password", "readonly-pass") \
    .config("spark.neo4j.query.partition", "10")  # 分10个并行任务读取，优化性能
    .getOrCreate()

# 抽取节点数据（建议返回业务主键，避免依赖Neo4j本地ID）
nodes_df = spark.read.format("org.neo4j.spark.DataSource") \
    .option("query", """
        MATCH (n) 
        RETURN 
            n.business_id AS business_id,  # 替换成你的节点业务主键
            labels(n) AS node_labels, 
            properties(n) AS node_props
    """) \
    .load()

# 抽取关系数据
rels_df = spark.read.format("org.neo4j.spark.DataSource") \
    .option("query", """
        MATCH (a)-[r]->(b) 
        RETURN 
            a.business_id AS start_business_id,
            b.business_id AS end_business_id,
            type(r) AS rel_type,
            properties(r) AS rel_props
    """) \
    .load()

# 可选：将数据写入HDFS做缓冲
nodes_df.write.mode("overwrite").parquet("hdfs://hdfs-namenode:9000/neo4j-sync/weekly/nodes/")
rels_df.write.mode("overwrite").parquet("hdfs://hdfs-namenode:9000/neo4j-sync/weekly/rels/")

3. 灾备库数据写入 #

编写写入脚本，从HDFS读取中间数据（或直接用抽取的DataFrame），写入灾备库。如果是全量同步，建议用MERGE保证幂等性：

from pyspark.sql import SparkSession

# 初始化Spark会话，连接灾备Neo4j
spark_dr = SparkSession.builder \
    .appName("Neo4j-DR-Write") \
    .config("spark.neo4j.bolt.url", "bolt://dr-neo4j-host:7687") \
    .config("spark.neo4j.authentication.username", "write-user") \
    .config("spark.neo4j.authentication.password", "write-pass") \
    .config("spark.neo4j.batch.size", "5000")  # 批量写入大小，优化性能
    .getOrCreate()

# 从HDFS读取节点数据
nodes_df_dr = spark_dr.read.parquet("hdfs://hdfs-namenode:9000/neo4j-sync/weekly/nodes/")

# 写入节点：用MERGE保证幂等性，即使灾备库已有数据也能正确更新
nodes_df_dr.write.format("org.neo4j.spark.DataSource") \
    .option("query", """
        MERGE (n {business_id: $business_id}) 
        SET n = $node_props, n:$node_labels
    """) \
    .mode("overwrite") \
    .save()

# 从HDFS读取关系数据
rels_df_dr = spark_dr.read.parquet("hdfs://hdfs-namenode:9000/neo4j-sync/weekly/rels/")

# 写入关系：通过业务主键匹配节点
rels_df_dr.write.format("org.neo4j.spark.DataSource") \
    .option("query", """
        MATCH (a {business_id: $start_business_id}), (b {business_id: $end_business_id})
        MERGE (a)-[r:$rel_type]->(b)
        SET r = $rel_props
    """) \
    .mode("overwrite") \
    .save()

4. 定时调度（Airflow） #

创建Airflow DAG，配置每周周末执行同步任务：

from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'neo4j-sync-team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(hours=1),
}

with DAG(
    'neo4j_prod_to_dr_weekly_sync',
    default_args=default_args,
    description='Weekly full sync from Neo4j Production to DR',
    schedule_interval='0 2 * * 0',  # 每周日凌晨2点执行
    catchup=False,
) as dag:

    extract_task = SparkSubmitOperator(
        task_id='extract_prod_data',
        application='/opt/spark/scripts/extract_neo4j_prod.py',
        conn_id='spark_cluster',
        packages='org.neo4j:neo4j-spark-connector_2.12:5.2.0',
        verbose=True,
    )

    write_task = SparkSubmitOperator(
        task_id='write_dr_data',
        application='/opt/spark/scripts/write_neo4j_dr.py',
        conn_id='spark_cluster',
        packages='org.neo4j:neo4j-spark-connector_2.12:5.2.0',
        verbose=True,
    )

    # 任务依赖：先抽取再写入
    extract_task >> write_task

• 增量同步优化：如果全量同步耗时太长，可在生产库节点/关系中添加updated_at字段，修改Cypher查询为MATCH (n) WHERE n.updated_at >= date_sub(current_date(), 7) RETURN ...，只同步一周内更新的数据
• 性能调优：调整Spark并行度（spark.sql.shuffle.partitions）、Neo4j Connector的batch.size，给生产库的查询字段加索引，避免全表扫描
• 数据校验：同步完成后，添加校验步骤：统计生产库和灾备库的节点/关系数量，或抽样校验数据一致性，比如：

# 校验节点数
prod_node_count = spark.read.format("org.neo4j.spark.DataSource").option("query", "MATCH (n) RETURN count(n) AS cnt").load().collect()[0]['cnt']
dr_node_count = spark_dr.read.format("org.neo4j.spark.DataSource").option("query", "MATCH (n) RETURN count(n) AS cnt").load().collect()[0]['cnt']
if prod_node_count != dr_node_count:
    raise Exception(f"Node count mismatch: Production {prod_node_count}, DR {dr_node_count}")

• 异常处理：在Spark脚本中添加日志记录和异常捕获，Airflow配置邮件告警，确保同步失败能及时通知运维人员

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