DuckDBとFastAPIで1kmメッシュの詳細・統計APIを作る ― 防災Web地図 Step 5
はじめに
ここまでのStepで、全国1kmメッシュと防災GISデータをGeoParquetへ整理し、表示用のPMTilesを作り、NginxとOpenLayersで地図を表示できるところまで進めました。
Step 4の時点でも地図は動きます。ただし、PMTilesへ長い説明文や出典情報、集計値をすべて詰め込むと、タイルが大きくなります。データセットの一覧、危険度クラス別の件数、第1次メッシュごとの集計なども、静的タイルだけで扱うには少し無理があります。
そこでStep 5では、表示と検索を分けます。
PMTiles
地図の形状
色分けに必要な最低限の属性
DuckDB + FastAPI
クリックしたメッシュの詳細
データセットの出典・版・基準日
危険度クラス別の統計
第1次メッシュ別の統計
経度・緯度からのメッシュ照会
全国1kmメッシュのgeometryはPMTilesが持っているため、FastAPI用のDuckDBには重ねて保存しません。Step 1とStep 2のGeoParquetから必要な属性だけを登録し、読み取り専用APIとして運用します。
この記事では、DuckDBの構築、検査済みDBの配置、FastAPI、OpenAPI、pytest、スモークテストまでをDocker Compose環境でまとめます。
シリーズ全体での位置付け
全体の構成
初回
Step 1 e-Statの全国1kmメッシュをGeoParquetへ変換
Step 2 防災区域を正規化し、1kmメッシュへ集計
Step 3 GeoParquetからPMTilesを作成
Step 4 NginxとOpenLayersでPMTilesを表示
Step 5 DuckDBとFastAPIで詳細・統計APIを追加 ← 今回
最終回 Step 1～5を1つのDocker Compose環境へ統合
Step 5が直接使うのは、Step 1のメッシュマスターと、Step 2の1km集計GeoParquetです。Step 3・Step 4のPMTilesはAPIの入力にはしません。
今回の構成
処理は大きく分けて次の順です。
- Step 1・Step 2の管理JSONとSHA-256を確認する
- GeoParquetの必須列、件数、コードを確認する
- geometryを除いた属性をDuckDBへ登録する
- 検索用の索引、統計テーブル、viewを作る
- DBを読み取り専用で開き直して検査する
- 検査済みDBをFastAPI用のruntimeへ配置する
- FastAPIを起動する
- OpenAPI、pytest、実HTTP要求を確認する
- 結果をmanifestへ追記する
PMTilesとDuckDBへ同じgeometryを持たせない
Step 4では、OpenLayersがPMTilesから1kmメッシュのポリゴンを読み込みます。
Nginx
└─ flood_maximum_1km__20260806_r001.pmtiles
└─ OpenLayers
ここでFastAPI用DuckDBにも全国分のポリゴンを登録すると、同じ形状を2か所で管理することになります。
PMTiles 全国1kmメッシュ形状
DuckDB 全国1kmメッシュ形状
この構成でも動きますが、Step 5で必要なのは次の処理です。
8桁のmesh_codeで1件検索
データセット一覧
危険度クラス別集計
第1次メッシュ別集計
出典・版・基準日の表示
いずれもgeometryなしで処理できます。そのため、DuckDBへ登録するメッシュマスターは次の列に絞りました。
feature_id
mesh_code
mesh1_code
mesh2_code
mesh3_code
center_lon
center_lat
Step 2の集計も属性だけを登録します。
feature_id
mesh_code
dataset_id
hazard_type
scenario_id
mesh1_code
has_hazard
max_class
max_class_label
max_value
affected_area_m2
mesh_area_m2
affected_area_ratio
source_feature_count
intersection_count
これで、地図表示用ファイルとAPI用DBの役割が分かれます。
Step 1・Step 2成果物を先に検査する
DuckDBへ読み込む前に、次の管理ファイルを確認します。
workspace/artifacts/step1/
├─ mesh_master_1km.parquet
├─ manifest.json
└─ validation_report.json
workspace/artifacts/step2/
├─ summary_1km/
├─ dataset_catalog_fragment.json
├─ manifest.json
└─ validation_report.json
Step 5では、次の状態なら処理を止めます。
manifestのstepが違う
statusがpassedではない
Step 2のcatalogとmanifestで成果物版が違う
GeoParquetのSHA-256が管理JSONと一致しない
必須列がない
mesh_codeが8桁ではない
1データセット内でmesh_codeが重複している
catalogとParquetのdataset_idが違う
catalogとParquetのscenario_idが違う
affected_area_ratioが0～1の範囲外
上流成果物を読み込む処理はsrc/runtime_db_step5/upstream.pyへ分けています。
from pathlib import Path
from .hashing import sha256_file
def _verify_hash(
path: Path,
expected_sha256: str,
*,
role: str,
) -> ArtifactFile:
"""ファイルの存在とSHA-256を確認し、ArtifactFileへまとめる。"""
if not path.is_file():
raise UpstreamArtifactError(
f"{role}がありません: {path}"
)
actual = sha256_file(path)
if actual != expected_sha256:
raise UpstreamArtifactError(
f"{role}のSHA-256が管理JSONと一致しません。\n"
f"path={path}\n"
f"expected={expected_sha256}\n"
f"actual={actual}"
)
return ArtifactFile(
role=role,
path=path,
sha256=actual,
)
単にファイルが存在するかではなく、Step 1・Step 2で検査したものと同じ内容かを確認してからDBを作ります。
DuckDBのテーブル構成
今回作成する主なテーブルは次のとおりです。
mesh_master
全国1kmメッシュの基本属性です。
CREATE TABLE mesh_master (
feature_id BIGINT PRIMARY KEY,
mesh_code VARCHAR NOT NULL UNIQUE,
mesh1_code VARCHAR NOT NULL,
mesh2_code VARCHAR NOT NULL,
mesh3_code VARCHAR NOT NULL,
center_lon DOUBLE NOT NULL,
center_lat DOUBLE NOT NULL
);
dataset_catalog
出典、版、基準日、シナリオ、利用条件などを保存します。
dataset_id
dataset_name
hazard_type
scenario_id
source_organization
source_version
reference_date
source_url
license_name
license_url
notes
class_labels_json
interpretation_notes_json
summary_sha256
summary_row_count
mesh_hazard_summary
Step 2の1km集計です。
CREATE TABLE mesh_hazard_summary (
feature_id BIGINT NOT NULL,
mesh_code VARCHAR NOT NULL,
dataset_id VARCHAR NOT NULL,
hazard_type VARCHAR NOT NULL,
scenario_id VARCHAR NOT NULL,
mesh1_code VARCHAR NOT NULL,
has_hazard BOOLEAN NOT NULL,
max_class INTEGER,
max_class_label VARCHAR,
max_value DOUBLE,
affected_area_m2 DOUBLE NOT NULL,
mesh_area_m2 DOUBLE NOT NULL,
affected_area_ratio DOUBLE NOT NULL,
source_feature_count BIGINT NOT NULL,
intersection_count BIGINT NOT NULL,
PRIMARY KEY (dataset_id, mesh_code)
);
集計済みテーブル
API要求のたびに全国集計をやり直さないよう、DB構築時に次を作ります。
dataset_statistics
データセット全体の件数、最大クラス、影響面積など
dataset_class_statistics
危険度クラス別のメッシュ件数、平均・最大面積率など
mesh1_statistics
第1次メッシュごとの集計
元データが更新されたときにDBを作り直す方式なので、API要求時の処理を軽くできます。
構築情報と上流ファイル
build_metadata
schema_version
artifact_version
DuckDB版
Step 1成果物版
Step 2成果物版
artifact_lineage
入力ファイルの役割
パス
SHA-256
件数
画面や運用ログから、どの版のデータを見ているか確認できます。
検索用の索引を作る
APIでよく使う条件に索引を作ります。
def _create_indexes_and_statistics(
connection: Any,
*,
master_mesh_count: int,
) -> None:
"""点検索用索引、統計テーブル、API向けviewを作る。"""
# PRIMARY KEYやUNIQUEだけに頼らず、APIで単独条件に使う列にも
# 索引を用意します。全国集計を速くするものではなく、クリック時の
# mesh_code検索やdataset_idの絞り込みを主な対象にしています。
connection.execute(
"CREATE INDEX mesh_summary_mesh_idx "
"ON mesh_hazard_summary(mesh_code)"
)
connection.execute(
"CREATE INDEX mesh_summary_dataset_idx "
"ON mesh_hazard_summary(dataset_id)"
)
connection.execute(
"CREATE INDEX mesh_summary_mesh1_idx "
"ON mesh_hazard_summary(mesh1_code)"
)
# この後、dataset_statistics、dataset_class_statistics、
# mesh1_statisticsを作成します。
大量のINSERTと索引作成が終わった後に、ANALYZEとCHECKPOINTを実行します。
# 大量INSERTの後に統計を更新し、結合順序を決めるための情報を整えます。
connection.execute("ANALYZE")
connection.execute("CHECKPOINT")
作成途中のDBを運用側へ見せない
DBを直接workspace/runtime/disaster.duckdbへ作ると、構築途中のファイルをFastAPIが開く可能性があります。
今回は次の順にしています。
作業ディレクトリにDBを作成
↓
索引・統計・メタデータを作成
↓
CHECKPOINT
↓
読み取り専用で開き直して検査
↓
Step 5成果物として確定
↓
APIを停止
↓
runtimeへコピー
↓
コピー後のSHA-256を再確認
↓
正式ファイル名へ切り替え
↓
APIを起動
promote.pyでは、コピー中のファイルと正式ファイルを分けています。
def promote_runtime_database(
artifact_dir: Path,
runtime_dir: Path,
) -> Path:
"""Step 5 DBをruntimeへコピーし、検査後に正式名へ切り替える。"""
runtime_dir.mkdir(parents=True, exist_ok=True)
working_path = runtime_dir / ".disaster.duckdb.copying"
final_path = runtime_dir / "disaster.duckdb"
working_path.unlink(missing_ok=True)
shutil.copyfile(database_path, working_path)
# コピー後にもハッシュを確認します。容量の大きなDBでコピーが途中終了した
# 場合や、意図しないファイルを参照した場合に、正式名へ切り替えません。
if sha256_file(working_path) != expected_sha:
working_path.unlink(missing_ok=True)
raise Step5Error(
"runtimeへのコピー後にSHA-256が一致しません。"
)
os.replace(working_path, final_path)
return final_path
APIを動かしたままDBだけ入れ替える構成にはしていません。運用スクリプトがAPIを停止し、配置後に再起動します。
FastAPIは読み取り専用で開く
APIはDBを書き換えません。
connection = duckdb.connect(
str(self.database_path),
read_only=True,
)
接続は少数だけ保持し、リクエストごとに1本借ります。
class DuckDBConnectionPool:
"""同時利用されない読み取り専用接続を少数保持する。
DuckDBの1接続を複数スレッドから同時に使用しないよう、要求ごとに
接続を1本借ります。返却された接続は次の要求で再利用します。
"""
@contextmanager
def connection(
self,
*,
timeout_seconds: float = 5.0,
) -> Iterator[Any]:
"""接続を1本貸し出し、処理後に必ずプールへ返す。"""
if not self._opened:
raise DuckDBPoolError(
"DuckDB接続プールが開かれていません。"
)
try:
connection = self._queue.get(
timeout=timeout_seconds
)
except Empty as exc:
raise DuckDBPoolError(
"利用可能なDuckDB接続がありません。"
) from exc
try:
yield connection
finally:
self._queue.put(connection)
FastAPIのlifespanで起動時に接続を作り、停止時に閉じます。
@asynccontextmanager
async def lifespan(
app: FastAPI,
) -> AsyncIterator[None]:
"""起動時にDBを確認し、停止時に全接続を閉じる。"""
pool = factory(settings)
pool.open()
pool.ping()
app.state.db_pool = pool
try:
yield
finally:
pool.close()
初期設定は次のとおりです。
Uvicorn worker 1
DuckDB接続数 2
1接続あたりthreads 2
memory_limit 1GB
最初からワーカー数とDuckDBスレッド数を大きくせず、実際の要求数とSQL時間を見ながら調整します。
API一覧
プロセスの生存確認
GET /api/v1/health/live
DuckDBを含む準備完了確認
GET /api/v1/health/ready
DB構築情報と上流成果物
GET /api/v1/metadata/build
データセット一覧
GET /api/v1/datasets
GET /api/v1/datasets?hazard_type=flood
GET /api/v1/datasets?scenario_id=maximum
1データセットの情報
GET /api/v1/datasets/{dataset_id}
データセット全体・クラス別統計
GET /api/v1/datasets/{dataset_id}/statistics
第1次メッシュ別統計
GET /api/v1/datasets/{dataset_id}/mesh1/{mesh1_code}/statistics
1kmメッシュ詳細
GET /api/v1/meshes/{mesh_code}
GET /api/v1/meshes/{mesh_code}?dataset_id=flood_a31a_2025
経度・緯度からメッシュ照会
GET /api/v1/identify?longitude=138.388&latitude=34.971
「データなし」と「該当なし」を分ける
防災データでは、集計行がない状態を安全とは判断できません。
元データの整備範囲外
対象シナリオが違う
Step 2で非該当メッシュを保存していない
条件を指定しすぎて該当行がない
これらは、危険がないこととは別です。
メッシュ詳細APIでは、次の3状態を分けます。
intersects_hazard
1件以上のhas_hazard=trueがある
no_hazard_in_summary
集計行はあるが、すべてhas_hazard=false
no_summary_record
条件に一致する集計行がない
組み立て処理は次のようにしています。
hazards = [
HazardSummary.model_validate(row)
for row in hazard_rows
]
if not hazards:
evaluation_status = "no_summary_record"
elif any(item.has_hazard for item in hazards):
evaluation_status = "intersects_hazard"
else:
evaluation_status = "no_hazard_in_summary"
レスポンスには、自治体の最新ハザードマップも確認するよう注意書きを含めます。
OpenLayersからの利用
Step 4のPMTilesにはmesh_codeを含めています。クリックした地物からコードを取得し、Step 5 APIへ問い合わせます。
map.on("singleclick", async (event) => {
const feature = map.forEachFeatureAtPixel(
event.pixel,
(candidate) => candidate,
);
if (!feature) {
return;
}
const meshCode = feature.get("mesh_code");
const response = await fetch(
`/api/v1/meshes/${encodeURIComponent(meshCode)}`,
);
if (!response.ok) {
throw new Error(
"メッシュ詳細を取得できませんでした。",
);
}
const detail = await response.json();
console.log(detail);
});
形状はPMTilesからすでに描画されているため、FastAPIからGeoJSONを返す必要はありません。
フォルダー構成
参考のために、ZIPファイルを置いておきますので、ダウンロードして活用してください。
japan_disaster_pmtiles_step5_r001/
├─ README.md
├─ VALIDATION.md
├─ japan_disaster_pmtiles_step5_qiita_r001.md
├─ docker-compose.yml
├─ Makefile
├─ .env.example
├─ pyproject.toml
│
├─ builder/
│ ├─ Dockerfile
│ └─ requirements.txt
│
├─ src/runtime_db_step5/
│ ├─ cli.py
│ ├─ config.py
│ ├─ database.py
│ ├─ demo.py
│ ├─ finalize.py
│ ├─ manifest.py
│ ├─ mesh_code.py
│ ├─ pipeline.py
│ ├─ promote.py
│ ├─ schema.py
│ ├─ upstream.py
│ └─ validation.py
│
├─ backend/
│ ├─ Dockerfile
│ ├─ requirements.txt
│ ├─ requirements-dev.txt
│ ├─ app/
│ │ ├─ main.py
│ │ ├─ config.py
│ │ ├─ db.py
│ │ ├─ dependencies.py
│ │ ├─ models.py
│ │ ├─ repositories.py
│ │ ├─ services.py
│ │ └─ routers/
│ └─ tools/
│ └─ smoke_api.py
│
├─ contracts/
├─ docs/
├─ tests/
├─ scripts/
└─ workspace/
Pythonソースには日本語docstring、型ヒント、引数、戻り値、例外条件を入れています。コメントは処理を日本語へ言い換えるのではなく、なぜその手順にしたのかを残しました。
Docker Compose
サービスは3つです。
builder
上流検査
DuckDB構築
読み戻し検査
runtimeへの配置
manifest確定
api
FastAPI
読み取り専用DuckDB
api-test
pytest
OpenAPI出力
HTTPスモークテスト
apiだけが常駐します。builderとapi-testは、データ更新や検査のときだけ起動します。
今回のDockerfileでは、次の版を固定しています。
Python 3.13
DuckDB 1.5.5
FastAPI 0.141.1
デモを実行する
ZIPを展開し、プロジェクトルートで実行します。
cp .env.example .env
make demo
デモでは、静岡駅周辺に架空の1kmメッシュと、架空の洪水・土砂災害集計を作ります。
Step 1互換のメッシュマスター
Step 2互換の洪水集計
Step 2互換の土砂災害集計
その後、次を自動で実行します。
Dockerイメージ作成
DuckDB構築
DB読み戻し検査
runtimeへの配置
FastAPI起動
OpenAPI出力
pytest
HTTPスモークテスト
manifest確定
APIドキュメントは次です。
http://localhost:8000/api/docs
停止します。
make down
デモ成果物を削除します。
make clean-demo
デモデータは処理確認専用であり、防災判断には使用できません。
実データで実行する
Step 1とStep 2の成果物を配置します。
workspace/artifacts/step1/
workspace/artifacts/step2/
一括実行します。
cp .env.example .env
make step5
個別に実行する場合は次の順です。
make build
make database
make promote
make up
make wait
make openapi
make test
make smoke
make finalize
make inspect
既存成果物を作り直す場合だけ、--overwriteを指定します。
./scripts/run_step5.sh --overwrite
成果物版は.envで管理します。
STEP5_ARTIFACT_VERSION=20260807_r001
作成される成果物
workspace/artifacts/step5/
├─ disaster.duckdb
├─ database_schema.json
├─ validation_report.json
├─ openapi.json
├─ api_test_report.xml
├─ api_smoke_report.json
├─ api_validation_report.json
└─ manifest.json
FastAPIが読むファイルは次へ配置します。
workspace/runtime/
├─ disaster.duckdb
└─ disaster.manifest.json
manifest.jsonには、Step 1・Step 2の版、入力SHA-256、DBのSHA-256、環境、件数、API検査結果を記録します。
検査内容
DB構築後に次を確認します。
DuckDBを読み取り専用で開ける
必須テーブル・viewがある
メッシュ件数、データセット件数、集計件数が一致
8桁でないmesh_codeがない
dataset_idとmesh_codeの重複がない
mesh_masterにないmesh_codeがない
dataset_catalogにないdataset_idがない
影響面積率が0～1の範囲
検索用索引がある
DBのschema版と成果物版が一致
API側では次を確認します。
OpenAPI JSONを生成できる
pytestの失敗・エラーが0
health/liveが200
health/readyが200
データセット一覧が取得できる
代表座標を照会できる
メッシュ詳細が取得できる
検査に失敗した場合はmanifest.jsonを最終的なpassedへ確定しません。
実装時に気を付けた点
APIへ更新処理を入れない
FastAPIは読み取りだけにしています。データ更新、Parquet読込、索引作成はbuilder側へ分けました。
生のSQLへ利用者入力を埋め込まない
dataset_idやmesh_codeはプレースホルダーへ渡します。
row = connection.execute(
"""
SELECT feature_id, mesh_code, center_lon, center_lat
FROM mesh_master
WHERE mesh_code = ?
""",
[mesh_code],
).fetchone()
MVTへ詳細属性を詰め込まない
OpenLayersの色分けに必要な属性はPMTiles、長い説明や統計はFastAPIと分けます。
集計レコードがない状態を安全と表示しない
no_summary_recordを明示し、注意文を返します。
DB更新中のファイルをAPIへ見せない
作業用DBを検査し、APIを止めてからruntimeへ配置します。
作成時の確認結果
配布前に、Pythonの構文、日本語docstring、Shell、JSON Schema、Docker Compose YAML、Markdownを確認しました。FastAPIのルーターと成果物管理を対象にしたテスト結果は次のとおりです。
10 passed, 1 skipped
スキップした1件は、デモGeoParquetからDuckDBを実際に構築する統合テストです。この記事を作成した環境にはDockerとduckdb Pythonパッケージがなかったため、ローカルでは実行していません。DockerイメージにはDuckDBを含めており、利用環境ではmake demoとmake testで同じ統合テストを実行できます。
make demo
make test
DockerでのDB構築、runtimeへの配置、FastAPI起動、HTTPスモークテストまで確認した後、workspace/artifacts/step5/manifest.jsonのstatusがpassedになっていることを確認してください。
おわりに
Step 5までで、地図表示と検索APIを分けた構成になりました。
GeoParquet
再集計・分析・再生成の元データ
PMTiles + Nginx
全国規模の固定レイヤーを静的配信
OpenLayers
地図表示、レイヤー切替、クリック
DuckDB + FastAPI
詳細、出典、統計、座標照会
次は、Step 3のPMTiles、Step 4のWeb画面、Step 5のDuckDBとFastAPIを、1つのDocker Compose環境へまとめます。更新時には新しい成果物を検査してから公開側へ切り替え、旧版へ戻せる構成にします。
参考資料
- DuckDB: Concurrency
- DuckDB Python DB API
- DuckDB: Reading and Writing Parquet Files
- DuckDB: Indexing
- DuckDB: ANALYZE Statement
- FastAPI: Lifespan Events
- FastAPI: Dependencies
- FastAPI: Testing Dependencies with Overrides
- 総務省統計局：地域メッシュ統計