RedisGraph深度解析：当图数据库遇上内存引擎

// 创建账户实体节点（含属性索引）
GRAPH.QUERY Financial "
CREATE INDEX ON :Account(id)
CREATE (:Account {id: 'A1', risk_level:3, balance:50000}),
       (:Account {id: 'A2', risk_level:1, balance:150000}),
       (:Account {id: 'A3', risk_level:5, balance:2000})
"
// 创建交易关系（含时序和金额属性）
GRAPH.QUERY Financial "
MATCH (a:Account {id:'A1'}), (b:Account {id:'A2'}) 
CREATE (a)-[:TRANSFER {timestamp: 1628000000, amount: 5000, currency:'USD'}]->(b)
"

GRAPH.QUERY Financial "
MATCH (a1)-[t1:TRANSFER]->(a2),
      (a2)-[t2:TRANSFER]->(a3),
      (a3)-[t3:TRANSFER]->(a1)
WHERE t1.timestamp < t2.timestamp < t3.timestamp
  AND t1.amount + t2.amount + t3.amount > 100000
RETURN a1.id, a2.id, a3.id 
ORDER BY t3.timestamp DESC
LIMIT 10
"

GRAPH.EXPLAIN Financial "
MATCH (a:Account)-[t:TRANSFER]->()
WHERE a.risk_level > 4
  AND t.amount > 50000
RETURN COUNT(*)
"
/* 输出：
1. Project
2. Aggregate
3. Filter
4. Conditional Traverse
5. Node By Label Scan
*/

GRAPH.EXPLAIN Financial "
MATCH (a:Account)-[t:TRANSFER]->()
WHERE t.amount > 50000
  AND a.risk_level > 4
RETURN COUNT(*)
"
/* 优化后计划：
1. Project  
2. Aggregate
3. Conditional Traverse
4. Filter
5. Node By Label Scan
*/

GRAPH.QUERY Financial "CREATE INDEX FOR (a:Account)-[t:TRANSFER]->(b:Account) 
                      ON (a.risk_level, t.amount, b.risk_level)"

| 数据规模 | 无索引(ms) | 复合索引(ms) |
|----------|------------|--------------|
| 10万交易 | 1200       | 45           |
| 百万交易 | 超时        | 220          |

import redis
from redisgraph import Graph
r = redis.Redis()
graph = Graph('Financial', r)
# 创建流消费者组
r.xgroup_create('transfers', 'fraud_team', id='0', mkstream=True)
# 实时处理交易流
while True:
    transfers = r.xreadgroup('fraud_team', 'consumer1', 
                           {'transfers': '>'}, count=1)
    
    for transfer in transfers[0][1]:
        _, data = transfer
        graph.query("""
            MATCH (a {id: $from}), (b {id: $to})
            CREATE (a)-[:TRANSFER {
                amount: $amt, 
                timestamp: $ts
            }]->(b)
        """, params=data)
        
        # 实时检测规则
        graph.query("""
            MATCH (a)-[t1]->(b)-[t2]->(c)-[t3]->(a)
            WHERE t1.timestamp > now()-3600
              AND t1.amount + t2.amount + t3.amount > 200000
            RETURN a.id, path
        """)

# redis.conf
graph.maxmemory 10GB
graph.memory_allocator jemalloc
graph.cache_threshold 0.3
graph.persistence_mode hybrid

| 数据访问模式 | 纯内存(ops/sec) | 混合存储(ops/sec) |
|--------------|-----------------|-------------------|
| 全热数据     | 45000           | 42000             |
| 30%冷数据    | 45000           | 39000             | 
| 70%冷数据    | 45000           | 21000             |

# 节点配置
redis-server --port 7000 --cluster-enabled yes
redis-server --port 7001 --cluster-enabled yes
redis-server --port 7002 --cluster-enabled yes
# 数据分片策略
redis-cli --cluster create 127.0.0.1:7000 127.0.0.1:7001 127.0.0.1:7002

GRAPH.QUERY Financial "MATCH ()-[r:TRANSFER]->() RETURN count(r)"
/* 
分片前：单节点1亿数据查询耗时12.3秒
分片后：3节点集群查询耗时4.1秒 
*/