如何加速Neo4j 4.4.10加载64.8万条CDR CSV数据？

问题 #

需将含64.8万条记录、7列、大小48MB的CDR CSV文件导入Neo4j 4.4.10，已耗时4天仍未完成。硬件配置为100GB内存+Intel Xeon E5 CPU，CSV列包含OP_Name、TP_Name等7项，使用的Cypher代码如下：

:auto load csv with headers from 'file:///cdr.csv' as line FIELDTERMINATOR  ','
    with line
    where line['Called_Number'] is not null and line['OP_ANI'] is not null
    with line['OP_ANI'] as OP_Phone,
        (CASE line['OP_Name']
            WHEN 'TIC' THEN 'IRAN'
            ELSE 'Foreign' END) AS OP_country,
        line['Called_Number'] as Called_Phone,
        (CASE line['TP_Name']
            WHEN 'TIC' THEN 'IRAN'
            ELSE 'Foreign' END) AS TP_country,
        line['Setup_Time'] as Setup_Time, 
        line['Duration'] as Duration, 
        line['OP_Price'] as OP_Price
    
    call {
        with  OP_Phone, OP_country, Called_Phone, TP_country, Setup_Time, Duration, OP_Price
        
MERGE (c:Customer{phone: toInteger(Called_Phone)})
            on create set c.country = TP_country
            WITH c, OP_Phone, OP_country, Called_Phone, TP_country, Setup_Time, Duration, OP_Price
            CALL apoc.create.addLabels( c, [ c.country ] ) YIELD node
        
MERGE (c2:Customer{phone: toInteger(OP_Phone)})
            on create set c2.country = OP_country
            WITH c2, OP_Phone, OP_country, Called_Phone, TP_country, Setup_Time, Duration, OP_Price, c
            CALL apoc.create.addLabels( c2, [ c2.country ] )  YIELD node
        
        MERGE (c2)-[r:CALLED{setupTime: Setup_Time, 
                    duration: Duration,
                    OP_Price: OP_Price}]->(c)
       
    } IN TRANSACTIONS

加速导入的优化方案 #

1. 创建唯一索引，解决MERGE全表扫描问题 #

MERGE操作依赖Customer.phone定位节点，必须先创建唯一索引，否则每一次MERGE都会遍历所有节点，这是导入极慢的核心原因：

CREATE CONSTRAINT FOR (c:Customer) REQUIRE c.phone IS UNIQUE;

2. 移除动态标签操作，改用静态标签或属性存储 #

apoc.create.addLabels的动态标签操作会带来大量额外性能损耗。由于你的国家标签只有IRAN和Foreign两种，可直接用静态标签或属性替代：

• 方案1（需标签过滤场景）：MERGE时指定静态标签

-- 处理被叫节点
MERGE (c:Customer:${TP_country}{phone: toInteger(Called_Phone)})
ON CREATE SET c.country = TP_country

-- 处理主叫节点
MERGE (c2:Customer:${OP_country}{phone: toInteger(OP_Phone)})
ON CREATE SET c2.country = OP_country

• 方案2（通用场景，性能更优）：仅保留Customer标签，用属性存储国家

MERGE (c:Customer{phone: toInteger(Called_Phone)})
ON CREATE SET c.country = TP_country

MERGE (c2:Customer{phone: toInteger(OP_Phone)})
ON CREATE SET c2.country = OP_country

3. 增大事务批次，减少事务提交次数 #

原代码默认事务批次为1000行，建议调大到20000-50000行，减少事务日志写入次数：

CALL {...} IN TRANSACTIONS OF 20000 ROWS

4. 优化LOAD CSV批量读取 #

在LOAD CSV中添加ROWS PER BATCH参数，让Neo4j一次性读取更多行到内存，降低IO开销：

:auto LOAD CSV WITH HEADERS FROM 'file:///cdr.csv' AS line FIELDTERMINATOR ',' ROWS PER BATCH 10000

5. 简化Cypher逻辑，减少冗余变量传递 #

原代码中WITH传递了大量重复变量，优化后可减少不必要的变量流转，降低内存消耗。

6. 临时调整Neo4j配置，降低IO压力 #

导入期间可临时修改neo4j.conf的以下配置，完成后恢复：

• 设置事务日志周期性刷新：dbms.tx_log.flush.strategy=periodic
• 减少事务日志保留量：dbms.tx_log.rotation.retention_policy=100M size
• 关闭非必要监控：dbms.metrics.enabled=false

优化后的完整Cypher示例 #

-- 先执行创建索引
CREATE CONSTRAINT FOR (c:Customer) REQUIRE c.phone IS UNIQUE;

-- 导入数据
:auto LOAD CSV WITH HEADERS FROM 'file:///cdr.csv' AS line FIELDTERMINATOR ',' ROWS PER BATCH 10000
WITH line
WHERE line['Called_Number'] IS NOT NULL AND line['OP_ANI'] IS NOT NULL
WITH 
    toInteger(line['OP_ANI']) AS OP_Phone,
    CASE line['OP_Name'] WHEN 'TIC' THEN 'IRAN' ELSE 'Foreign' END AS OP_country,
    toInteger(line['Called_Number']) AS Called_Phone,
    CASE line['TP_Name'] WHEN 'TIC' THEN 'IRAN' ELSE 'Foreign' END AS TP_country,
    line['Setup_Time'] AS Setup_Time, 
    line['Duration'] AS Duration, 
    line['OP_Price'] AS OP_Price

CALL {
    WITH OP_Phone, OP_country, Called_Phone, TP_country, Setup_Time, Duration, OP_Price
    
    -- 处理被叫节点
    MERGE (c:Customer{phone: Called_Phone})
    ON CREATE SET c.country = TP_country
    
    -- 处理主叫节点
    MERGE (c2:Customer{phone: OP_Phone})
    ON CREATE SET c2.country = OP_country
    
    -- 创建通话关系
    MERGE (c2)-[r:CALLED{setupTime: Setup_Time, duration: Duration, OP_Price: OP_Price}]->(c)
} IN TRANSACTIONS OF 20000 ROWS

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