diff --git a/pipelines/datasets/us_dol_oflc/constants.py b/pipelines/datasets/us_dol_oflc/constants.py index 48ee677354..7e495d3ca8 100644 --- a/pipelines/datasets/us_dol_oflc/constants.py +++ b/pipelines/datasets/us_dol_oflc/constants.py @@ -32,13 +32,23 @@ class constants(Enum): IMPERSONATE = "chrome" BASE_URL = "https://www.dol.gov" - # Link text on the performance page is inconsistent, so files are found by - # matching the href against these patterns, one per program. + # Files are found by matching the published file name against these + # patterns, one per program. They are anchored and deliberately narrow: the + # performance page publishes companion workbooks beside each disclosure file + # — LCA Appendix A and Worksites, the H-2A Addendums, the H-2B Appendixes — + # which carry different layouts and, under a looser pattern, collapse onto + # the same local name and overwrite the file we actually want. + # + # Only the case-level disclosure file is matched, and only in the modern + # naming the pipeline ever sees: the historical layouts (H-1B_Case_Data_FY2008, + # Icert_ LCA_ FY2009, LCA_FY2012_Q4 …) were onboarded once and are never + # re-fetched, because a run only ever touches the open fiscal year and the + # one before it. FILE_PATTERNS = { - "lca": r"(LCA|H-1B|H1B|Icert)[_ ].*(Disclosure|Case_Data|iCert|FY)", - "perm": r"PERM.*(Disclosure|FY)", - "h2a": r"H-?2A.*(Disclosure|FY)", - "h2b": r"H-?2B.*(Disclosure|FY)", + "lca": r"^LCA_Disclosure_Data_FY_?\d{4}(_Q[1-4])?\.xlsx?$", + "perm": r"^PERM_Disclosure_Data_(New_Form_)?FY_?\d{4}(_Q[1-4])?\.xlsx?$", + "h2a": r"^H-?2A_Disclosure_Data_FY_?\d{4}(_Q[1-4])?(_(new|old)_form)?\.xlsx?$", + "h2b": r"^H-?2B_Disclosure_(Data_)?FY_?\d{4}(_Q[1-4])?\.xlsx?$", } # A fiscal year is frozen once its final file has landed and the year is diff --git a/pipelines/datasets/us_dol_oflc/flows.py b/pipelines/datasets/us_dol_oflc/flows.py index 326e616e00..2239a33b63 100644 --- a/pipelines/datasets/us_dol_oflc/flows.py +++ b/pipelines/datasets/us_dol_oflc/flows.py @@ -198,6 +198,15 @@ def us_dol_oflc_flow( {"cron": "23 14 5,12,19,26 2,5,8,11 *", "timezone": "America/Sao_Paulo"} ] # The clean step holds one fiscal year of LCA (~700k rows x 57 columns) in -# pandas while it is written. +# pandas while it is written, and calamine builds a Python object per cell of +# the workbook it is reading. +# +# The keys are memory_limit / memory_request: this work pool's job template +# defines no "memory" variable, so the {"memory": "8Gi"} spelling used by most +# datasets in this repo is silently discarded and the pod runs at the pool +# default of 4Gi. That is what OOM-killed the first full run here. # pyrefly: ignore [missing-attribute] -us_dol_oflc_flow.job_variables = {"memory": "8Gi"} +us_dol_oflc_flow.job_variables = { + "memory_limit": "12Gi", + "memory_request": "4Gi", +} diff --git a/pipelines/datasets/us_dol_oflc/utils.py b/pipelines/datasets/us_dol_oflc/utils.py index 7b2499c5d7..30bea243fb 100644 --- a/pipelines/datasets/us_dol_oflc/utils.py +++ b/pipelines/datasets/us_dol_oflc/utils.py @@ -40,6 +40,16 @@ "prevailing_wage_annual", } + +class UnknownLayoutError(RuntimeError): + """A source workbook whose column layout the crosswalk does not describe. + + Raised rather than ``SystemExit`` so a Prefect run reports it as Failed — + an ordinary data problem — instead of Crashed, which reads like the + infrastructure died. + """ + + # Values the source uses for "blank". NULLISH = {"", "NA", "N/A", "NULL", "NONE", "UNKNOWN", "-", "--", "."} @@ -125,7 +135,7 @@ def _to_date(v: object) -> str | None: def load_crosswalk(program: str) -> dict[tuple[int, str], dict[str, str]]: - """(fiscal_year, source_file) -> {source_column: canonical_column}.""" + """(fiscal_year, local_file) -> {source_column: canonical_column}.""" out: dict[tuple[int, str], dict[str, str]] = defaultdict(dict) with open(CROSSWALK_DIR / f"{program}.csv") as fh: for row in csv.DictReader(fh): @@ -137,6 +147,50 @@ def load_crosswalk(program: str) -> dict[tuple[int, str], dict[str, str]]: return out +def load_crosswalk_headers( + program: str, +) -> dict[frozenset[str], dict[str, str]]: + """Header signature -> mapping, for files the crosswalk knows by layout. + + The crosswalk is keyed on the file name the onboarding run happened to give + each workbook, but the recurring pipeline derives its own names from the + published file names, and the two do not always agree — the FY2025 LCA Q4 + file is ``lca_2025.xlsx`` in the crosswalk and ``lca_2025q4.xlsx`` when the + pipeline downloads it. + + Matching on the set of source columns instead removes that coupling + entirely, and it is the more meaningful key: what determines how a workbook + is read is its layout, not its name. A new quarterly file with an unchanged + layout therefore resolves on its own, while a genuine form revision still + finds no match and fails loudly, which is what the crosswalk is for. + """ + by_file: dict[tuple[int, str], set[str]] = defaultdict(set) + with open(CROSSWALK_DIR / f"{program}.csv") as fh: + for row in csv.DictReader(fh): + if row["source_column"]: + by_file[(int(row["fiscal_year"]), row["source_file"])].add( + row["source_column"] + ) + mapped = load_crosswalk(program) + return { + frozenset(columns): mapped[key] + for key, columns in by_file.items() + if key in mapped + } + + +def read_header(path: Path) -> list[str]: + """The header row of a workbook, without materialising the rest of it. + + ``to_python()`` builds a Python object per cell, so reading a 437k x 98 + workbook to look at one row costs gigabytes — enough to OOM the worker. The + layout pre-flight only needs the header, so it reads only the header. + """ + ws = pc.CalamineWorkbook.from_path(str(path)).get_sheet_by_index(0) + rows = ws.to_python(nrows=1) + return [str(c).strip() for c in rows[0]] if rows else [] + + def read_sheet(path: Path) -> tuple[list[str], list[list]]: ws = pc.CalamineWorkbook.from_path(str(path)).get_sheet_by_index(0) rows = ws.to_python() @@ -174,13 +228,20 @@ def read_file( order: list[str], types: dict[str, str], xw, + by_header, unknown_units, ) -> pd.DataFrame: """One source workbook as a canonical-schema DataFrame.""" + header, rows = read_sheet(path) mapping = xw.get((fy, path.name)) if not mapping: - raise SystemExit(f"No crosswalk entry for {path.name} (FY{fy})") - header, rows = read_sheet(path) + mapping = by_header.get(frozenset(header)) + if not mapping: + raise UnknownLayoutError( + f"No crosswalk entry for {path.name} (FY{fy}) and its column layout " + f"matches no known layout for {program}. Rebuild the crosswalk with " + f"build_crosswalk.py and review what changed." + ) idx = {col: i for i, col in enumerate(header)} data: dict[str, list] = {} for src, canon in mapping.items(): @@ -260,6 +321,7 @@ def build( order = [c for c, _ in spec] types = dict(spec) xw = load_crosswalk(program) + by_header = load_crosswalk_headers(program) unknown_units: dict[str, int] = defaultdict(int) typed = pa.schema( @@ -290,7 +352,9 @@ def build( print(f" FY{fy}: already written, skipping", flush=True) continue frames = [ - read_file(p, fy, program, order, types, xw, unknown_units) + read_file( + p, fy, program, order, types, xw, by_header, unknown_units + ) for p in paths ] df = ( @@ -429,11 +493,25 @@ def fiscal_year_of(name: str) -> int | None: def local_name(program: str, name: str) -> str: - """Local file name for a source workbook, matching the crosswalk key.""" + """Local file name for a source workbook. + + Two source files must never collapse onto one name — that silently replaces + one with the other. A fiscal year can legitimately be published as several + files: one per quarter, and in a form-transition year one per form version + (PERM FY2024, H-2A FY2025), so both are carried into the name. + + The crosswalk is resolved by column layout rather than by this name, so the + name only has to be unique, not to match anything. + """ fy = fiscal_year_of(name) quarter = re.search(r"_Q([1-4])", name, re.I) - suffix = f"q{quarter.group(1)}" if quarter and fy and fy >= 2020 else "" - return f"{program}_{fy}{suffix}{Path(name).suffix}" + parts = [program, str(fy)] + if quarter and fy and fy >= 2020: + parts.append(f"q{quarter.group(1)}") + form = re.search(r"(new|old)[_ ]form", name, re.I) + if form: + parts.append(form.group(1).lower()) + return "".join([parts[0], "_", "".join(parts[1:])]) + Path(name).suffix # -------------------------------------------------------------------------- @@ -476,10 +554,23 @@ def download_fiscal_years( session = _session() input_dir.mkdir(parents=True, exist_ok=True) got: list[Path] = [] - for name, url in sorted(list_source_files(program).items()): - fy = fiscal_year_of(name) - if fy not in years: - continue + wanted = { + name: url + for name, url in sorted(list_source_files(program).items()) + if fiscal_year_of(name) in years + } + # A collision would silently replace one source file with another, so it is + # an error rather than something to resolve by ordering. + names: dict[str, str] = {} + for name in wanted: + local = local_name(program, name) + if local in names: + raise UnknownLayoutError( + f"{program}: {name} and {names[local]} both map to {local}. " + f"Two source files cannot share one local name." + ) + names[local] = name + for name, url in wanted.items(): dest = input_dir / local_name(program, name) if dest.exists() and dest.stat().st_size > 10_000: got.append(dest)