深入解析NoSQL数据库：从文档存储到图数据库的全场景实践

-- 物联网设备数据表设计
CREATE TABLE iot.sensor_data (
  device_id text,
  bucket timestamp,  -- 按天分桶
  event_time timestamp,
  temperature float,
  humidity float,
  PRIMARY KEY ((device_id, bucket), event_time)
) WITH CLUSTERING ORDER BY (event_time DESC)
  AND compaction = { 
    'class' : 'TimeWindowCompactionStrategy',
    'compaction_window_unit' : 'DAYS',
    'compaction_window_size' : 1 
  };

// 发现资金环状转移
MATCH path=(a:Account)-[t:TRANSFER*3..5]->(a)
WHERE ALL(r IN relationships(path) WHERE r.amount > 10000)
WITH nodes(path) AS accounts, relationships(path) AS transfers
RETURN accounts, 
       sum(t.amount) AS totalAmount
ORDER BY totalAmount DESC
LIMIT 10;

// 基于协同过滤的推荐
MATCH (u:User {id: "1001"})-[:PURCHASED]->(i:Item)<-[:PURCHASED]-(similar:User)
WITH u, similar, COUNT(i) AS commonItems 
ORDER BY commonItems DESC LIMIT 10
MATCH (similar)-[:PURCHASED]->(rec:Item)
WHERE NOT EXISTS((u)-[:PURCHASED]->(rec))
RETURN rec.id AS recommendation, COUNT(*) AS score
ORDER BY score DESC LIMIT 5;

# AWS DMS跨云迁移配置
resource "aws_dms_endpoint" "cosmosdb" {
  endpoint_id   = "cosmos-target"
  endpoint_type = "target"
  engine_name   = "cosmosdb"
  cosmosdb_settings {
    service_access_key = var.cosmos_key
    database_name      = "migration_db"
  }
}

resource "aws_dms_replication_task" "mongo_to_cosmos" {
  migration_type           = "full-load-and-cdc"
  replication_task_id      = "mongo2cosmos"
  replication_instance_arn = aws_dms_replication_instance.main.arn
  source_endpoint_arn      = aws_dms_endpoint.mongo.arn
  target_endpoint_arn      = aws_dms_endpoint.cosmosdb.arn
  table_mappings           = jsonencode({
    "rules": [{
      "rule-type": "selection",
      "rule-id": "1",
      "object-locator": { "schema": "shop", "table": "%" }
    }]
  })
}

# redis.conf核心配置
save 900 1          # 15分钟至少1次修改则快照
save 300 10         # 5分钟至少10次修改
appendonly yes      # 启用AOF
appendfsync everysec # 每秒刷盘
aof-use-rdb-preamble yes # 混合持久化格式

CREATE KEYSPACE global_data 
WITH replication = {
  'class': 'NetworkTopologyStrategy',
  'DC_NYC': 3,
  'DC_LDN': 2,
  'DC_TKO': 2
};

ALTER KEYSPACE system_auth 
WITH replication = {
  'class': 'NetworkTopologyStrategy',
  'DC_NYC': 3,
  'DC_LDN': 3
};

CREATE TABLE sensor_readings (
  device_id text,
  timestamp bigint,
  values map<text, float>,
  PRIMARY KEY (device_id, timestamp)
) WITH compaction = {
  'class': 'TimeWindowCompactionStrategy',
  'compaction_window_unit': 'DAYS',
  'compaction_window_size': 1
}
AND compression = {
  'sstable_compression': 'ZstdCompressor',
  'chunk_length_kb': 64
};

CALL gds.graph.project(
  'social_graph',
  'User',
  { FOLLOWS: { orientation: 'UNDIRECTED' } }
)

CALL gds.louvain.stream('social_graph')
YIELD nodeId, communityId
RETURN gds.util.asNode(nodeId).id AS user, communityId
ORDER BY communityId, user

MATCH (start:Warehouse {id: 'W1'}), (end:Store {id: 'S5'})
CALL gds.shortestPath.dijkstra.stream('logistics_network', {
  sourceNode: start,
  targetNode: end,
  relationshipWeightProperty: 'travel_time'
})
YIELD index, sourceNode, targetNode, totalCost, path
RETURN totalCost AS minutes,
       nodes(path) AS route
ORDER BY totalCost ASC
LIMIT 3