为何开发者：正在抛弃PostgreSQL、MySQL 和 MongoDB

像PostgreSQL和MySQL这类传统关系型数据库，以及MongoDB这样的文档存储数据库，长期以来一直是众多开发者的首选解决方案。然而，随着数据需求变得日益多样化和复杂，新一代的专用数据库应运而生，以应对特定的挑战。在本文中，我将揭示那些针对不同用例具备独特功能的替代数据库，并对它们的特性进行技术层面的探讨，同时提供实用的Python代码示例。

Milvus：面向人工智能和机器学习的可扩展向量数据库

Milvus是一款开源向量数据库，旨在为大规模的人工智能应用和相似性搜索提供支持。它在涉及机器学习、计算机视觉、自然语言处理以及推荐系统等应用中特别有用。

关键特性：

• 高效的相似性搜索：实现了多种索引算法，以便在高维空间中快速进行最近邻搜索。
• 可扩展性：支持分布式部署，可处理大规模向量数据。
• 混合搜索：将向量相似性搜索与标量筛选相结合，以处理复杂查询。
• CRUD操作：支持对向量数据进行创建、读取、更新和删除操作。
• 数据一致性：确保分布式环境中数据的强一致性。

适用场景：

• 图像和视频搜索
• 推荐系统
• 自然语言处理应用
• 欺诈检测

Python示例：

from pymilvus importCollection,CollectionSchema,DataType,FieldSchema, connections

defmain()->None:
# 连接到Milvus服务器
    connections.connect()

# 为集合定义模式
    fields =[
FieldSchema(name="id", dtype=DataType.INT64, is_primary=True),
FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=128),
]
    schema =CollectionSchema(fields,"Vector collection example")

# 创建一个集合
    collection =Collection("example_collection", schema)

# 插入数据（随机示例）
import numpy as np
    vectors = np.random.random([1000,128]).astype(np.float32)
    ids =list(range(1000))
    collection.insert([ids, vectors])

# 查询数据
    collection.load()
    results = collection.search(
        vectors[:1],
"vector",
        param={"metric_type":"L2","params":{"nprobe":10}},
        limit=3,
)
for hit in results[0]:
print(f"Hit ID: {hit.id}, Distance: {hit.distance}")

if __name__ =="__main__":
    main()

这个示例展示了如何在Milvus中创建一个集合、插入向量数据以及执行相似性搜索。

InfluxDB：掌控时间序列数据

InfluxDB是一款专门为高性能处理带时间戳的数据而构建的时间序列数据库。它在诸如物联网（IoT）应用监控和实时分析等场景中表现出色。

关键特性：

• 时间结构合并树（TSM）：InfluxDB使用一种针对时间序列数据进行优化的自定义存储引擎，可实现高效压缩和快速查询。
• Flux查询语言：一种专为时间序列数据分析设计的强大的函数式数据脚本语言。
• 连续查询：自动计算聚合数据，以优化频繁使用的查询。
• 保留策略：自动的数据管理策略，用于对旧数据进行降采样或使其过期。

适用场景：

• 实时分析应用
• 物联网数据监控
• 系统监控与指标收集

Python示例：

from datetime import datetime
import numpy as np
from influxdb_client importInfluxDBClient,Point,Task,WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS

# 设置InfluxDB客户端（请替换为你自己的令牌、组织和存储桶详细信息）
token ="your_token"
org ="your_org"
bucket ="example_bucket"
client =InfluxDBClient(url="http://localhost:8086", token=token)

# 示例1：向InfluxDB 2写入时间序列数据
write_api = client.write_api(write_options=SYNCHRONOUS)
# 模拟一些数据点
for i inrange(100):
    point =(
Point("cpu_load")
.tag("host","server01")
.field("value", np.random.uniform(0.5,1.5))
.time(datetime.now(),WritePrecision.NS)
)
    write_api.write(bucket=bucket, org=org, record=point)

# 示例2：使用Flux查询——计算移动平均值
query_api = client.query_api()
flux_query =f"""
from(bucket: "{bucket}")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "cpu_load")
  |> filter(fn: (r) => r._field == "value")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
  |> yield(name: "mean")
"""
tables = query_api.query(flux_query)
for table in tables:
for record in table.records:
print(
f"Time: {record.get_time()}, 5-Minute Moving Average: {record.get_value()}"
)

# 示例3：任务自动化——持续数据处理
# 创建一个每分钟自动计算并存储滚动平均值的任务
task =Task(
    name="cpu_load_moving_avg",
    flux=f"""
    option task = {{
        name: "cpu_load_moving_avg",
        every: 1m
    }}
    from(bucket: "{bucket}")
      |> range(start: -1m)
      |> filter(fn: (r) => r._measurement == "cpu_load")
      |> aggregateWindow(every: 1m, fn: mean)
      |> to(bucket: "{bucket}", org: "{org}")
    """,
)
tasks_api = client.tasks_api()
tasks_api.create_task(task=task)
print("任务已创建，将每分钟计算并存储滚动平均值。")

这个示例展示了以下内容：

• 连接到InfluxDB并设置写入API。
• 生成并写入带有多个字段和标签的模拟传感器数据。
• 使用Flux查询数据，计算过去一小时的平均值。
• 处理并展示查询结果。

Neo4j：解析复杂关系

Neo4j是一种图数据库，它使用节点、边和属性来表示和存储数据。这种结构非常适用于实体间关系至关重要的应用，比如社交网络或推荐系统。

关键特性：

• 属性图模型：将数据表示为节点、关系和属性，支持丰富的数据建模。
• Cypher查询语言：一种专为处理图数据设计的声明式查询语言。
• ACID事务：即使在高度关联的数据场景中，也能确保数据完整性。
• 原生图存储：针对关系遍历进行了优化，相比于关系型数据库，处理复杂查询的速度更快。

适用场景：

• 社交网络分析
• 推荐引擎
• 欺诈检测系统
• 知识图谱

Python示例：

from neo4j importGraphDatabase,ManagedTransaction

URI ="bolt://localhost:7687"
DRIVER =GraphDatabase.driver(URI, auth=("neo4j","password"))


defadd_friend(tx:ManagedTransaction, name:str, friend_name:str)->None:
    tx.run(
"CREATE (a:Person {name: $name})-[:FRIEND]->(b:Person {name: $friend_name})",
        name=name,
        friend_name=friend_name,
)


defmain()->None:
with DRIVER.session()as session:
        session.write_transaction(add_friend,"Alice","Bob")

with DRIVER.session()as session:
        result = session.run(
"MATCH (a:Person)-[:FRIEND]->(b:Person) RETURN a.name, b.name"
)
for record in result:
print(f"{record['a.name']} is friends with {record['b.name']}")

    DRIVER.close()


if __name__ =="__main__":
    main()

这个示例展示了以下内容：

• 创建一个可复用的Neo4j连接类。
• 向图中添加节点（人员）和关系（朋友关系）。
• 执行一个更复杂的查询来查找“朋友的朋友”——在关系型数据库中，这会是一项更复杂的任务。

DuckDB：闪电般速度的分析查询

DuckDB是一款轻量级的内存分析数据库，常被描述为“用于分析的SQLite”。它专为快速分析查询而设计，尤其适用于数据科学工作流程，并且能够与pandas DataFrames无缝集成。

关键特性：

• 向量化查询执行：按批次处理数据，而非逐行处理，这极大地提升了性能。
• 列存储：针对通常只涉及部分列的分析查询进行了优化。
• 零拷贝数据摄取：可以直接从诸如Pandas DataFrames这样的数据源查询数据，无需进行拷贝。
• ACID事务：确保数据的一致性和可靠性。

适用场景：

• 数据分析与探索
• ETL（抽取、转换、加载）流程
• Python应用中的嵌入式分析

Python示例：

import duckdb
import numpy as np
import pandas as pd

# 示例1：直接从CSV文件查询数据，无需加载到内存
duckdb.execute("""
CREATE TABLE employees AS SELECT * FROM 'employees.csv';
""")
result = duckdb.execute("SELECT COUNT(*) FROM employees").fetchall()
print(f"员工总数：{result[0][0]}")

# 示例2：将DuckDB与Pandas DataFrame集成
df = pd.DataFrame(
{
"employee_id":range(1,101),
"name":[f"Employee {i}"for i inrange(1,101)],
"salary": np.random.randint(50000,150000, size=100),
}
)

# 从DataFrame创建一个DuckDB表
duckdb.execute("CREATE TABLE df_employees AS SELECT * FROM df")

# 直接在DataFrame上执行查询
high_earners = duckdb.query_df(df,"df","SELECT * FROM df WHERE salary > 100000").df()
print("高收入者：")
print(high_earners)

# 示例3：使用窗口函数进行分析查询
window_query ="""
SELECT
    employee_id,
    salary,
    AVG(salary) OVER () AS avg_salary,
    RANK() OVER (ORDER BY salary DESC) AS rank
FROM
    df_employees
"""

ranked_salaries = duckdb.execute(window_query).df()
print("带有平均工资的薪资排名：")
print(ranked_salaries)

# 示例4：表连接与复杂聚合
department_df = pd.DataFrame(
{
"employee_id":range(1,101),
"department": np.random.choice(
["工程","人力资源","销售","市场营销"], size=100
),
}
)

duckdb.execute("CREATE TABLE departments AS SELECT * FROM department_df")

# 执行连接和聚合操作
aggregation_query ="""
SELECT
    d.department,
    COUNT(e.employee_id) AS num_employees,
    AVG(e.salary) AS avg_salary
FROM
    df_employees e
JOIN
    departments d
ON
    e.employee_id = d.employee_id
GROUP BY
    d.department
"""

department_stats = duckdb.execute(aggregation_query).df()
print("部门统计信息：")
print(department_stats)

这个示例展示了以下内容：

• 创建一个内存中的DuckDB数据库，并将一个Pandas DataFrame注册为表。
• 执行一个复杂的分析查询，包括：每日聚合、移动平均、用于日环比比较的滞后计算。
• 对数据集进行整体统计计算。

Redis：高性能数据结构

Redis是一种快速的内存数据结构存储，可用作数据库、缓存和消息代理。与传统数据库不同，Redis完全在内存中运行，这使得其读写操作速度极快。

关键特性：

• 内存数据存储：提供极快的读写操作。
• 数据结构支持：支持字符串、哈希、列表、集合、有序集合、位图、基数估计（HyperLogLog）以及地理空间索引等数据结构。
• 发布/订阅（Pub/Sub）消息机制：内置的发布/订阅消息范式，用于构建实时应用。
• Lua脚本：允许在服务器端原子性地执行复杂操作。

适用场景：

• 用于性能提升的缓存层
• 实时分析
• 会话管理
• 排行榜与计数

Python示例：

import subprocess
import threading
import time

import redis

typeRedisProcess= subprocess.Popen[bytes]

# 用于控制监听器线程的标志
stop_listening = threading.Event()


# 在本地启动Redis服务器（这将作为一个子进程运行）
defstart_redis_server()->RedisProcess:
    process = subprocess.Popen(
["redis-server"], stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
    time.sleep(0.5)# 给Redis服务器一秒钟启动时间
return process


# 停止Redis服务器
defstop_redis_server(process:RedisProcess)->None:
    process.terminate()
    process.wait()


defreal_time_data(r: redis.Redis)->None:
    r.set("page_views",0)
    r.incr("page_views")
    r.incr("page_views")
    page_views = r.get("page_views").decode("utf-8")
print(f"页面浏览量：{page_views}")


deftask_queue(r: redis.Redis)->None:
    r.rpush("task_queue","task1")
    r.rpush("task_queue","task2")
    r.rpush("task_queue","task3")

    task = r.lpop("task_queue").decode("utf-8")
print(f"正在处理任务：{task}")


defredis_set(r: redis.Redis)->None:
    r.sadd("unique_users","user1")
    r.sadd("unique_users","user2")
    r.sadd("unique_users","user1")# 重复项，不会被添加

    unique_users = r.smembers("unique_users")
print(f"唯一用户：{[user.decode('utf-8') for user in unique_users]}")


defexpiring_data(r: redis.Redis)->None:
    r.set("session_token","abc123", ex=3600)# 一小时后过期
    session_token = r.get("session_token").decode("utf-8")
print(f"会话令牌：{session_token}（将在一小时后过期）")


defpub_sub_messaging(r: redis.Redis)->None:
defmessage_handler(message):
print(f"收到消息：{message['data'].decode('utf-8')}")

deflisten_for_messages():
        p = r.pubsub()
        p.subscribe("my-channel")
# 监听消息，设置超时时间以便定期检查停止信号
whilenot stop_listening.is_set():
            message = p.get_message(timeout=1.0)# 超时时间设为1秒
if message and message["type"]=="message":
                message_handler(message)

# 启动一个线程来监听消息
    listener_thread = threading.Thread(target=listen_for_messages)
    listener_thread.start()

# 给监听器一点启动时间
    time.sleep(0.5)

# 模拟向频道发送一条消息
    r.publish("my-channel","Hello, Redis!")

# 留出一些时间让消息得到处理
    time.sleep(0.5)

# 发出信号让监听器线程停止
    stop_listening.set()
    listener_thread.join()# 等待监听器线程结束


defmain()->None:
# 启动Redis服务器
    redis_process = start_redis_server()

# 连接到本地启动的Redis服务器
    r = redis.Redis(host="localhost", port=6379, db=0)

# 示例1：实时数据——将Redis用作计数器
    real_time_data(r)

# 示例2：使用Redis列表实现实时任务队列
    task_queue(r)

# 示例3：使用Redis集合跟踪唯一项目
    redis_set(r)

# 示例4：处理过期数据
    expiring_data(r)

# 示例5：使用Redis的发布/订阅消息机制
    pub_sub_messaging(r)

# 停止Redis服务器
    stop_redis_server(redis_process)
print("Redis服务器已停止。")


if __name__ =="__main__":
    main()

这个示例展示了Redis的各种特性，包括计数器、任务队列、用于唯一项目的集合、过期数据以及发布/订阅消息机制。

Tile38：地理空间数据库与地理围栏服务器

Tile38是一款专门的地理空间数据库，允许你实时存储、查询和分析地理空间数据。它非常适用于诸如车队管理、资产追踪以及基于位置的服务等应用。

关键特性：

• 地理空间索引：高效地对空间数据进行索引和查询。
• 实时地理围栏：当物体进入或离开定义的区域时，支持即时通知。
• 多种对象类型：可处理点、边界框、XYZ瓦片、GeoJSON等。
• 脚本支持：支持Lua脚本用于复杂的地理空间操作。
• 发布/订阅（Pub/Sub）模型：允许对空间事件进行实时通知。

适用场景：

• 车队管理系统
• 基于位置的服务
• 资产追踪
• 地理围栏应用

Python示例：

import threading
import time
from pyle38 importTile38

# 连接到Tile38
tile38 =Tile38(url="redis://localhost:9851")# Tile38使用Redis协议

# 示例1：向Tile38添加位置（物体）信息
asyncdefadd_locations():
await tile38.set("fleet").point("truck1",33.5123,-112.2693).exec()
await tile38.set("fleet").point("truck2",33.5011,-112.1710).exec()
await tile38.set("fleet").point("truck3",40.7128,-74.0010).exec()

# 示例2：创建地理围栏
asyncdefcreate_geofence():
# 在特定区域（例如配送区域）周围创建地理围栏
await(
        tile38.within("fleet")
.circle(33.5,-112.2,5000)# 半径为5000米
.detect("enter","exit")
.object("delivery_zone")
)

# 示例3：实时地理围栏监控
defmonitor_geofence():
    pubsub = tile38.pubsub()
    pubsub.subscribe("delivery_zone")
for message in pubsub.listen():
print(f"地理围栏警报：{message}")

# 启动一个线程来监控地理围栏警报
monitor_thread = threading.Thread(target=monitor_geofence)
monitor_thread.start()

asyncdefmain():
# 添加位置信息
await add_locations()
# 创建地理围栏
await create_geofence()

# 示例4：模拟物体移入地理围栏
    time.sleep(1)# 给监控器一点启动时间
await tile38.set("fleet","truck4").point(33.4018,-112.8965).exec()# 在地理围栏内

# 留出时间以便检测事件
    time.sleep(5)

# 清理：停止监控线程
    monitor_thread.join()

if __name__ =="__main__":
import asyncio
    asyncio.run(main())

这个示例展示了如何向Tile38添加地理空间对象、创建地理围栏、查询地理围栏内的对象以及为地理围栏活动设置实时通知。

结论

正如我们所探讨的，这些替代数据库各自都具备独特的功能：

InfluxDB擅长处理时间序列数据，使其非常适用于物联网和监控应用。Neo4j在处理高度关联的数据时表现出色，对于社交网络和推荐系统来说堪称完美。DuckDB能提供闪电般快速的分析查询，对数据科学家和分析师特别有用。Redis提供高性能的内存数据结构，非常适合缓存和实时应用。Milvus专注于向量相似性搜索，对人工智能和机器学习应用至关重要。Tile38侧重于地理空间数据和实时地理围栏，对于基于位置的服务必不可少。

虽然这些专用数据库具备强大的功能，但必须记住它们也会给你的架构带来额外的复杂性。每引入一个新数据库，在托管、管理、备份和安全等方面都会增加开销。

在采用这些替代方案之前，请考虑以下几点：

• 从简入手：不要一开始就使你的架构过于复杂。像PostgreSQL或MongoDB这样的传统数据库能够有效应对许多用例。
• 评估需求：仔细评估你的项目是否真的需要这些数据库所提供的专用功能。
• 考虑扩展性：思考随着项目的发展，你的数据和查询需求可能会如何演变。
• 维护开销：将维护和保障额外数据库系统所需的时间和资源纳入考量范围。
• 云解决方案：探索云服务提供商提供的托管数据库服务，这可以简化部署和维护工作。

请记住，目标是在性能和可维护性之间取得平衡。有时候，使用知名数据库的简单解决方案要优于包含多个专用数据库的复杂系统。