06-NoSQL数据库¶

from pymongo import MongoClient, ASCENDING, DESCENDING
from datetime import datetime

# 连接数据库
client = MongoClient('mongodb://admin:password@localhost:27017/')

# 选择数据库
db = client['mydb']

# 选择集合
collection = db['users']

# 插入文档
user = {
    'name': '张三',
    'age': 25,
    'email': 'zhangsan@example.com',
    'tags': ['AI', 'Python'],
    'created_at': datetime.now()
}
result = collection.insert_one(user)
print(f"插入ID: {result.inserted_id}")

# 插入多条
users = [
    {'name': '李四', 'age': 30},
    {'name': '王五', 'age': 28}
]
result = collection.insert_many(users)

# 查询单条
user = collection.find_one({'name': '张三'})
print(user)

# 查询多条
for user in collection.find({'age': {'$gt': 25}}):
    print(user)

# 投影
for user in collection.find({}, {'name': 1, 'email': 1, '_id': 0}):
    print(user)

# 排序
for user in collection.find().sort('age', DESCENDING):
    print(user)

# 分页
for user in collection.find().skip(10).limit(5):
    print(user)

# 更新
collection.update_one(
    {'name': '张三'},
    {'$set': {'age': 26}}
)

# 更新或插入（upsert）
collection.update_one(
    {'name': '赵六'},
    {'$set': {'age': 35}},
    upsert=True
)

# 删除
collection.delete_one({'name': '张三'})
collection.delete_many({'age': {'$lt': 25}})

# 聚合管道
pipeline = [
    {'$match': {'age': {'$gte': 25}}},
    {'$group': {'_id': '$age', 'count': {'$sum': 1}}},
    {'$sort': {'count': -1}}
]
for result in collection.aggregate(pipeline):
    print(result)

# 创建索引
collection.create_index([('email', ASCENDING)], unique=True)
collection.create_index([('name', TEXT)])

# 关闭连接
client.close()

from pymongo import MongoClient
from datetime import datetime

client = MongoClient('mongodb://localhost:27017/')
db = client['ml_platform']
experiments = db['experiments']

# 记录实验
def log_experiment(exp_name, config, metrics, artifacts=None):
    experiment = {
        'exp_name': exp_name,
        'config': config,  # 灵活的超参数配置
        'metrics': metrics,  # 训练指标
        'artifacts': artifacts or [],  # 模型文件、日志等
        'status': 'running',
        'created_at': datetime.now(),
        'updated_at': datetime.now()
    }
    return experiments.insert_one(experiment).inserted_id

# 更新实验指标
def update_metrics(exp_id, step, metrics):
    experiments.update_one(
        {'_id': exp_id},
        {
            '$push': {'history': {'step': step, **metrics}},
            '$set': {'metrics': metrics, 'updated_at': datetime.now()}
        }
    )

# 查询最佳实验
def get_best_experiments(model_type, metric='accuracy', top_k=5):
    return list(experiments.find(
        {'config.model_type': model_type}
    ).sort(
        f'metrics.{metric}', -1
    ).limit(top_k))

# 使用示例
exp_id = log_experiment(
    'resnet50_imagenet',
    {
        'model_type': 'ResNet50',
        'lr': 0.001,
        'batch_size': 32,
        'optimizer': 'Adam',
        'augmentation': {
            'random_crop': True,
            'horizontal_flip': True
        }
    },
    {'accuracy': 0, 'loss': 0}
)

# 训练过程中更新
for step in range(100):
    # ... 训练代码 ...
    update_metrics(exp_id, step, {
        'accuracy': 0.95,
        'loss': 0.1
    })

# 查询
best = get_best_experiments('ResNet50', 'accuracy', 3)
print(best)

import redis
import json
from datetime import timedelta

# 连接Redis
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True  # 自动解码为字符串
)

# 字符串操作
r.set('name', '张三')
print(r.get('name'))

# 设置过期时间
r.setex('temp_key', 60, 'value')  # 60秒过期
r.set('key', 'value', ex=3600)    # 1小时过期

# 序列化存储
data = {'model': 'ResNet50', 'accuracy': 0.95}
r.set('model:info', json.dumps(data))
info = json.loads(r.get('model:info'))

# 哈希操作
r.hset('user:1', mapping={
    'name': '张三',
    'age': 25,
    'email': 'zhangsan@example.com'
})
print(r.hget('user:1', 'name'))
print(r.hgetall('user:1'))

# 列表操作（队列）
r.lpush('tasks', 'task1', 'task2', 'task3')
task = r.rpop('tasks')
print(f"处理任务: {task}")

# 列表操作（栈）
r.lpush('stack', 'item1', 'item2')
item = r.lpop('stack')

# 集合操作
r.sadd('tags', 'AI', 'Python', 'ML')
print(r.smembers('tags'))
print(r.sismember('tags', 'AI'))

# 有序集合（排行榜）
r.zadd('leaderboard', {'player1': 100, 'player2': 200})
r.zincrby('leaderboard', 50, 'player1')
top_players = r.zrevrange('leaderboard', 0, 2, withscores=True)
print(top_players)

# 管道（批量操作）
pipe = r.pipeline()
for i in range(100):
    pipe.set(f'key:{i}', f'value:{i}')
pipe.execute()

# 发布订阅
# 发布者
r.publish('channel', 'Hello Subscribers!')

# 订阅者（在另一个线程/进程）
import signal

pubsub = r.pubsub()
pubsub.subscribe('channel')

def signal_handler(sig, frame):
    """信号处理函数，优雅退出"""
    pubsub.unsubscribe()
    print("\n取消订阅并退出")
    exit(0)

# 注册Ctrl+C信号处理
signal.signal(signal.SIGINT, signal_handler)

try:  # try/except捕获异常
    for message in pubsub.listen():
        if message['type'] == 'message':
            print(f"收到消息: {message['data']}")
except Exception as e:
    print(f"订阅出错: {e}")
finally:
    pubsub.close()  # 确保连接被关闭

import redis
import hashlib
import json
import time

class ModelCache:
    def __init__(self, redis_host='localhost'):
        self.r = redis.Redis(host=redis_host, port=6379, decode_responses=True)
        self.default_ttl = 3600  # 1小时

    def _generate_key(self, model_name, input_data):
        """生成缓存键"""
        data_str = json.dumps(input_data, sort_keys=True)  # json.dumps将Python对象转为JSON字符串
        hash_val = hashlib.md5(f"{model_name}:{data_str}".encode()).hexdigest()
        return f"pred:{model_name}:{hash_val}"

    def get_prediction(self, model_name, input_data):
        """获取缓存的预测结果"""
        key = self._generate_key(model_name, input_data)
        cached = self.r.get(key)
        if cached:
            return json.loads(cached)  # json.loads将JSON字符串转为Python对象
        return None

    def set_prediction(self, model_name, input_data, prediction, ttl=None):
        """缓存预测结果"""
        key = self._generate_key(model_name, input_data)
        value = json.dumps({
            'prediction': prediction,
            'cached_at': time.time()
        })
        self.r.setex(key, ttl or self.default_ttl, value)

    def predict_with_cache(self, model, model_name, input_data):
        """带缓存的预测"""
        # 先查缓存
        cached = self.get_prediction(model_name, input_data)
        if cached:
            print("缓存命中")
            return cached['prediction']

        # 执行预测
        print("执行预测")
        prediction = model.predict(input_data)

        # 存入缓存
        self.set_prediction(model_name, input_data, prediction)

        return prediction

# 特征缓存
class FeatureCache:
    def __init__(self, redis_host='localhost'):
        self.r = redis.Redis(host=redis_host, port=6379, decode_responses=True)

    def get_features(self, entity_type, entity_id, feature_names):
        """获取实体特征"""
        key = f"features:{entity_type}:{entity_id}"
        features = self.r.hmget(key, feature_names)
        return {k: float(v) if v else 0 for k, v in zip(feature_names, features)}  # zip并行遍历多个可迭代对象

    def set_features(self, entity_type, entity_id, features, ttl=300):
        """设置实体特征"""
        key = f"features:{entity_type}:{entity_id}"
        self.r.hset(key, mapping=features)
        self.r.expire(key, ttl)

# 使用示例
cache = ModelCache()

# 模拟模型
class DummyModel:
    def predict(self, data):
        time.sleep(1)  # 模拟耗时
        return {'class': 'cat', 'confidence': 0.95}

model = DummyModel()

# 第一次预测（会执行模型）
result1 = cache.predict_with_cache(model, 'classifier', {'image': 'cat.jpg'})
print(result1)

# 第二次预测（命中缓存，瞬间返回）
result2 = cache.predict_with_cache(model, 'classifier', {'image': 'cat.jpg'})
print(result2)

# Pinecone示例 (v7.0+ API)
from pinecone import Pinecone, ServerlessSpec

# 初始化（v3.0+ 新方式）
pc = Pinecone(api_key="your-api-key")

# 创建索引（如果尚未创建）
# v3.0+ 使用 list_indexes() 返回 IndexList 对象
existing_indexes = [idx.name for idx in pc.list_indexes()]
if "semantic-search" not in existing_indexes:
    pc.create_index(
        name="semantic-search",
        dimension=1536,  # OpenAI text-embedding-ada-002 维度
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

# 连接索引
index = pc.Index("semantic-search")

# 插入向量（upsert）
index.upsert([
    ("id1", [0.1, 0.2, ...], {"text": "文档1", "source": "web"}),
    ("id2", [0.3, 0.4, ...], {"text": "文档2", "source": "pdf"})
])

# 相似度搜索
results = index.query(
    vector=[0.1, 0.2, ...],
    top_k=5,
    include_metadata=True
)

# 输出结果
for match in results.matches:
    print(f"ID: {match.id}, Score: {match.score}, Metadata: {match.metadata}")