说明

在图数据库的日常使用中，将数据导出为通用格式（如 CSV）是一项非常常见的需求，无论是用于数据备份、迁移、分析还是与其他系统集成。对于 Neo4j 用户而言，APOC 插件是功能最强大的导出工具。然而，在某些情况下，您可能没有权限或不想在服务器上安装任何插件。

本文将介绍一种轻量级、无需任何服务器插件的解决方案：使用 Python 脚本直接连接 Neo4j 数据库，并将所有节点数据优雅地导出为 CSV 文件。

一、方案概述

我们将创建一个 Python 脚本，它将完成以下任务：

1. 连接数据库：使用 Neo4j 官方 Python 驱动连接到您的 Neo4j 实例。
2. 执行查询：运行一个 Cypher 查询来获取数据库中的所有节点及其属性和标签。
3. 动态处理表头：智能地收集所有节点的所有属性名，自动生成一个完整的 CSV 表头，即使不同节点的属性不一致也能完美处理。
4. 数据格式化：将查询结果转换为适合 CSV 格式的行数据，并包含节点的标签信息。
5. 写入文件：将处理好的数据写入一个 CSV 文件，并使用 utf-8-sig 编码以确保 Microsoft Excel 等工具能正确识别中文字符。

二、准备工作

在开始之前，请确保您已完成以下准备：

1. 安装 Python：如果尚未安装，请从 Python 官网 下载并安装。
2. 安装 Neo4j Python 驱动：打开您的命令行工具（CMD, PowerShell, 或 Terminal），运行以下命令来安装官方驱动包：

   bash

   运行

   pip install neo4j
   

3. 确认 Neo4j 连接信息：您需要知道 Neo4j 数据库的连接 URI、用户名和密码。

三、完整 Python 脚本

下面是完整的 Python 脚本。您可以将其保存为 export_neo4j_to_csv.py 文件。

python

运行

from neo4j import GraphDatabase
import csv
import os

# --- 请根据您的实际情况修改以下配置 ---
# Neo4j 数据库连接信息
NEO4J_URI = "bolt://localhost:7687"
NEO4J_USER = "neo4j"
NEO4J_PASSWORD = "picasso-6955"  # <--- 替换为您的密码

# Cypher 查询语句，用于获取所有节点的属性和标签
# properties(n) 返回节点的所有属性，labels(n) 返回节点的所有标签
CYPHER_QUERY = "MATCH (n) RETURN properties(n) as props, labels(n) as labels"

# 导出的 CSV 文件路径和名称
OUTPUT_CSV_FILE = "D:/neo4j/data/all_nodes_exported.csv"


class Neo4jCsvExporter:
    """
    一个用于将 Neo4j 节点数据导出到 CSV 文件的类。
    """

    def __init__(self, uri, user, password):
        """
        初始化并连接到 Neo4j 数据库。
        """
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
        print("✅ 成功连接到 Neo4j 数据库。")

    def close(self):
        """
        关闭数据库连接。
        """
        self.driver.close()
        print("🔌 数据库连接已关闭。")

    def export_to_csv(self, query, output_file):
        """
        执行查询并将结果导出到 CSV 文件。
        """
        print("🚀 开始执行查询，请稍候...")
        with self.driver.session() as session:
            # 使用 .data() 获取所有结果，返回一个字典列表
            results = session.run(query).data()

        node_count = len(results)
        print(f"📊 查询完成，共找到 {node_count} 个节点。")

        if node_count == 0:
            print("ℹ️ 没有找到任何节点，导出操作已取消。")
            return

        # 1. 动态收集所有可能的表头（属性名）
        all_headers = set()
        for record in results:
            props = record.get('props', {})
            if props:
                all_headers.update(props.keys())
        # 将 'labels' 也加入表头
        all_headers.add('labels')
        # 转换为有序列表，保证每次导出的列顺序一致
        all_headers = sorted(list(all_headers))

        # 2. 准备要写入 CSV 的行数据
        csv_rows = []
        for record in results:
            props = record.get('props', {})
            labels = record.get('labels', [])

            row_data = {}
            # 为每个表头填充值，如果节点没有该属性，则为空字符串
            for header in all_headers:
                if header == 'labels':
                    # 将标签列表用 '|' 连接成一个字符串
                    row_data[header] = '|'.join(labels)
                else:
                    # 获取属性值，如果不存在则为空字符串
                    row_data[header] = props.get(header, '')

            csv_rows.append(row_data)

        # 3. 写入 CSV 文件
        # 确保输出目录存在
        output_dir = os.path.dirname(output_file)
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
            print(f"已创建目录: {output_dir}")

        print(f"正在将数据写入 CSV 文件: {output_file}")
        try:

            # encoding='utf-8-sig' 确保 Excel 能正确识别中文
            with open(output_file, 'w', newline='', encoding='utf-8-sig') as f:
                # 使用 DictWriter，它可以直接根据字典的键来写入列
                writer = csv.DictWriter(f, fieldnames=all_headers)

                # 写入表头
                writer.writeheader()

                # 写入所有行数据
                writer.writerows(csv_rows)

            print("=" * 40)
            print(f"导出成功！文件已保存至: {output_file}")
            print(f"CSV 文件包含表头: {', '.join(all_headers)}")
            print("=" * 40)

        except Exception as e:
            print(f" 写入文件时发生错误: {e}")


# --- 主程序入口 ---
if __name__ == "__main__":
    exporter = None
    try:
        # 创建导出器实例
        exporter = Neo4jCsvExporter(NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD)
        # 执行导出操作
        exporter.export_to_csv(CYPHER_QUERY, OUTPUT_CSV_FILE)
    except Exception as e:
        print(f"程序执行失败: {e}")
    finally:
        if exporter:
            exporter.close()

四、如何使用

1、修改配置：

         打开 export_neo4j_to_csv.py 文件，将 NEO4J_PASSWORD 和 OUTPUT_CSV_FILE 修改为您自己的信息。

2、运行脚本

1. 在命令行中，切换到您保存脚本的目录，然后运行：

   bash

   运行

   python export_neo4j_to_csv.py
   

2. 检查结果：脚本运行完毕后，您将在指定的输出路径下找到一个名为 all_nodes_exported.csv 的文件。

五、脚本核心功能解析

• from neo4j import GraphDatabase：导入 Neo4j 官方 Python 驱动的核心模块，用于建立数据库连接。
• Neo4jCsvExporter 类：将所有功能封装在一个类中，使代码结构更清晰，易于复用。
• __init__ 和 close 方法：负责数据库连接的建立和关闭，确保资源被正确管理。
• export_to_csv 方法：这是脚本的核心。
  • session.run(query).data()：执行 Cypher 查询并一次性获取所有结果。.data() 方法返回一个字典列表，每个字典代表一行查询结果，非常方便处理。
  • 动态表头生成：通过遍历所有节点的属性，使用 set 数据结构自动收集所有不重复的属性名作为 CSV 的表头。这解决了节点属性不一致的问题。
  • 处理标签：将节点的标签列表 ['Person', 'Employee'] 转换为字符串 'Person|Employee'，方便在 CSV 中存储和查看。
  • csv.DictWriter：Python 内置的 csv 模块中的一个强大工具。它允许您使用字典来写入行，字典的键会自动匹配 CSV 的表头，极大地简化了数据写入过程。
  • encoding='utf-8-sig'：这是一个非常重要的细节。标准的 utf-8 编码在 Excel 中打开可能会出现中文乱码，而 utf-8-sig 在文件开头添加了一个 BOM (Byte Order Mark)，Excel 能够正确识别它，从而完美显示中文字符。

六、总结

这个 Python 脚本为您提供了一个灵活、强大且无插件依赖的 Neo4j 数据导出方案。它不仅能满足基本的导出需求，还通过动态表头和中文编码处理等细节，提升了用户体验和数据兼容性。

1、与 APOC 插件的对比

• APOC 插件：功能更全面，在服务器端执行，性能极高，适合大规模数据导出和自动化运维。是生产环境的首选。
• Python 脚本：轻量级，无需服务器权限，配置简单，易于根据特定需求进行修改和扩展。非常适合临时的、一次性的数据导出任务，或者作为应用程序的一部分。

2、扩展方向

• 导出关系：您可以轻松修改 Cypher 查询和数据处理逻辑，以支持导出节点之间的关系。
• 增量导出：通过添加时间戳或 ID 过滤条件，可以实现只导出新增或修改的数据。
• 命令行参数化：使用 argparse 库，可以将数据库连接信息和输出路径作为命令行参数传入，使脚本更加通用。

     使用 Python脚本将Neo4j数据导出代码，希望能帮助到大家！