Survive Contabo PUT throttling in the lake extractor
All checks were successful
PR build (required check) / build-frontend (pull_request) Has been skipped
PR build (required check) / gate (pull_request) Successful in 3s
PR build (required check) / validate-observability (pull_request) Has been skipped
secrets-guard / encrypted (pull_request) Successful in 6s
PR build (required check) / changes (pull_request) Successful in 8s
PR build (required check) / build-backend (pull_request) Successful in 1m3s
All checks were successful
PR build (required check) / build-frontend (pull_request) Has been skipped
PR build (required check) / gate (pull_request) Successful in 3s
PR build (required check) / validate-observability (pull_request) Has been skipped
secrets-guard / encrypted (pull_request) Successful in 6s
PR build (required check) / changes (pull_request) Successful in 8s
PR build (required check) / build-backend (pull_request) Successful in 1m3s
Contabo answers 429 to PUT bursts; at 16 concurrent uploads botocore's default 4 retries ran out and the run died at tile 54/390. Adaptive retry mode (client-side rate limiting, 10 attempts), a halved upload pool, and per-tile failure isolation (a failed tile logs and is retried next run — it never reached the manifest) keep the run alive through throttling.
This commit is contained in:
parent
8e09155aaf
commit
33157766cb
1 changed files with 24 additions and 13 deletions
|
|
@ -65,11 +65,16 @@ class _Sink:
|
||||||
self._s3 = None
|
self._s3 = None
|
||||||
if not self.local:
|
if not self.local:
|
||||||
import boto3 # noqa: PLC0415 - extractor-only dep
|
import boto3 # noqa: PLC0415 - extractor-only dep
|
||||||
|
from botocore.config import Config # noqa: PLC0415
|
||||||
|
# Contabo throttles bursts with 429s (seen live at 16 concurrent
|
||||||
|
# PUTs); adaptive mode rate-limits client-side and retries with
|
||||||
|
# backoff instead of dying after botocore's default 4 attempts.
|
||||||
self._s3 = boto3.client(
|
self._s3 = boto3.client(
|
||||||
"s3", endpoint_url=self.cfg["endpoint"],
|
"s3", endpoint_url=self.cfg["endpoint"],
|
||||||
region_name=self.cfg["region"],
|
region_name=self.cfg["region"],
|
||||||
aws_access_key_id=self.cfg["access_key"],
|
aws_access_key_id=self.cfg["access_key"],
|
||||||
aws_secret_access_key=self.cfg["secret_key"])
|
aws_secret_access_key=self.cfg["secret_key"],
|
||||||
|
config=Config(retries={"max_attempts": 10, "mode": "adaptive"}))
|
||||||
|
|
||||||
def put(self, key: str, df: pl.DataFrame) -> None:
|
def put(self, key: str, df: pl.DataFrame) -> None:
|
||||||
if self.local:
|
if self.local:
|
||||||
|
|
@ -147,7 +152,7 @@ def _write_tile(sink: _Sink, ti: int, tj: int, tile_df: pl.DataFrame) -> pl.Data
|
||||||
for (year, month), mdf in parts.group_by(["year", "month"]):
|
for (year, month), mdf in parts.group_by(["year", "month"]):
|
||||||
puts.append((era5lake.daily_part_key(ti, tj, year, month),
|
puts.append((era5lake.daily_part_key(ti, tj, year, month),
|
||||||
mdf.drop(["year", "month"]).sort(["lat_idx", "lon_idx", "date"])))
|
mdf.drop(["year", "month"]).sort(["lat_idx", "lon_idx", "date"])))
|
||||||
with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool:
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
||||||
for f in [pool.submit(sink.put, k, df) for k, df in puts]:
|
for f in [pool.submit(sink.put, k, df) for k, df in puts]:
|
||||||
f.result() # surface the first failure instead of hiding it
|
f.result() # surface the first failure instead of hiding it
|
||||||
return pl.DataFrame(rows)
|
return pl.DataFrame(rows)
|
||||||
|
|
@ -197,29 +202,35 @@ def main() -> None:
|
||||||
done = set() if manifest is None or args.overwrite else \
|
done = set() if manifest is None or args.overwrite else \
|
||||||
set(manifest["tile"].unique().to_list())
|
set(manifest["tile"].unique().to_list())
|
||||||
|
|
||||||
written = skipped = empty = 0
|
written = skipped = empty = failed = 0
|
||||||
for n, (ti, tj) in enumerate(todo, 1):
|
for n, (ti, tj) in enumerate(todo, 1):
|
||||||
if f"{ti}_{tj}" in done:
|
if f"{ti}_{tj}" in done:
|
||||||
skipped += 1
|
skipped += 1
|
||||||
continue
|
continue
|
||||||
t0 = time.time()
|
t0 = time.time()
|
||||||
tile_df = _tile_frame(ds, ti, tj, end)
|
try:
|
||||||
if tile_df is None:
|
tile_df = _tile_frame(ds, ti, tj, end)
|
||||||
empty += 1
|
if tile_df is None:
|
||||||
|
empty += 1
|
||||||
|
continue
|
||||||
|
if args.dry_run:
|
||||||
|
print(f"[dry-run] tile {ti}_{tj}: {tile_df.height} rows "
|
||||||
|
f"({time.time() - t0:.1f}s); nothing written")
|
||||||
|
print(tile_df.head())
|
||||||
|
return
|
||||||
|
rows = _write_tile(sink, ti, tj, tile_df)
|
||||||
|
except Exception as e: # noqa: BLE001 - keep going; the tile isn't in the
|
||||||
|
failed += 1 # manifest, so the next run retries it
|
||||||
|
print(f"[{n}/{len(todo)}] FAILED tile {ti}_{tj}: {e}")
|
||||||
continue
|
continue
|
||||||
if args.dry_run:
|
|
||||||
print(f"[dry-run] tile {ti}_{tj}: {tile_df.height} rows "
|
|
||||||
f"({time.time() - t0:.1f}s); nothing written")
|
|
||||||
print(tile_df.head())
|
|
||||||
return
|
|
||||||
rows = _write_tile(sink, ti, tj, tile_df)
|
|
||||||
manifest = rows if manifest is None else pl.concat(
|
manifest = rows if manifest is None else pl.concat(
|
||||||
[manifest.filter(pl.col("tile") != f"{ti}_{tj}"), rows])
|
[manifest.filter(pl.col("tile") != f"{ti}_{tj}"), rows])
|
||||||
sink.put(era5lake.MANIFEST_KEY, manifest) # after every tile: resumable
|
sink.put(era5lake.MANIFEST_KEY, manifest) # after every tile: resumable
|
||||||
written += 1
|
written += 1
|
||||||
print(f"[{n}/{len(todo)}] tile {ti}_{tj}: {rows.height} points "
|
print(f"[{n}/{len(todo)}] tile {ti}_{tj}: {rows.height} points "
|
||||||
f"in {time.time() - t0:.1f}s")
|
f"in {time.time() - t0:.1f}s")
|
||||||
print(f"done: tiles written={written} skipped(done)={skipped} sea={empty}")
|
print(f"done: tiles written={written} skipped(done)={skipped} sea={empty} "
|
||||||
|
f"failed={failed}")
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue