memory: release Arrow buffers between symbols, smaller DuckDB limit, faster stale requeue
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
4 changed files +21 −7
modified
services/ingest/src/findata_ingest/queue.py
+1 −1
@@ -91,7 +91,7 @@ def get_checkpoint(scope: str, key: str) -> dict[str, Any] | None: | |||||
| 91 | return r["value"] if r else None | 91 | return r["value"] if r else None | ||
| 92 | 92 | ||||
| 93 | 93 | ||||
| 94 | −def requeue_stale(max_age_minutes: int = 240) -> int: | 94 | +def requeue_stale(max_age_minutes: int = 90) -> int: | ||
| 95 | """Jobs left 'running' by a crashed worker are re-queued (idempotent tasks + checkpoints make this safe).""" | 95 | """Jobs left 'running' by a crashed worker are re-queued (idempotent tasks + checkpoints make this safe).""" | ||
| 96 | with transaction() as conn: | 96 | with transaction() as conn: | ||
| 97 | r = conn.execute("UPDATE ingest_jobs SET status='queued', locked_by=NULL, next_run_at=now() WHERE status='running' " | 97 | r = conn.execute("UPDATE ingest_jobs SET status='queued', locked_by=NULL, next_run_at=now() WHERE status='running' " | ||
modified
services/ingest/src/findata_ingest/reader.py
+1 −1
@@ -20,7 +20,7 @@ from findata_ingest.lake import Lake, live_files | |||||
| 20 | 20 | ||||
| 21 | 21 | ||||
| 22 | def _con(read_only: bool = True) -> duckdb.DuckDBPyConnection: | 22 | def _con(read_only: bool = True) -> duckdb.DuckDBPyConnection: | ||
| 23 | − con = duckdb.connect(database=":memory:", config={"threads": 8, "memory_limit": "8GB", "TimeZone": "UTC"}) | 23 | + con = duckdb.connect(database=":memory:", config={"threads": 4, "memory_limit": "4GB", "TimeZone": "UTC"}) | ||
| 24 | return con | 24 | return con | ||
| 25 | 25 | ||||
| 26 | 26 | ||||
modified
services/ingest/src/findata_ingest/tasks/backfill.py
+12 −0
@@ -27,6 +27,15 @@ from findata_ingest.tasks.common import ( | |||||
| 27 | ) | 27 | ) | ||
| 28 | 28 | ||||
| 29 | 29 | ||||
| 30 | +def _release_memory() -> None: | ||||
| 31 | + """Return freed Arrow/Polars buffers to the OS between symbols (RSS otherwise grows to several GB per worker).""" | ||||
| 32 | + import contextlib | ||||
| 33 | + import gc | ||||
| 34 | + gc.collect() | ||||
| 35 | + with contextlib.suppress(Exception): | ||||
| 36 | + pa.default_memory_pool().release_unused() | ||||
| 37 | + | ||||
| 38 | + | ||||
| 30 | def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | 39 | def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | ||
| 31 | p = job.get("params") or {} | 40 | p = job.get("params") or {} | ||
| 32 | asset = p.get("type", "stock") | 41 | asset = p.get("type", "stock") | ||
@@ -109,6 +118,9 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 109 | stats["files"] += len(written) | 118 | stats["files"] += len(written) | ||
| 110 | stats["anomalies"] += len(rep.anomalies) | 119 | stats["anomalies"] += len(rep.anomalies) | ||
| 111 | done.add(batch.symbol) | 120 | done.add(batch.symbol) | ||
| 121 | + del t, rep, written | ||||
| 122 | + if stats["symbols"] % 10 == 0: | ||||
| 123 | + _release_memory() | ||||
| 112 | queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes, | 124 | queue.set_checkpoint(f"job:{job['id']}", "state", {"raw_path": raw_rel, "sha256": res.sha256, "bytes": res.bytes, | ||
| 113 | "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)}) | 125 | "fetched_at": res.fetched_at.isoformat(), "done_symbols": sorted(done)}) | ||
| 114 | queue.progress(job["id"], {"stage": "publish", **stats, "last_symbol": batch.symbol}) | 126 | queue.progress(job["id"], {"stage": "publish", **stats, "last_symbol": batch.symbol}) | ||
modified
services/ingest/src/findata_ingest/tasks/derive.py
+7 −5
@@ -58,15 +58,17 @@ def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]: | |||||
| 58 | written.append(lake.write_table(dst, ty, "base", {"symbol": s, "year": str(y)})) | 58 | written.append(lake.write_table(dst, ty, "base", {"symbol": s, "year": str(y)})) | ||
| 59 | retire.extend(o["path"] for o in old) | 59 | retire.extend(o["path"] for o in old) | ||
| 60 | rows += agg.num_rows | 60 | rows += agg.num_rows | ||
| 61 | − tables[s] = agg | 61 | + with transaction() as conn: | ||
| 62 | + upsert_coverage(conn, dst, agg, target) | ||||
| 63 | + del t, agg | ||||
| 64 | + import gc | ||||
| 65 | + gc.collect() | ||||
| 66 | + pa.default_memory_pool().release_unused() | ||||
| 62 | if not written: | 67 | if not written: | ||
| 63 | return {"rows": 0, "symbols": len(symbols)} | 68 | return {"rows": 0, "symbols": len(symbols)} | ||
| 64 | pub = lake.publish(dst, written, sorted(set(retire)), run_id, note=f"derive {target.value} (base) from {src} for {len(symbols)} symbols") | 69 | pub = lake.publish(dst, written, sorted(set(retire)), run_id, note=f"derive {target.value} (base) from {src} for {len(symbols)} symbols") | ||
| 65 | − with transaction() as conn: | ||||
| 66 | − for t in tables.values(): | ||||
| 67 | − upsert_coverage(conn, dst, t, target) | ||||
| 68 | log.info("derive.done", rows=rows, files=len(written), **pub) | 70 | log.info("derive.done", rows=rows, files=len(written), **pub) | ||
| 69 | − return {"rows": rows, "files": len(written), "symbols": len(tables), **pub} | 71 | + return {"rows": rows, "files": len(written), "symbols": len(symbols), **pub} | ||
| 70 | for d in dates: | 72 | for d in dates: | ||
| 71 | t, _ = read_bars(src, all_syms, d, d, lake=lake) | 73 | t, _ = read_bars(src, all_syms, d, d, lake=lake) | ||
| 72 | if t.num_rows == 0: | 74 | if t.num_rows == 0: | ||
| 73 | 75 | ||||