Run data prep with new nationwide history file and visualize results.

Author: njhenry

scripts/osm_data/data_viz.py

@@ -185,9 +185,13 @@ def get_preds_df(model_stub: str, subset: str | None = None) -> Path:
         )
         fig_save(fig, stub = f"osm_changes_{TAG_KEY}_all_preds")
 
-    # Create multi-panel plots for the top tags in each OSM category
+    # Create multi-panel plots for the top tags in each OSM category. Only
+    # iterate keys that are actually columns in the observations CSV — the
+    # snapshot pipeline's filter_keys list can expand without the observations
+    # CSV being regenerated, so we defend against the mismatch here.
     TOP_N_TYPES = config.get("osm_data", "top_n_types")
-    for subtype in OSM_KEYS:
+    present_keys = [k for k in OSM_KEYS if k in to_plot_df.columns]
+    for subtype in present_keys:
         fig = change_multiplot_create(
             observations = to_plot_df,
             col = subtype,

scripts/osm_data/format_tabular.py

@@ -8,6 +8,13 @@
 value, the timestamps of the previous tag assignment and the current
 observation, and a flag for whether the tag changed.
 
+At US scale the versions + changes Parquets together exceed typical RAM, so
+this script uses ``format_observations_duckdb`` to pivot changes wide, LEFT
+JOIN them against versions, and ORDER BY (type, id, version) inside DuckDB,
+spilling the sort to disk as needed. A Python scan then runs the per-POI
+state machine over the sorted row stream and writes observations directly to
+the output CSV — peak RSS stays bounded to roughly ``DUCKDB_MEMORY_LIMIT``.
+
 Config keys used (config.yaml):
     directories.osm_data          — directory containing input and output files
     download.osm.filter_keys      — all tag keys collected (passed as keep_keys)
@@ -24,10 +31,9 @@
         plus all keep_keys columns for grouping
 """
 
-import pandas as pd
 from config_versioned import Config
 
-from openpois.osm.format_observations import format_observations
+from openpois.osm.format_observations import format_observations_duckdb
 
 
 # ----------------------------------------------------------------------------------------
@@ -40,25 +46,29 @@
 OSM_KEYS = config.get("download", "osm", "filter_keys")
 TAG_KEY = config.get("osm_data", "tag_key")
 
+CHANGES_PATH = config.get_file_path("osm_data", "osm_changes")
+VERSIONS_PATH = config.get_file_path("osm_data", "osm_versions")
+OUT_PATH = SAVE_DIR / f"osm_observations_{TAG_KEY}.csv"
+
+# DuckDB execution limits. The sort operator spills past memory_limit so this
+# caps peak RAM independent of input size. Threads default to os.cpu_count()
+# when left as None.
+DUCKDB_MEMORY_LIMIT = "4GB"
+DUCKDB_THREADS = None
+
 
 # ----------------------------------------------------------------------------------------
 # Main workflow
 # ----------------------------------------------------------------------------------------
 
 if __name__ == "__main__":
-    # Read files
-    changes_df = pd.read_parquet(config.get_file_path("osm_data", "osm_changes"))
-    versions_df = pd.read_parquet(config.get_file_path("osm_data", "osm_versions"))
-
-    # Format changes and versions into observations
-    observations_df = format_observations(
-        changes_df = changes_df,
-        versions_df = versions_df,
+    n_written = format_observations_duckdb(
+        changes_path = CHANGES_PATH,
+        versions_path = VERSIONS_PATH,
+        output_path = OUT_PATH,
         tag_key = TAG_KEY,
         keep_keys = OSM_KEYS,
+        duckdb_memory_limit = DUCKDB_MEMORY_LIMIT,
+        duckdb_threads = DUCKDB_THREADS,
     )
-
-    # Save observations
-    out_path = SAVE_DIR / f"osm_observations_{TAG_KEY}.csv"
-    observations_df.to_csv(out_path, index = False)
-    print(f"Saved {len(observations_df)} observations to {out_path}")
+    print(f"Saved {n_written} observations to {OUT_PATH}")