neonというDBサービスが話題なので移行を検討してみた
karrinn
ソフトウェアエンジニア目次
Neonとは
筆者はデータベースは特段の事情がなければずっとPostgreSQLを使用していますが、NeonというPostgreSQL互換のデータベースサービスがあるようで、本ブログはまだ記事もほぼなく、無料枠もあるようなので実験がてらこのブログのDBをNeonに移行してみることにしました。
先に結論：日本で本番運用はさすがにきつい
アカウント作ってる途中からうすうすそうだろうなとは思っていましたが、今のところ本番運用は厳しそうでした。
理由:東京（日本）リージョンがない
Neonは本記事執筆（2025年3月）時点では東京リージョンがなく、一番近くてもシンガポールリージョンでした。
筆者はこれまでDBはRDS、CloudSQL、またはVPSに直インストールしたDBサーバを使っていましたので、ほとんど地理的レイテンシーの問題にぶつかることはありませんでした。
地理的レイテンシーそれなりに出るだろうなとは思っていましたが、実際移行してみるとこんなにか、ってくらい厳しいものがありました。さすがに本番運用は厳しいですね。
スペックはほぼ問題なさそう
作りにもよりますが、ちゃんと作っていればブログのような小規模プロジェクトではスペック的部分では問題にならないんじゃないかなと思いました。
表にまとめてみました
実際地理的レイテンシー、マシンスペックによるレイテンシーはどんなもんかを調べてまとめてみました
| 平均値 (ms) | 最小-最大 (ms) | |
|---|---|---|
接続時間 |
505.64 |
480.22-531.63 |
単純クエリ |
180.77 |
160.61-226.59 |
複雑クエリ |
100.93 |
89.92-122.14 |
初回接続時のオーバーヘッドが非常に大きい。認証と地理的距離の両方が影響している。
| 平均値 (ms) | 最小-最大 (ms) | |
|---|---|---|
少量データ |
107.37 |
79.29-163.23 |
中量データ |
80.02 |
79.79-80.15 |
大量データ |
85.77 |
79.99-91.39 |
基本的な往復時間（約80ms）に加え、初回は少し時間がかかっている。 複雑な処理でもレイテンシーは往復時間に近く、スペックの制約はあまり見られない。
測定ロジック
import time
import psycopg2
from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional
from statistics import mean
@dataclass
class MeasurementResult:
"""測定結果を格納するデータクラス"""
avg: float
min: float
max: float
all_times: List[float] = field(default_factory=list)
@dataclass
class GeographicalLatencyResult:
"""地理的レイテンシー測定の結果"""
connection: MeasurementResult
simple_query: MeasurementResult
complex_query: MeasurementResult
@dataclass
class SpecLatencyResult:
"""マシンスペック影響測定の結果"""
small_data: MeasurementResult
medium_data: MeasurementResult
large_data: MeasurementResult
class DatabaseLatencyTester:
"""データベースのレイテンシーを測定するクラス"""
def __init__(self, connection_string: str):
"""
初期化メソッド
Args:
connection_string: データベース接続文字列
(形式: postgresql://user:password@host:port/dbname)
"""
# 接続文字列から認証情報を隠す
self.connection_string = self._sanitize_conn_string(connection_string)
self.original_connection_string = connection_string
def _sanitize_conn_string(self, conn_string: str) -> str:
"""接続文字列から認証情報を隠す"""
if '://' in conn_string:
# ユーザー名とパスワードを隠す
parts = conn_string.split('://')
if '@' in parts[1]:
auth_part = parts[1].split('@')[0]
rest_part = parts[1].split('@')[1]
if ':' in auth_part:
username = auth_part.split(':')[0]
return f"{parts[0]}://{username}:<password>@{rest_part}"
return conn_string
def measure_geographical_latency(self, iterations: int = 5) -> GeographicalLatencyResult:
"""
地理的レイテンシーを測定する
Args:
iterations: 測定回数
Returns:
地理的レイテンシー測定結果
"""
# 結果を格納する辞書
results: Dict[str, List[float]] = {
"connection_times": [],
"simple_query_times": [],
"complex_query_times": []
}
for i in range(iterations):
# 接続時間測定
start = time.time()
conn = psycopg2.connect(self.original_connection_string)
end = time.time()
results["connection_times"].append((end - start) * 1000)
cursor = conn.cursor()
# 単純クエリ（往復時間を計測）
start = time.time()
cursor.execute("SELECT 1")
cursor.fetchall()
end = time.time()
results["simple_query_times"].append((end - start) * 1000)
# 複雑なクエリ（データ処理時間+往復時間を計測）
start = time.time()
cursor.execute("""
SELECT version(), pg_sleep(0.01);
""")
cursor.fetchall()
end = time.time()
results["complex_query_times"].append((end - start) * 1000)
conn.close()
time.sleep(1) # 測定間隔
# 測定結果を集計して返す
return GeographicalLatencyResult(
connection=MeasurementResult(
avg=mean(results["connection_times"]),
min=min(results["connection_times"]),
max=max(results["connection_times"]),
all_times=results["connection_times"]
),
simple_query=MeasurementResult(
avg=mean(results["simple_query_times"]),
min=min(results["simple_query_times"]),
max=max(results["simple_query_times"]),
all_times=results["simple_query_times"]
),
complex_query=MeasurementResult(
avg=mean(results["complex_query_times"]),
min=min(results["complex_query_times"]),
max=max(results["complex_query_times"]),
all_times=results["complex_query_times"]
)
)
def measure_spec_latency(self, table_name: str = 'wagtailcore_page', iterations: int = 3) -> SpecLatencyResult:
"""
マシンスペックの影響を測定する
Args:
table_name: 測定に使用するテーブル名
iterations: 測定回数
Returns:
スペック影響測定結果
"""
# 結果を格納する辞書
results: Dict[str, List[float]] = {
"small_data_times": [],
"medium_data_times": [],
"large_data_times": []
}
conn = psycopg2.connect(self.original_connection_string)
cursor = conn.cursor()
for i in range(iterations):
# 少量データ取得（スペックの影響少）
start = time.time()
cursor.execute(f"SELECT id, title FROM {table_name} LIMIT 10")
cursor.fetchall()
end = time.time()
results["small_data_times"].append((end - start) * 1000)
# 中量データ取得
start = time.time()
cursor.execute(f"""
SELECT id, title, slug, path
FROM {table_name}
ORDER BY id DESC
LIMIT 100
""")
cursor.fetchall()
end = time.time()
results["medium_data_times"].append((end - start) * 1000)
# 大量データ処理（スペックの影響大）
start = time.time()
cursor.execute(f"""
WITH RECURSIVE tree AS (
SELECT id, title, path, 1 AS level
FROM {table_name}
WHERE depth = 1
UNION ALL
SELECT p.id, p.title, p.path, t.level + 1
FROM {table_name} p
JOIN tree t ON p.path LIKE (t.path || '%')
WHERE p.id != t.id
)
SELECT id, title, path, level
FROM tree
ORDER BY path
LIMIT 200;
""")
cursor.fetchall()
end = time.time()
results["large_data_times"].append((end - start) * 1000)
time.sleep(2) # 測定間隔
conn.close()
# 測定結果を集計して返す
return SpecLatencyResult(
small_data=MeasurementResult(
avg=mean(results["small_data_times"]),
min=min(results["small_data_times"]),
max=max(results["small_data_times"]),
all_times=results["small_data_times"]
),
medium_data=MeasurementResult(
avg=mean(results["medium_data_times"]),
min=min(results["medium_data_times"]),
max=max(results["medium_data_times"]),
all_times=results["medium_data_times"]
),
large_data=MeasurementResult(
avg=mean(results["large_data_times"]),
min=min(results["large_data_times"]),
max=max(results["large_data_times"]),
all_times=results["large_data_times"]
)
)
def print_results(self, geo_results: GeographicalLatencyResult, spec_results: SpecLatencyResult) -> None:
"""
測定結果を整形して表示する
Args:
geo_results: 地理的レイテンシー測定結果
spec_results: スペック影響測定結果
"""
print("\n===== マシンスペック影響測定結果 =====")
print(f"少量データ: 平均 {spec_results.small_data.avg:.2f}ms (最小: {spec_results.small_data.min:.2f}ms, 最大: {spec_results.small_data.max:.2f}ms)")
print(f"中量データ: 平均 {spec_results.medium_data.avg:.2f}ms (最小: {spec_results.medium_data.min:.2f}ms, 最大: {spec_results.medium_data.max:.2f}ms)")
print(f"大量データ: 平均 {spec_results.large_data.avg:.2f}ms (最小: {spec_results.large_data.min:.2f}ms, 最大: {spec_results.large_data.max:.2f}ms)")
print("\n===== 地理的レイテンシー測定結果 =====")
print(f"接続時間: 平均 {geo_results.connection.avg:.2f}ms (最小: {geo_results.connection.min:.2f}ms, 最大: {geo_results.connection.max:.2f}ms)")
print(f"単純クエリ: 平均 {geo_results.simple_query.avg:.2f}ms (最小: {geo_results.simple_query.min:.2f}ms, 最大: {geo_results.simple_query.max:.2f}ms)")
print(f"複雑クエリ: 平均 {geo_results.complex_query.avg:.2f}ms (最小: {geo_results.complex_query.min:.2f}ms, 最大: {geo_results.complex_query.max:.2f}ms)")
def main():
"""メイン関数"""
# データベース接続情報
connection_string = "postgresql://<username>:<password>@<hostname>.<region>.aws.neon.tech/<dbname>?sslmode=require"
# テスターのインスタンスを作成
tester = DatabaseLatencyTester(connection_string)
# 測定を実行
print("データベースレイテンシー測定を開始します...")
print(f"対象データベース: {tester.connection_string}")
geo_results = tester.measure_geographical_latency(iterations=5)
spec_results = tester.measure_spec_latency(iterations=3)
# 結果を表示
tester.print_results(geo_results, spec_results)
print("\n測定完了。")
if __name__ == "__main__":
main()とはいえ開発環境、テスト環境としては最高
neonでは無料枠プランでもWebGUIでDBの中身を見れたり（DBeaverやA5m2みたいな）DBにブランチを切ったりできます。
開発環境、テスト環境として使うDBなら最高じゃないでしょうか。
筆者はたいていdocker composeにDBも入れてそれ使っちゃってますが、クラウド上に切り離していると何かと便利なシーンがあります。
とはいえ今の開発スタイルになれちゃってますし、特段不満もないので結局のところ開発環境を移行するかは迷うところですが、選択肢には入れて考えてみたいと思います。
コメント欄もありますので、なにか気になることがあればお気軽にコメントください！
まだコメントはありません。最初のコメントを残しませんか？