当前位置：首页 > news >正文

第4章：数据获取与质量控制

news 2025/11/12 10:28:52

4.1 引言：从“能拿到”到“值得信任”

数据“从哪来、靠不靠谱、能不能复用”是每个 GIS 项目的首要问题。你可能面对的是开放平台上千条下载链接、不同坐标与字段、复杂的许可条款。本章以一个“城市公共设施选址”的准备阶段为例，展示如何从来源到质量评估构建一条可复制的数据准备管线，让每一次数据使用都有迹可循、可量化、可复用。

4.2 学习目标

识别与评估主流开放数据源（OSM、Copernicus/Sentinel、NASA Earthdata）
构建“来源-下载-标准化-质量评估-发布”的最小管线
掌握几何有效性、坐标一致性、异常检测与质量评分方法
输出可复现的质量报告与元数据字典，支持后续分析与共享

4.3 环境要求

Python 基础与包管理；能运行 requests、geopandas、rasterio
GIS 基础：CRS/投影、字段与属性表概念、文件格式认识

4.4 流程与架构（整体闭环）

下面的流程图总结了推荐的最小闭环：

4.5 核心概念与原则

来源可信：明确数据来源、采集时间、更新频率与许可条款；记录版本号
坐标一致：统一到项目目标 CRS；保留原始 CRS 以便追溯
字段规范：建立字段映射与编码规范，避免跨源拼接时出现语义歧义
几何合法：矢量几何必须可解析且拓扑合法；栅格需明确 nodata
异常可测：缺失率、离群值、重叠冲突等指标可量化、可报告

4.6 内容讲解（逐环节展开）

数据源盘点与许可

OSM：道路/建筑/POI 丰富，更新快；许可为 ODbL，需标注来源与衍生作品
Sentinel：多波段遥感影像，适合环境与变化监测；使用欧盟 Copernicus 许可
NASA：DEM、气象等多源数据；校验下载条件与使用条款

下载策略与缓存

批次下载：按地区或网格分块，避免大文件失败；记录“分块索引与范围”
断点续传：对 HTTP 下载设置重试与校验（哈希或大小）
缓存：对稳定数据使用本地缓存与版本目录（data/raw/<source>/<version>）

标准化与统一 CRS

对矢量使用 to_crs(target_crs)；对栅格使用重投影与重采样方法（最近邻/双线性）
字段标准化：建立字段映射字典（例如将 amenity/name 统一到项目标准）
元数据字典：记录每个字段的来源、含义、类型与允许值范围

几何有效性与拓扑修复

自交与裂缝：使用 buffer(0) 或 make_valid 修复；记录修复比例与剩余异常
重叠冲突：对多源边界进行差异分析并选择权威源或融合策略

异常检测与质量评分

缺失率：统计字段缺失与空几何比例
离群值：基于分位/标准差识别异常数值
一致性：跨源同字段的一致率与冲突计数
综合评分：加权汇总为 0–100 的质量分，以便项目选择与对比

质量报告与合规留痕（工程化）

质量报告：以 JSON/Markdown 输出质量摘要、指标定义与计算口径；保留时间戳与源文件版本。
元数据字典：字段含义、来源、类型、允许值与缺失率；将字典随数据集发布。
合规留痕：记录许可条款、署名要求与衍生作品约束；生成 LICENSE 与 CITATION 建议片段。
过程追踪：清洗脚本、参数与环境信息写入 run.json；便于复现与审计。

4.7 代码示例（可运行，含注释）

以下片段节选自 gis_examples/modules/01_basics/data_quality_pipeline.py，展示标准化、几何校验与质量摘要的关键步骤。

import os
import json
import time
import typing as T
import geopandas as gpd
import pandas as pd
from shapely.geometry import shape# 假设：我们已经通过外部工具/接口获取了原始 GeoJSON/GeoPackage 文件
# 目标：统一 CRS、规范字段、修复几何，并产出质量报告TARGET_CRS = "EPSG:3857"def normalize_schema(gdf: gpd.GeoDataFrame, mapping: dict) -> gpd.GeoDataFrame:"""将不同来源的字段映射为统一规范。参数 mapping 如 {"amenity": "poi_type", "name": "poi_name"}"""gdf = gdf.copy()for src, dst in mapping.items():if src in gdf.columns:gdf[dst] = gdf[src]return gdfdef to_target_crs(gdf: gpd.GeoDataFrame, target: str = TARGET_CRS) -> gpd.GeoDataFrame:# 统一坐标系；若原始 CRS 未定义，需根据数据文档或上下文补全if gdf.crs is None:raise ValueError("源数据未声明 CRS，请检查元数据或手动指定。")return gdf.to_crs(target)def make_valid(geom):# 使用 buffer(0) 简化的合法化；复杂场景可接入更强修复算法try:return geom.buffer(0)except Exception:return geomdef validate_geometries(gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:gdf = gdf.copy()gdf["is_valid"] = gdf.geometry.is_validinvalid = ~gdf["is_valid"]if invalid.any():gdf.loc[invalid, "geometry"] = gdf.loc[invalid, "geometry"].apply(make_valid)gdf["is_valid"] = gdf.geometry.is_validreturn gdfdef quality_summary(gdf: gpd.GeoDataFrame, key_fields: T.List[str]) -> dict:# 统计缺失率、几何合法性比例与字段覆盖率total = len(gdf)valid_ratio = float(gdf["is_valid"].mean()) if "is_valid" in gdf.columns else float((gdf.geometry.is_valid).mean())missing = {k: float(gdf[k].isna().mean()) if k in gdf.columns else 1.0 for k in key_fields}score = 100.0 * (0.5 * valid_ratio + 0.5 * (1.0 - float(pd.Series(missing).mean())))return {"total_records": int(total),"valid_ratio": valid_ratio,"missing_rate": missing,"quality_score": round(score, 2),}def write_quality_report(info: dict, out_path: str):os.makedirs(os.path.dirname(out_path), exist_ok=True)with open(out_path, "w", encoding="utf-8") as f:json.dump(info, f, ensure_ascii=False, indent=2)def run_pipeline(input_path: str, out_dir: str, field_map: dict, key_fields: T.List[str]):gdf = gpd.read_file(input_path)gdf = normalize_schema(gdf, field_map)gdf = to_target_crs(gdf, TARGET_CRS)gdf = validate_geometries(gdf)# 质量摘要summary = quality_summary(gdf, key_fields)summary["source_file"] = os.path.basename(input_path)summary["timestamp"] = time.strftime("%Y-%m-%d %H:%M:%S")# 输出数据与报告os.makedirs(out_dir, exist_ok=True)cleaned_path = os.path.join(out_dir, "cleaned.gpkg")gdf.to_file(cleaned_path, driver="GPKG")write_quality_report(summary, os.path.join(out_dir, "quality.json"))# 额外：生成元数据字典与过程记录
def write_metadata_dict(gdf: gpd.GeoDataFrame, out_path: str):info = {"fields": {col: {"dtype": str(gdf[col].dtype) if col in gdf.columns else None,"unique_sample": list(map(str, gdf[col].dropna().unique()[:10])) if col in gdf.columns else []} for col in gdf.columns if col != "geometry"},"crs": str(gdf.crs),"record_count": int(len(gdf))}os.makedirs(os.path.dirname(out_path), exist_ok=True)with open(out_path, "w", encoding="utf-8") as f:json.dump(info, f, ensure_ascii=False, indent=2)def write_run_trace(out_path: str, **kwargs):os.makedirs(os.path.dirname(out_path), exist_ok=True)with open(out_path, "w", encoding="utf-8") as f:json.dump({"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), **kwargs}, f, ensure_ascii=False, indent=2)if __name__ == "__main__":# 示例执行入口（可根据需要移动到独立脚本）src = "data/osm_poi.geojson"out_dir = "outputs/ch4_quality"fmap = {"amenity": "poi_type", "name": "poi_name"}keys = ["poi_type", "poi_name"]run_pipeline(src, out_dir, fmap, keys)gdf = gpd.read_file(os.path.join(out_dir, "cleaned.gpkg"))write_metadata_dict(gdf, os.path.join(out_dir, "metadata.json"))write_run_trace(os.path.join(out_dir, "run.json"), source=src, target_crs=TARGET_CRS, field_map=fmap)