如何从MySQL数据库快速导入海量数据至Memgraph？

1. 官方mg_import_csv工具（首选最快方案） #

这是Memgraph原生优化的导入工具，专门针对CSV格式做了性能优化，适合百万级规模的数据集。

• 步骤1：从MySQL导出CSV
  用MySQL的SELECT ... INTO OUTFILE命令直接导出节点和关系数据，避免中间转换：

  -- 导出节点表（示例：nodes表含id、name、label字段）
  SELECT id, name, label 
  FROM nodes 
  INTO OUTFILE '/tmp/nodes.csv' 
  FIELDS TERMINATED BY ',' ENCLOSED BY '"' 
  LINES TERMINATED BY '\n';
  
  -- 导出关系表（示例：relationships表含start_node_id、end_node_id、type字段）
  SELECT start_node_id, end_node_id, type 
  FROM relationships 
  INTO OUTFILE '/tmp/relationships.csv' 
  FIELDS TERMINATED BY ',' ENCLOSED BY '"' 
  LINES TERMINATED BY '\n';
  

  注意：确保MySQL有写入目标路径的权限，路径需是本地文件系统路径。

• 步骤2：用mg_import_csv导入
  执行导入命令，指定节点和关系的CSV文件，以及Memgraph的连接信息：

  mg_import_csv --nodes /tmp/nodes.csv --relationships /tmp/relationships.csv --host localhost --port 7687
  

  该工具会跳过事务日志的写入优化，直接批量写入存储，性能比普通Cypher插入高5-10倍。

2. Bolt协议批量插入（无中间文件方案） #

如果不想生成CSV文件，可通过Bolt协议直接从MySQL读取数据，批量发送Cypher语句插入。核心是用批量UNWIND语句减少网络交互次数。

• Python示例脚本（兼容Memgraph的Bolt驱动）：

  import mysql.connector
  from neo4j import GraphDatabase
  
  # 连接MySQL
  mysql_conn = mysql.connector.connect(
      host="你的MySQL地址",
      user="用户名",
      password="密码",
      database="目标数据库"
  )
  cursor = mysql_conn.cursor(dictionary=True)
  
  # 连接Memgraph
  memgraph_driver = GraphDatabase.driver("bolt://localhost:7687")
  
  # 批量插入节点（每1000条为一批）
  batch_size = 1000
  cursor.execute("SELECT id, name, label FROM nodes")
  batch = []
  for row in cursor:
      batch.append({"id": row["id"], "name": row["name"], "label": row["label"]})
      if len(batch) == batch_size:
          with memgraph_driver.session() as session:
              session.run("""
                  UNWIND $batch AS node_data
                  CREATE (n:`{label}` {{id: node_data.id, name: node_data.name}})
              """.format(label=batch[0]["label"]), batch=batch)
          batch = []
  # 处理剩余数据
  if batch:
      with memgraph_driver.session() as session:
          session.run("""
              UNWIND $batch AS node_data
              CREATE (n:`{label}` {{id: node_data.id, name: node_data.name}})
          """.format(label=batch[0]["label"]), batch=batch)
  
  # 批量插入关系（逻辑类似）
  cursor.execute("SELECT start_node_id, end_node_id, type FROM relationships")
  batch = []
  for row in cursor:
      batch.append({
          "start_id": row["start_node_id"],
          "end_id": row["end_node_id"],
          "rel_type": row["type"]
      })
      if len(batch) == batch_size:
          with memgraph_driver.session() as session:
              session.run("""
                  UNWIND $batch AS rel_data
                  MATCH (s {id: rel_data.start_id}), (e {id: rel_data.end_id})
                  CREATE (s)-[:`{rel_type}`]->(e)
              """.format(rel_type=batch[0]["rel_type"]), batch=batch)
          batch = []
  if batch:
      with memgraph_driver.session() as session:
          session.run("""
              UNWIND $batch AS rel_data
              MATCH (s {id: rel_data.start_id}), (e {id: rel_data.end_id})
              CREATE (s)-[:`{rel_type}`]->(e)
          """.format(rel_type=batch[0]["rel_type"]), batch=batch)
  
  mysql_conn.close()
  memgraph_driver.close()
  

• 优势：无需生成中间文件，适合需要实时转换数据的场景，批量插入比单条插入效率提升数十倍。

3. 自定义ETL脚本（复杂数据映射场景） #

如果你的MySQL数据需要复杂的关联转换（比如多表join生成节点/关系），可以用Memgraph的mgp模块编写C++或Python扩展，直接在Memgraph进程内读取MySQL数据并插入，减少网络和序列化开销。

关键优化建议 #

• 导入前关闭索引/约束：先删除所有节点标签索引和关系类型约束，导入完成后再重新创建，避免每次插入触发索引更新。
• 分配足够内存：百万级数据集建议Memgraph的可用内存至少是数据集大小的2-3倍，避免频繁磁盘交换。
• 并行处理：如果用脚本插入，可多线程读取MySQL数据并批量插入（注意Memgraph的并发写入限制，避免过载）。

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