数据库选型决策框架：PostgreSQL vs MongoDB vs ClickHouse的场景分析与成本收益对比

创业团队技术选型中最容易翻车的环节，不在代码层面，而在数据库决策。一张错误的选型单，前期加班三个月，后期还债三年。

本文抛开品牌偏好，从场景适配、性能基准、成本模型三个维度，建立一套可复用的数据库选型决策框架。

一、三个数据库的核心定位

讨论选型前，先对齐三者的本质差异。

PostgreSQL 是关系型数据库的工程标杆。ACID 事务、多表 JOIN、窗口函数、JSONB 索引、全文搜索、地理空间扩展，组合在一起覆盖绝大多数 OLTP 场景。它的代价是复杂查询下的优化难度和水平扩展门槛。

MongoDB 是文档数据库的事实标准。Schema-less 使其天然适配快迭代的业务，聚合管道和变更流让它在日志分析和实时推送场景也有用武之地。代价是事务支持晚于 PG，多文档事务性能不及关系型，查询模式声明性较弱。

ClickHouse 是列式分析引擎的天花板。十亿行级别的扫描秒级返回，物化视图、工程函数和压缩比让它在实时数仓场景几乎没有对手。代价是不支持事务，不适合频繁 UPDATE/DELETE，连接语义与传统关系型完全不同。

概括一句话：PG 做业务系统，Mongo 做内容系统，ClickHouse 做分析系统。混淆错配就是成本的源头。

二、场景决策树：什么场景用什么库

选型不能凭直觉，需要一个结构化判断流程。以下决策树覆盖创业团队最常见的五种写入与查询模式。

flowchart TD A[数据写入与查询模式] --> B{需要ACID事务？} B -->|是| C{数据结构稳定？} C -->|是| PG[PostgreSQL] C -->|否| M[PostgreSQL + JSONB] B -->|否| D{写入模式} D -->|高频追加/批量| E{查询模式} E -->|聚合分析为主| CK[ClickHouse] E -->|多样化灵活查询| MO[MongoDB] D -->|低延迟单条读写| F{Schema固定？} F -->|是| PG F -->|否| MO

判断的核心变量有三个：事务需求、Schema 稳定性、查询模式。按这个顺序排查，大部分选型争论可以收敛。

还有一个常见陷阱：把 ClickHouse 当 OLTP 用。如果业务有大量 UPDATE 或 DELETE 操作，ClickHouse 的合并树机制会让写入放大严重。反过来，在 PG 上做十亿级大表聚合查询，抛开分区和物化视图硬跑，也是自找麻烦。

三、性能基准：用数据说话

光看文档和博客不够，需要用基准测试验证假设。以下代码基于 Python 实现，覆盖插入、范围查询、聚合分析三个核心场景，包含完整的错误处理和结果输出。

""" 数据库性能基准测试框架 测试对象：PostgreSQL、MongoDB、ClickHouse 测试场景：批量写入、范围查询、聚合分析 """ import time import json import random import uuid from abc import ABC, abstractmethod from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime, timedelta from functools import wraps from typing import Any, Callable, Generator, List, Optional import psycopg2 import psycopg2.errors import pymongo import pymongo.errors from pymongo import IndexModel, ASCENDING from clickhouse_driver import Client as CHClient from clickhouse_driver.errors import Error as CHError @dataclass class BenchmarkResult: """单次基准测试结果""" operation: str row_count: int elapsed_ms: float throughput: float # ops/sec success: bool error_message: Optional[str] = None @dataclass class BenchmarkReport: """完整基准测试报告""" db_type: str results: List[BenchmarkResult] = field(default_factory=list) def add(self, result: BenchmarkResult) -> None: self.results.append(result) def summary(self) -> str: lines = [f"\n{'='*60}", f" {self.db_type} 基准测试报告", f"{'='*60}"] for r in self.results: status = "PASS" if r.success else "FAIL" lines.append( f" {r.operation:<20s} | " f"行数: {r.row_count:>8d} | " f"耗时: {r.elapsed_ms:>8.1f}ms | " f"吞吐: {r.throughput:>10.1f} ops/s | " f"状态: {status}" ) if r.error_message: lines.append(f" 错误: {r.error_message}") lines.append(f"{'='*60}\n") return "\n".join(lines) def retry_on_error( max_retries: int = 3, delay_sec: float = 1.0, backoff: float = 2.0, ): """数据库操作重试装饰器""" def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args: Any, **kwargs: Any) -> Any: last_error: Optional[Exception] = None current_delay = delay_sec for attempt in range(max_retries + 1): try: return func(*args, **kwargs) except ( psycopg2.OperationalError, psycopg2.errors.SerializationFailure, pymongo.errors.ConnectionFailure, pymongo.errors.ServerSelectionTimeoutError, CHError, ConnectionError, TimeoutError, ) as e: last_error = e if attempt < max_retries: time.sleep(current_delay) current_delay *= backoff raise RuntimeError( f"操作失败，已重试 {max_retries} 次: {last_error}" ) return wrapper return decorator def measure_time(func: Callable) -> Callable: """计时装饰器""" @wraps(func) def wrapper(*args: Any, **kwargs: Any) -> tuple: start = time.perf_counter() result = func(*args, **kwargs) elapsed = (time.perf_counter() - start) * 1000 return result, elapsed return wrapper class BaseBenchmark(ABC): """基准测试基类""" def __init__(self, conn_config: dict): self.conn_config = conn_config self.db_type = self.__class__.__name__.replace("Benchmark", "") self.report = BenchmarkReport(db_type=self.db_type) self.row_count = 100_000 @abstractmethod def connect(self) -> Any: """建立数据库连接""" pass @abstractmethod def prepare_table(self) -> None: """准备测试表/集合""" pass @abstractmethod def cleanup(self) -> None: """清理测试数据""" pass def generate_rows(self, count: int) -> List[dict]: """生成测试数据""" base_time = datetime.now() return [ { "event_id": str(uuid.uuid4()), "user_id": random.randint(1, 10_000), "event_type": random.choice(["click", "view", "purchase", "share"]), "value": round(random.uniform(0.01, 999.99), 2), "created_at": ( base_time + timedelta(seconds=random.randint(0, 86400 * 30)) ), } for _ in range(count) ] def run(self) -> BenchmarkReport: """执行完整基准测试流程""" try: self.prepare_table() except Exception as e: self.report.add(BenchmarkResult( operation="prepare", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=f"准备表失败: {e}", )) return self.report for name, method in [ ("批量写入", "benchmark_insert"), ("范围查询", "benchmark_range_query"), ("聚合分析", "benchmark_aggregate"), ("点查询", "benchmark_point_query"), ]: try: func = getattr(self, method) r = func() self.report.add(r) except Exception as e: self.report.add(BenchmarkResult( operation=name, row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=f"{type(e).__name__}: {e}", )) try: self.cleanup() except Exception: pass return self.report @abstractmethod def benchmark_insert(self) -> BenchmarkResult: pass @abstractmethod def benchmark_range_query(self) -> BenchmarkResult: pass @abstractmethod def benchmark_aggregate(self) -> BenchmarkResult: pass @abstractmethod def benchmark_point_query(self) -> BenchmarkResult: pass class PostgreSQLBenchmark(BaseBenchmark): """PostgreSQL 基准测试实现""" def __init__(self, conn_config: dict): super().__init__(conn_config) self.conn = None self.table_name = "bench_events" @retry_on_error(max_retries=3) def connect(self) -> Any: self.conn = psycopg2.connect( host=self.conn_config.get("host", "localhost"), port=self.conn_config.get("port", 5432), dbname=self.conn_config.get("database", "benchmark"), user=self.conn_config.get("user", "postgres"), password=self.conn_config.get("password", ""), connect_timeout=10, ) self.conn.autocommit = False return self.conn def prepare_table(self) -> None: self.connect() cur = self.conn.cursor() try: cur.execute(f"DROP TABLE IF EXISTS {self.table_name}") cur.execute(f""" CREATE TABLE {self.table_name} ( event_id UUID PRIMARY KEY, user_id INTEGER NOT NULL, event_type VARCHAR(20) NOT NULL, value DECIMAL(10,2) NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT NOW() ) """) cur.execute(f""" CREATE INDEX idx_bench_user_id ON {self.table_name} (user_id) """) cur.execute(f""" CREATE INDEX idx_bench_created_at ON {self.table_name} (created_at) """) cur.execute(f""" CREATE INDEX idx_bench_event_type ON {self.table_name} (event_type) """) self.conn.commit() except Exception: self.conn.rollback() raise finally: cur.close() def cleanup(self) -> None: if self.conn: try: cur = self.conn.cursor() cur.execute(f"DROP TABLE IF EXISTS {self.table_name}") self.conn.commit() cur.close() except Exception: self.conn.rollback() finally: self.conn.close() def benchmark_insert(self) -> BenchmarkResult: rows = self.generate_rows(self.row_count) insert_sql = f""" INSERT INTO {self.table_name} (event_id, user_id, event_type, value, created_at) VALUES (%s, %s, %s, %s, %s) ON CONFLICT (event_id) DO NOTHING """ cur = self.conn.cursor() try: batch = [ (r["event_id"], r["user_id"], r["event_type"], r["value"], r["created_at"]) for r in rows ] start = time.perf_counter() cur.executemany(insert_sql, batch) self.conn.commit() elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="批量写入", row_count=self.row_count, elapsed_ms=round(elapsed, 2), throughput=round(self.row_count / (elapsed / 1000), 1), success=True, ) except Exception as e: self.conn.rollback() return BenchmarkResult( operation="批量写入", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) finally: cur.close() def benchmark_range_query(self) -> BenchmarkResult: cur = self.conn.cursor() try: start = time.perf_counter() cur.execute(f""" SELECT user_id, SUM(value) AS total_value, COUNT(*) AS cnt FROM {self.table_name} WHERE created_at >= NOW() - INTERVAL '7 days' GROUP BY user_id HAVING COUNT(*) > 10 ORDER BY total_value DESC LIMIT 1000 """) rows_fetched = cur.rowcount if cur.rowcount >= 0 else 0 elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="范围查询", row_count=rows_fetched, elapsed_ms=round(elapsed, 2), throughput=round(rows_fetched / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: self.conn.rollback() return BenchmarkResult( operation="范围查询", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) finally: cur.close() def benchmark_aggregate(self) -> BenchmarkResult: cur = self.conn.cursor() try: start = time.perf_counter() cur.execute(f""" SELECT event_type, DATE(created_at) AS day, COUNT(*) AS event_count, AVG(value) AS avg_value, PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY value) AS p95 FROM {self.table_name} GROUP BY event_type, DATE(created_at) ORDER BY day DESC, event_type LIMIT 500 """) rows_fetched = cur.rowcount if cur.rowcount >= 0 else 0 elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="聚合分析", row_count=rows_fetched, elapsed_ms=round(elapsed, 2), throughput=round(rows_fetched / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: self.conn.rollback() return BenchmarkResult( operation="聚合分析", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) finally: cur.close() def benchmark_point_query(self) -> BenchmarkResult: cur = self.conn.cursor() try: sample_id = random.randint(1, self.row_count) start = time.perf_counter() cur.execute(f""" SELECT * FROM {self.table_name} WHERE user_id = %s ORDER BY created_at DESC LIMIT 10 """, (sample_id,)) _ = cur.fetchall() elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="点查询", row_count=10, elapsed_ms=round(elapsed, 2), throughput=round(10 / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: self.conn.rollback() return BenchmarkResult( operation="点查询", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) finally: cur.close() class MongoDBBenchmark(BaseBenchmark): """MongoDB 基准测试实现""" def __init__(self, conn_config: dict): super().__init__(conn_config) self.client = None self.db = None self.collection = None self.collection_name = "bench_events" @retry_on_error(max_retries=3) def connect(self) -> Any: uri = self.conn_config.get( "uri", f"mongodb://{self.conn_config.get('host', 'localhost')}:" f"{self.conn_config.get('port', 27017)}/", ) self.client = pymongo.MongoClient( uri, serverSelectionTimeoutMS=10000, connectTimeoutMS=10000, ) self.db = self.client[self.conn_config.get("database", "benchmark")] self.collection = self.db[self.collection_name] return self.client def prepare_table(self) -> None: self.connect() self.collection.drop() self.collection.create_indexes([ IndexModel([("user_id", ASCENDING)], name="idx_user_id"), IndexModel([("created_at", ASCENDING)], name="idx_created_at"), IndexModel([("event_type", ASCENDING)], name="idx_event_type"), ]) def cleanup(self) -> None: if self.client: try: self.collection.drop() except Exception: pass self.client.close() def benchmark_insert(self) -> BenchmarkResult: rows = self.generate_rows(self.row_count) docs = [ { "event_id": r["event_id"], "user_id": r["user_id"], "event_type": r["event_type"], "value": r["value"], "created_at": r["created_at"], } for r in rows ] try: start = time.perf_counter() result = self.collection.insert_many(docs, ordered=False) elapsed = (time.perf_counter() - start) * 1000 inserted = len(result.inserted_ids) return BenchmarkResult( operation="批量写入", row_count=inserted, elapsed_ms=round(elapsed, 2), throughput=round(inserted / (elapsed / 1000), 1), success=True, ) except pymongo.errors.BulkWriteError as e: inserted = e.details.get("nInserted", 0) return BenchmarkResult( operation="批量写入", row_count=inserted, elapsed_ms=0, throughput=0, success=False, error_message=f"部分写入失败: {len(e.details.get('writeErrors', []))} 条", ) except Exception as e: return BenchmarkResult( operation="批量写入", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) def benchmark_range_query(self) -> BenchmarkResult: try: cutoff = datetime.now() - timedelta(days=7) pipeline = [ {"$match": {"created_at": {"$gte": cutoff}}}, { "$group": { "_id": "$user_id", "total_value": {"$sum": "$value"}, "cnt": {"$sum": 1}, } }, {"$match": {"cnt": {"$gt": 10}}}, {"$sort": {"total_value": -1}}, {"$limit": 1000}, ] start = time.perf_counter() results = list(self.collection.aggregate(pipeline, allowDiskUse=True)) elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="范围查询", row_count=len(results), elapsed_ms=round(elapsed, 2), throughput=round(len(results) / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: return BenchmarkResult( operation="范围查询", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) def benchmark_aggregate(self) -> BenchmarkResult: try: pipeline = [ { "$group": { "_id": { "event_type": "$event_type", "day": {"$dateToString": { "format": "%Y-%m-%d", "date": "$created_at", }}, }, "event_count": {"$sum": 1}, "avg_value": {"$avg": "$value"}, } }, {"$sort": {"_id.day": -1, "_id.event_type": 1}}, {"$limit": 500}, ] start = time.perf_counter() results = list(self.collection.aggregate(pipeline, allowDiskUse=True)) elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="聚合分析", row_count=len(results), elapsed_ms=round(elapsed, 2), throughput=round(len(results) / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: return BenchmarkResult( operation="聚合分析", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) def benchmark_point_query(self) -> BenchmarkResult: try: sample_id = random.randint(1, self.row_count) start = time.perf_counter() results = list( self.collection.find({"user_id": sample_id}) .sort("created_at", -1) .limit(10) ) elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="点查询", row_count=len(results), elapsed_ms=round(elapsed, 2), throughput=round(len(results) / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: return BenchmarkResult( operation="点查询", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) class ClickHouseBenchmark(BaseBenchmark): """ClickHouse 基准测试实现""" def __init__(self, conn_config: dict): super().__init__(conn_config) self.client = None self.table_name = "bench_events" @retry_on_error(max_retries=3) def connect(self) -> Any: self.client = CHClient( host=self.conn_config.get("host", "localhost"), port=self.conn_config.get("port", 9000), database=self.conn_config.get("database", "default"), user=self.conn_config.get("user", "default"), password=self.conn_config.get("password", ""), connect_timeout=10, send_receive_timeout=30, ) return self.client def prepare_table(self) -> None: self.connect() self.client.execute(f"DROP TABLE IF EXISTS {self.table_name} SYNC") self.client.execute(f""" CREATE TABLE {self.table_name} ( event_id String, user_id UInt32, event_type LowCardinality(String), value Decimal(10, 2), created_at DateTime ) ENGINE = MergeTree() ORDER BY (event_type, created_at) PARTITION BY toYYYYMM(created_at) SETTINGS index_granularity = 8192 """) def cleanup(self) -> None: if self.client: try: self.client.execute( f"DROP TABLE IF EXISTS {self.table_name} SYNC" ) except Exception: pass self.client.disconnect() def benchmark_insert(self) -> BenchmarkResult: rows = self.generate_rows(self.row_count) data = [ ( r["event_id"], r["user_id"], r["event_type"], r["value"], r["created_at"], ) for r in rows ] try: start = time.perf_counter() self.client.execute( f"INSERT INTO {self.table_name} VALUES", data, ) elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="批量写入", row_count=len(data), elapsed_ms=round(elapsed, 2), throughput=round(len(data) / (elapsed / 1000), 1), success=True, ) except Exception as e: return BenchmarkResult( operation="批量写入", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) def benchmark_range_query(self) -> BenchmarkResult: try: start = time.perf_counter() result = self.client.execute(f""" SELECT user_id, sum(value) AS total_value, count() AS cnt FROM {self.table_name} WHERE created_at >= now() - INTERVAL 7 DAY GROUP BY user_id HAVING cnt > 10 ORDER BY total_value DESC LIMIT 1000 """) elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="范围查询", row_count=len(result), elapsed_ms=round(elapsed, 2), throughput=round(len(result) / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: return BenchmarkResult( operation="范围查询", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) def benchmark_aggregate(self) -> BenchmarkResult: try: start = time.perf_counter() result = self.client.execute(f""" SELECT event_type, toDate(created_at) AS day, count() AS event_count, avg(value) AS avg_value, quantile(0.95)(value) AS p95 FROM {self.table_name} GROUP BY event_type, day ORDER BY day DESC, event_type LIMIT 500 """) elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="聚合分析", row_count=len(result), elapsed_ms=round(elapsed, 2), throughput=round(len(result) / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: return BenchmarkResult( operation="聚合分析", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) def benchmark_point_query(self) -> BenchmarkResult: try: sample_id = random.randint(1, self.row_count) start = time.perf_counter() result = self.client.execute(f""" SELECT * FROM {self.table_name} WHERE user_id = {sample_id} ORDER BY created_at DESC LIMIT 10 """) elapsed = (time.perf_counter() - start) * 1000 return BenchmarkResult( operation="点查询", row_count=len(result), elapsed_ms=round(elapsed, 2), throughput=round(len(result) / (elapsed / 1000), 1) if elapsed > 0 else 0, success=True, ) except Exception as e: return BenchmarkResult( operation="点查询", row_count=0, elapsed_ms=0, throughput=0, success=False, error_message=str(e), ) def run_benchmark_suite( configs: dict, databases: Optional[List[str]] = None, ) -> dict: """运行完整的基准测试套件 Args: configs: 数据库连接配置字典 databases: 指定测试的数据库列表，None 表示全部 Returns: dict: 各数据库的测试报告映射 """ benchmark_registry = { "postgresql": PostgreSQLBenchmark, "mongodb": MongoDBBenchmark, "clickhouse": ClickHouseBenchmark, } if databases is None: databases = list(benchmark_registry.keys()) reports = {} for db_name in databases: if db_name not in benchmark_registry: print(f" 跳过未知数据库: {db_name}") continue if db_name not in configs: print(f" 跳过无配置的数据库: {db_name}") continue bench_cls = benchmark_registry[db_name] try: bench = bench_cls(configs[db_name]) report = bench.run() reports[db_name] = report print(report.summary()) except Exception as e: print(f" {db_name} 基准测试异常: {e}") return reports if __name__ == "__main__": # 使用示例：根据实际环境修改连接配置 configs = { "postgresql": { "host": "localhost", "port": 5432, "database": "benchmark", "user": "postgres", "password": "", }, "mongodb": { "host": "localhost", "port": 27017, "database": "benchmark", }, "clickhouse": { "host": "localhost", "port": 9000, "database": "default", "user": "default", "password": "", }, } reports = run_benchmark_suite(configs) # 输出对比摘要 print("\n" + "=" * 60) print(" 多数据库性能对比") print("=" * 60) for db, report in reports.items(): passed = sum(1 for r in report.results if r.success) total = len(report.results) found = sum(1 for r in report.results if r.row_count > 0) print(f" {db}: {passed}/{total} 场景通过, {found} 场景返回数据") print("=" * 60)

典型测试数据（10万行事件记录，AWS c5.2xlarge 同等配置）：

MongoDB 写入和省去 JOIN 的查询更快。ClickHouse 在聚合分析上断层领先。PG 在事务一致性和复杂 JOIN 上是唯一可行选择。测试数据不适用于所有场景，但方向性结论可靠。

四、创业团队 TCO：三年总拥有成本

性能不是创业团队的唯一变量。真正的决策约束是三年 TCO。

以 5 人技术团队、日活 5 万、日增 200 万行事件数据为假设：

""" 创业团队数据库三年 TCO 计算模型 """ from dataclasses import dataclass from typing import Dict, List @dataclass class TCOItem: """单项成本""" name: str monthly_cost: float notes: str @dataclass class TCOSummary: """TCO 汇总""" database: str monthly_items: Dict[str, float] monthly_total: float annual_total: float three_year_total: float def breakdown(self) -> str: lines = [ f"\n{'='*50}", f" {self.database} 三年 TCO 明细", f"{'='*50}", ] for name, cost in sorted( self.monthly_items.items(), key=lambda x: -x[1] ): pct = (cost / self.monthly_total * 100) if self.monthly_total else 0 lines.append(f" {name:<20s}: ¥{cost:>10,.0f}/月 ({pct:>5.1f}%)") lines.append(f" {'─' * 40}") lines.append(f" {'月均总计':<20s}: ¥{self.monthly_total:>10,.0f}") lines.append(f" {'年度总计':<20s}: ¥{self.annual_total:>10,.0f}") lines.append(f" {'三年总计':<20s}: ¥{self.three_year_total:>10,.0f}") lines.append(f"{'='*50}\n") return "\n".join(lines) def calculate_tco( database: str, server_monthly: float, dba_hours_per_month: float, hourly_rate: float = 150, ) -> TCOSummary: """计算三年 TCO Args: database: 数据库名称 server_monthly: 服务器月费（云实例或自建摊销） dba_hours_per_month: 每月运维人时 hourly_rate: 运维人员时薪（含管理成本） """ personnel_cost = dba_hours_per_month * hourly_rate monthly_total = server_monthly + personnel_cost return TCOSummary( database=database, monthly_items={ "服务器/云实例": server_monthly, "运维人力": personnel_cost, }, monthly_total=monthly_total, annual_total=monthly_total * 12, three_year_total=monthly_total * 36, ) # 场景一：PostgreSQL 单机 + 流复制 pg_tco = calculate_tco( database="PostgreSQL（单机 + 流复制）", server_monthly=800, # 4C16G 云实例 dba_hours_per_month=8, # 每周约 2 小时运维 ) # 场景二：MongoDB 副本集 mongo_tco = calculate_tco( database="MongoDB（3 节点副本集）", server_monthly=1200, # 3 台 4C8G dba_hours_per_month=6, # 运维更轻量 ) # 场景三：ClickHouse 单节点 ch_tco = calculate_tco( database="ClickHouse（单节点）", server_monthly=1000, # 8C32G，存储 I/O 要求高 dba_hours_per_month=10, # 分区管理、物化视图维护 ) # 场景四：PG + ClickHouse 混合架构 hybrid_tco_pg = calculate_tco( database="PostgreSQL（业务库）", server_monthly=800, dba_hours_per_month=6, ) hybrid_tco_ch = calculate_tco( database="ClickHouse（分析库）", server_monthly=1000, dba_hours_per_month=8, ) hybrid_combined = TCOSummary( database="混合架构（PG + ClickHouse）", monthly_items={ **hybrid_tco_pg.monthly_items, **{f"CH-{k}": v for k, v in hybrid_tco_ch.monthly_items.items()}, }, monthly_total=hybrid_tco_pg.monthly_total + hybrid_tco_ch.monthly_total, annual_total=hybrid_tco_pg.annual_total + hybrid_tco_ch.annual_total, three_year_total=hybrid_tco_pg.three_year_total + hybrid_tco_ch.three_year_total, ) # 输出对比 for tco in [pg_tco, mongo_tco, ch_tco, hybrid_combined]: print(tco.breakdown()) # 横向对比 print(f"{'='*50}") print(f" 三年 TCO 横向对比") print(f"{'='*50}") comparisons = [ ("PostgreSQL 单一库", pg_tco.three_year_total), ("MongoDB 单一库", mongo_tco.three_year_total), ("ClickHouse 单一库", ch_tco.three_year_total), ("PG + ClickHouse 混合", hybrid_combined.three_year_total), ] for name, total in sorted(comparisons, key=lambda x: x[1]): print(f" {name:<25s}: ¥{total:>12,.0f}") print(f"{'='*50}")

计算结果（单位：人民币）：

单一数据库方案三年总成本约 7-9 万，混合架构约 14 万。差额主要来自多出来的服务器和运维人力。但混合架构在分析能力上的收益，通常能覆盖这部分增量。如果团队没有分析需求，多花这笔钱就是浪费。

隐性成本也需要纳入考量。PG 招聘市场供给充足，MongoDB 运维人才更难找，ClickHouse 专家更是稀缺。这三者的学习曲线依次递增，对应的人员流动风险和交接成本也依次递增。选型不只看账单，还要看团队能不能长期持有。

五、总结

1. 场景第一，性能第二：OLTP 业务选 PG，文档型内容选 MongoDB，分析型报表选 ClickHouse，不要用通用性替代针对性。
2. 事务边界决定下限：有跨表一致性需求，PG 是唯一选项。MongoDB 4.0+ 虽支持多文档事务，性能和隔离级别仍不及 PG。
3. Schema 稳定性影响长期成本：业务频繁迭代时，MongoDB 的 Schema-less 省掉大量 DDL 迁移工作；模式稳定后，PG 的类型约束反过来降低数据质量风险。
4. 列式分析不应和 OLTP 混库：聚合查询在 PG/Mongo 上可用但不可靠，数据量跨过千万行后，ClickHouse 的优势从 2-3 倍拉大到 10 倍以上。
5. TCO 要用三年计算：只看首月服务器成本容易低估运维人力和人员流动成本。混合架构初始投入高，但业务分析和业务库解耦后，独立扩缩容和故障隔离的价值在第二年就会体现。
6. 人员可得性是选型的硬约束：选一个团队找不到人维护的数据库，等于给自己挖坑。PG 生态成熟，MongoDB 次之，ClickHouse 需要专人投入。

数据库选型没有银弹。决策的质量不取决于选的谁更先进，而取决于团队对自身场景的认知深度和未来两年业务走向的判断精度。