第18章 - 云原生影像处理流水线

把”扫描件变 COG、COG 变 STAC Item、Item 变可出图服务”这三步连起来，就是一个影像流水线。听起来是三个脚本串起来，实际做起来 80% 的工程量在四件事上：幂等（重复跑不产生重复数据）、校验（坏文件不静默通过）、可观测（失败能定位到哪个文件）、可重入（部分失败能只补跑失败的部分）。

这一章按”一次真实入库”的顺序走完整条链，每一步给出可运行的命令，最后给出失败处理与成本控制的判断标准。

18.1 流水线的六个阶段

  原始数据（TIF/JP2/HDF，可能在 FTP/S3/本地盘）
      │  ① 摄取：下载或直读，重命名成可路由的 key
      ▼
  暂存层（原格式，保留，作为回滚依据）
      │  ② 转换：rio cogeo create → COG（重采样 / 压缩 / overview）
      ▼
  COG 层（s3://bucket/cog/{collection}/{id}.tif）
      │  ③ 校验：rio cogeo validate + 头信息检查
      ▼
  通过校验的 COG
      │  ④ 元数据抽取：rio-stac / pystac 生成 Item
      ▼
  STAC Item（写目录 / 写 pgstac）
      │  ⑤ 索引：生成或更新 MosaicJSON
      ▼
  可出图（TiTiler 读 COG，按 MosaicJSON 或 STAC 检索）
      │  ⑥ 验收：抽样瓦片比对
      ▼
  上线

六个阶段的失败代价完全不同：转换最贵（CPU 密集，单景几分钟），索引最便宜（几秒），校验最应该早做（它挡住所有后续浪费）。所以工程上的顺序通常是先校验输入、再转换、转换后立刻校验、校验通过才写 Item。任何一步失败都不应该污染下游。

18.2 转换：rio cogeo create 的参数

rio cogeo create 是整条流水线里参数最多的一步。完整选项很长，真正影响产出质量与体积的是这九个：

压缩算法怎么选，取决于数据内容与用途：

展示用途用 jpeg 或 webp 能把体积降一个数量级，代价是不能再用作分析源。实践中常见做法是同一份数据出两个版本：cog/analysis/ 用 deflate，cog/display/ 用 jpeg，TiTiler 按用途指向不同目录。

一条完整的转换命令：

rio cogeo create \
  input.tif output.tif \
  --cog-profile deflate \
  --blocksize 512 \
  --overview-resampling average \
  --add-mask \
  --co PREDICTOR=2 --co ZLEVEL=6 \
  --co NUM_THREADS=ALL_CPUS \
  --use-cog-driver

--co PREDICTOR=2 与 ZLEVEL=6 是 deflate 的黄金组合：predictor 2 是水平差分，对连续色调影像特别有效；ZLEVEL 6 是压缩率与速度的平衡点，9 只多 3%～5% 的压缩率却慢一倍。

--overview-resampling average 对连续数据是对的，对分类数据必须用 nearest。这一点和第 14 章的重采样选择规则完全一致——金字塔本质就是一次降采样。

--blocksize 不要盲目设大。512 是 GDAL 与 COG 规范的常用值，1024 能减少每张瓦片的请求数但增加单次读取量。对 256×256 的瓦片输出，512 的块意味着一次瓦片请求通常只碰 1～4 个块，是最优的。

18.3 校验：三层检查

rio cogeo validate 只检查结构（是否 tiled、是否有 overview、IFD 顺序），不检查内容。生产环境三层都要做：

# 1. 结构校验
rio cogeo validate output.tif

# 2. 严格模式，警告也当错误（缺 overview 会被揪出来）
rio cogeo validate --strict output.tif

# 3. 关键头信息，供人工或脚本判断
rio cogeo info output.tif

rio cogeo info 的输出里有几个字段值得写进自动化检查：

COG: True
Compression: DEFLATE
    Tiled:            True
    Dtype:            uint16
    NoData:           0.0
    Alpha Band:       False
    Internal Mask:    False
IFD
    Id      Size           BlockSize     Decimation
    0       10980x10980    512x512       0
    1       5490x5490      128x128       2

检查项：COG 必须为 True；Tiled 必须为 True；Internal Mask 或 Alpha Band 至少要有一个（否则透明区会变黑）；IFD 的 Decimation 必须是 2 的幂递增（否则金字塔层级错乱）；最底层 overview 的尺寸应小于 blocksize（否则等于白建）。

第三层是内容校验：抽样比对原图与 COG 的统计值。压缩是有损的时候（jpeg/webp），必须比对；无损时可以只比对 band 数与 dtype。

import rasterio
from rasterio.enums import Resampling


def sanity_check(src_path: str, cog_path: str) -> dict:
    """比对源与 COG 的元数据和抽样统计。"""
    with rasterio.open(src_path) as src, rasterio.open(cog_path) as dst:
        assert dst.profile.get("tiled"), "输出未分块"
        assert dst.overviews(1), "缺少 overview"
        assert src.count == dst.count, "波段数不一致"

        # 用 overview 读一个低分辨率版本做统计比对
        scale = max(1, dst.width // 512)
        arr = dst.read(
            1,
            out_shape=(dst.height // scale, dst.width // scale),
            resampling=Resampling.average,
        )
        valid = arr[arr != (dst.nodata or 0)]
        return {
            "dtype": str(dst.dtypes[0]),
            "nodata": dst.nodata,
            "overview_levels": len(dst.overviews(1)),
            "mean_sampled": float(valid.mean()),
        }

arr[arr != nodata].mean() 这个写法在有 nodata 时是必要的，直接把 nodata 值算进均值会让结果完全失真——nodata 常写成 0 或 -9999，两者都会把均值拉偏。

18.4 元数据：从 COG 生成 STAC Item

rio-stac 提供了 create_stac_item()，能把一个 COG 的投影、栅格、EO 信息写成标准 Item。TiTiler 的 /cog/stac 端点就是它的 HTTP 封装。

from rio_stac.stac import create_stac_item

item = create_stac_item(
    "s3://bucket/cog/s2/2024/01/S2A_50TMK_20240101.tif",
    id="S2A_50TMK_20240101",
    collection="sentinel-2-l2a",
    input_datetime="2024-01-01T03:12:00Z",
    asset_name="visual",
    asset_media_type="image/tiff; application=geotiff; profile=cloud-optimized",
    with_proj=True,
    with_raster=True,
    with_eo=True,
    raster_max_size=1024,
)

item.set_self_href(
    "https://catalog.example.com/collections/sentinel-2-l2a/items/S2A_50TMK_20240101"
)
print(item.to_dict()["properties"]["proj:epsg"])

asset_media_type 必须写对。第 16 章提过，STACReader 用媒体类型白名单筛资产，写成随意字符串会导致 /stac/assets 里看不到它。用 rio-stac 的默认值比自己手写可靠。

set_self_href() 不是可选的。相对路径的资产 href 要靠它解析，缺了就会出现”Item 能读、瓦片全黑”——这和第 16 章的排错项是同一条。

生成后要过一遍 schema 校验，stac-validator 或 pystac 的 validate_all() 都行。校验能挡掉 80% 的”看起来对但读不出来”的问题：必填字段缺失、bbox 与 geometry 不一致、assets 里没有 href。

18.5 编排：三种形态与选择

流水线的编排形态决定了单景成本和运维复杂度：

Lambda 的 15 分钟硬上限是关键约束。一个 10980×10980 的 4 波段 Sentinel-2 场景，rio cogeo create 在 1 vCPU 上约 3～5 分钟，2 vCPU 上 2 分钟左右；但 30000×30000 的航拍正射影像轻松超过 15 分钟。判断标准很简单：在目标实例上转换时间的 P95 是否小于上限的一半。小于就用 Lambda，否则用容器任务。

Lambda 的另一个约束是 /tmp 只有 512 MB（可配到 10 GB）和内存上限 10 GB。COG 转换是流式的，内存占用通常是源文件解压后的一块，不大；但中间文件写到 /tmp 时要注意清理，否则同一执行环境复用会撑爆。

一个最小的幂等脚本，核心是三件事：检查输出已存在则跳过、失败写标记、成功写清单：

import json
import subprocess
from pathlib import Path


def convert_one(src: Path, dst: Path, manifest: Path) -> bool:
    """转换单个文件，幂等。返回是否真的执行了转换。"""
    if dst.exists():
        print(f"skip {dst.name}: already exists")
        return False

    tmp = dst.with_suffix(".tmp.tif")
    cmd = [
        "rio", "cogeo", "create", str(src), str(tmp),
        "--cog-profile", "deflate",
        "--overview-resampling", "average",
        "--add-mask",
        "--co", "PREDICTOR=2",
        "--co", "ZLEVEL=6",
    ]
    subprocess.run(cmd, check=True)

    # 先校验再改名，避免半成品被当成成品
    subprocess.run(["rio", "cogeo", "validate", "--strict", str(tmp)], check=True)
    tmp.rename(dst)

    with manifest.open("a", encoding="utf-8") as f:
        f.write(json.dumps({"src": str(src), "dst": str(dst)}) + "\n")
    return True

tmp.rename(dst) 这一步是幂等的关键。直接把输出写成目标文件名，转换失败会留下半成品，下次运行看到文件存在就跳过了——一个损坏的 COG 就这样进了库。先写临时名、校验通过再改名，是唯一可靠的顺序。

18.6 索引更新

COG 入完，MosaicJSON 要更新。全量重建简单但耗时，增量更新快但要处理并发。

# 全量重建：适合每天一次、文件数在万级以内
find /data/cog -name "*.tif" > list.txt
cogeo-mosaic create list.txt -o s3://bucket/mosaic/ortho.json.gz \
  --minzoom 8 --maxzoom 16 --quadkey-zoom 10 --min-tile-cover 5

# 增量更新：把新文件追加进已有 mosaic
cogeo-mosaic update new_files.txt -o s3://bucket/mosaic/ortho.json.gz

update 的行为要特别小心：它读已有 mosaic、合并新文件、写回。两个人同时跑，后写的会覆盖先写的。生产环境要用锁（比如 S3 的条件写、DynamoDB 的乐观锁），或者干脆串行化到一个任务里。

--min-tile-cover 5 在增量更新时尤其重要：新文件往往是边缘补片，覆盖率低，不设阈值会让 MosaicJSON 迅速膨胀。

18.7 失败重试与断点续跑

批量入库几乎不可能一次全成功：源文件损坏、网络抖一下、对象存储返回 503。要把重跑做成常态，而不是事故处理。三个机制缺一不可。

按文件记录状态，而不是按批次。 批次级的“成功/失败”粒度太粗，一景坏掉整批重跑。改成每个文件一行状态记录，重跑时只处理非 done 的行：

import json
import time
from pathlib import Path


def load_state(path: Path) -> dict[str, dict]:
    if not path.exists():
        return {}
    rows = path.read_text(encoding="utf-8").splitlines()
    return {r["key"]: r for r in map(json.loads, rows)}


def retry_pending(src_dir: Path, cog_dir: Path, state_path: Path, max_attempts: int = 3) -> None:
    """只重跑未完成、且失败次数未超限的文件。"""
    state = load_state(state_path)
    for src in sorted(src_dir.glob("*.tif")):
        key = src.stem
        rec = state.get(key, {"key": key, "status": "pending", "attempts": 0})
        if rec["status"] == "done" or rec["attempts"] >= max_attempts:
            continue
        rec["attempts"] += 1
        try:
            convert_one(src, cog_dir / f"{key}.tif", state_path)
            rec["status"] = "done"
        except Exception as exc:  # noqa: BLE001
            rec["status"] = "failed"
            rec["error"] = f"{type(exc).__name__}: {exc}"
        rec["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        with state_path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")

状态文件用 JSONL 追加写，是因为它能被 tail、被 grep、被并发追加（POSIX 下单行写入小于管道缓冲时是原子的）。读的时候按 key 取最后一条，就得到最新状态。

区分可重试与不可重试的错误。 全量重试是浪费，全部不重试是掩盖问题：

指数退避的最小实现：

import time


def with_retry(fn, attempts: int = 3, base: float = 2.0):
    """对瞬态错误做指数退避重试。"""
    last: Exception | None = None
    for i in range(attempts):
        try:
            return fn()
        except (ConnectionError, TimeoutError) as exc:
            last = exc
            time.sleep(base ** i)
    raise last  # type: ignore[misc]

把失败信息落到可查询的地方。 只写日志的话，几百个失败文件要翻半小时。把 status、error、attempts 写进状态文件后，两行命令就能汇总：

# 按状态汇总
jq -r '.status' state.jsonl | sort | uniq -c

# 只对失败的按错误类型汇总
jq -r 'select(.status=="failed") | .error' state.jsonl | sort | uniq -c | sort -rn

断点续跑的前提是幂等，而幂等的前提是“半成品不会冒充成品”。18.5 节的“临时名 → 校验 → 改名”就是为此：重跑时看到的目标文件一定完整，直接跳过是安全的。若流水线直接写目标文件名，重跑就得先删掉可疑文件，否则会把坏文件当成功。

最后一条经验：别在失败时立即重试。对象存储的 503 往往持续几秒到几十秒，立即重试只是把同一个错误重复三次。退避 1/2/4 秒起步、最多 3 次，成功率远高于“立刻重试 10 次”。

18.8 排错

转换成功但 TiTiler 返回 500，日志是 RasterioIOError。 最常见的是权限：写 COG 的凭证和读 COG 的凭证不是同一个。TiTiler 的容器角色需要 s3:GetObject 与 s3:GetObjectAttributes，少一个就会 403 再被包装成 500。

rio cogeo validate 报 The file is not a valid COG，但 QGIS 能打开。 validate 检查的是 IFD 顺序与分块，一些老 TIF 能被 GDAL 读但不满足 COG 规范。这类文件必须重新转换，”能打开就上线”是债。

overview 没生成。 --overview-resampling nearest 配 --blocksize 1024 时，如果源图小于 2 倍 blocksize，就没有可建的 overview 层级。检查源图尺寸与 --overview-level，必要时显式指定层数。

转换后体积反而变大。 源文件已经是 JPEG 压缩的 TIF，再用 deflate 重新压一遍，压缩率不如原来的 JPEG。这种情况该用 --cog-profile jpeg 保持有损压缩，或者直接跳过转换（如果源文件已经是合规 COG）。

批量任务跑到一半 OOM。 rio cogeo create 默认在小于 1.2 亿像素时全内存处理，超过就分块。容器内存小于 4 GB 时对超大文件要加 --no-in-memory，并确保 CPL_TMPDIR 指向有空间的目录。

Item 写进去了但搜不到。 pgstac 的 ingest 有延迟（分区创建、索引重建），刚写完立即检索可能为空。等 10～30 秒再查，或者查 /collections/{id}/items 而不是 /search（前者走主键）。

18.9 生产建议

• 原始文件永久保留。COG 是派生产物，转换参数改了要重跑。原始文件删了就再也回不去，这是最贵的一类错误。
• 转换参数写进文件元数据。在 COG 的 TIFF tag 或同名 sidecar JSON 里记下 --cog-profile、--blocksize、--overview-resampling，半年后排查”为什么这批图比那批大”时能立刻回答。
• 按 collection 分目录，不要在同一个前缀下混放。s3://bucket/cog/{collection}/{year}/{month}/{id}.tif 这个结构同时满足按前缀列举、按生命周期规则清理、按 collection 做权限。
• 幂等靠”临时名 + 校验 + 改名”，不靠数据库状态。数据库会漂移，文件系统不会。
• 把校验失败的文件单独归档，不要直接丢弃。失败样本是改进转换参数的最好素材，攒上几十个就能看出规律（比如某个传感器的特定压缩方式）。

18.10 本章要点

下一章把结果交给前端：MapLibre 与 Leaflet 怎么接 TiTiler 的瓦片、tileSize 与 scheme 怎么对齐、跨域和鉴权头怎么配。