Make RAS ES sync incremental with hash-based change detection

Replace CREATE OR REPLACE with MERGE using content_hash (SHA256 of key fields)
so only changed rows get a new refreshed_at timestamp. ES sync filters by
refreshed_at in incremental mode, reducing ~1h40m full syncs to minutes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: caseydm

jobs/walden_end2end.yaml

@@ -234,6 +234,8 @@ resources:
             - task_key: Sync_to_Elasticsearch
           notebook_task:
             notebook_path: notebooks/elastic/sync_affiliation_strings_to_elastic_v2
+            base_parameters:
+              is_full_sync: "false"
             source: GIT
           job_cluster_key: es_sync_cluster
         - task_key: Wunpaywall

notebooks/elastic/sync_affiliation_strings_to_elastic_v2.py

@@ -25,12 +25,15 @@
 log = logging.getLogger(__name__)
 
 ELASTIC_URL = dbutils.secrets.get(scope="elastic", key="elastic_url")
+IS_FULL_SYNC = dbutils.widgets.get("is_full_sync").lower() == "true"
 
 CONFIG = {
     "table_name": "openalex.institutions.affiliation_strings_lookup_with_counts",
     "index_name": "raw-affiliation-strings-v2"
 }
 
+print(f"IS_FULL_SYNC: {IS_FULL_SYNC}")
+
 def create_index_if_not_exists(client, index_name):
     """Create index with accent-folding analyzer if it doesn't exist."""
     if client.indices.exists(index=index_name):
@@ -130,10 +133,17 @@ def format_institution_id(id_val):
 
 format_id_udf = F.udf(format_institution_id, StringType())
 
-print(f"\n=== Processing {CONFIG['table_name']} ===")
+print(f"\n=== Processing {CONFIG['table_name']} (full_sync={IS_FULL_SYNC}) ===")
 
 try:
-    df = (spark.table(CONFIG['table_name'])
+    source_df = spark.table(CONFIG['table_name'])
+
+    if not IS_FULL_SYNC:
+        source_df = source_df.filter(
+            F.col("refreshed_at") >= F.expr("current_date() - INTERVAL 3 DAYS")
+        )
+
+    df = (source_df
         # Create a hashed ID from raw_affiliation_string
         .withColumn("id", F.sha2(F.col("raw_affiliation_string"), 256))
         # Format institution IDs with I prefix
@@ -159,9 +169,21 @@ def format_institution_id(id_val):
         .filter(F.col("id").isNotNull())
     )
 
-    df = df.repartition(8)  # Reduced from 32 to limit ES load
     record_count = df.count()
-    print(f"Total records to process: {record_count}")
+    print(f"Total records to process: {record_count:,}")
+
+    # Dynamic partitioning based on record volume
+    if IS_FULL_SYNC:
+        num_partitions = 8
+    elif record_count < 100_000:
+        num_partitions = 2
+    elif record_count < 1_000_000:
+        num_partitions = 4
+    else:
+        num_partitions = 8
+
+    df = df.repartition(num_partitions)
+    print(f"Using {num_partitions} partitions")
 
     def send_partition_wrapper(partition):
         return send_partition_to_elastic(

notebooks/end2end/RefreshRasWorksCounts.ipynb

@@ -3,7 +3,7 @@
   {
    "cell_type": "markdown",
    "metadata": {},
-   "source": "# Refresh RAS Works Counts\n\nRebuilds the `affiliation_strings_lookup_with_counts` table with fresh works counts\nfrom `work_authorships` and institution IDs from the MV (which includes curations).\n\nThis keeps the affiliations dashboard in sync with actual works data.\n\n**Runs after**: Guardrails (needs finalized works data)\n**Feeds**: `sync_affiliation_strings_to_elastic_v2` (ES sync for dashboard)"
+   "source": "# Refresh RAS Works Counts\n\nRebuilds the `affiliation_strings_lookup_with_counts` table with fresh works counts\nfrom `OpenAlex_works` and institution IDs from the MV (which includes curations).\n\nUses a MERGE with content hashing to detect changes — only rows with changed data\nget a new `refreshed_at` timestamp, enabling incremental ES sync downstream.\n\n**Runs after**: Guardrails (needs finalized works data)\n**Feeds**: `sync_affiliation_strings_to_elastic_v2` (ES sync for dashboard)"
   },
   {
    "cell_type": "markdown",
@@ -15,7 +15,7 @@
   {
    "cell_type": "code",
    "metadata": {},
-   "source": "-- Rebuild works counts by exploding authorships from work_authorships.\n-- Uses work_authorships instead of OpenAlex_works for a much faster scan (narrow table).\n-- This replaces the entire table with fresh counts.\nCREATE OR REPLACE TABLE openalex.institutions.affiliation_string_works_counts AS\nSELECT \n    raw_aff_string,\n    COUNT(DISTINCT w.work_id) as works_count\nFROM openalex.works.work_authorships w\nLATERAL VIEW EXPLODE(authorships) AS authorship\nLATERAL VIEW EXPLODE(authorship.raw_affiliation_strings) AS raw_aff_string\nGROUP BY raw_aff_string",
+   "source": "-- Rebuild works counts by exploding authorships from OpenAlex_works.\n-- This replaces the entire counts table with fresh data.\nCREATE OR REPLACE TABLE openalex.institutions.affiliation_string_works_counts AS\nSELECT\n    raw_aff_string,\n    COUNT(DISTINCT w.id) as works_count\nFROM openalex.works.OpenAlex_works w\nLATERAL VIEW EXPLODE(authorships) AS authorship\nLATERAL VIEW EXPLODE(authorship.raw_affiliation_strings) AS raw_aff_string\nGROUP BY raw_aff_string",
    "outputs": [],
    "execution_count": null
   },
@@ -37,47 +37,41 @@
   {
    "cell_type": "markdown",
    "metadata": {},
-   "source": [
-    "## Step 2: Rebuild lookup with counts\n",
-    "\n",
-    "Joins the MV (which has curations applied via 3-layer priority) with fresh counts.\n",
-    "Only keeps RAS that appear in at least one work."
-   ]
+   "source": "## Step 2: MERGE lookup with counts (hash-based change detection)\n\nBuilds a staging table with a `content_hash` of key fields, then MERGEs into the\ntarget. Only rows where the hash changed get `refreshed_at` updated, enabling\nincremental ES sync. New rows are inserted, removed rows are deleted."
   },
   {
    "cell_type": "code",
+   "source": "-- Enable schema auto-merge so the MERGE can add content_hash and refreshed_at\n-- columns to the existing table on first run (they'll start as NULLs).\nSET spark.databricks.delta.schema.autoMerge.enabled = true",
    "metadata": {},
-   "source": [
-    "%sql\n",
-    "CREATE OR REPLACE TABLE openalex.institutions.affiliation_strings_lookup_with_counts AS\n",
-    "SELECT \n",
-    "    mv.raw_affiliation_string,\n",
-    "    mv.institution_ids AS institution_ids_final,\n",
-    "    mv.model_institution_ids AS institution_ids_from_model,\n",
-    "    mv.institution_ids_override,\n",
-    "    mv.countries,\n",
-    "    mv.source,\n",
-    "    mv.created_datetime,\n",
-    "    mv.updated_datetime,\n",
-    "    c.works_count\n",
-    "FROM openalex.institutions.raw_affiliation_strings_institutions_mv mv\n",
-    "INNER JOIN openalex.institutions.affiliation_string_works_counts c\n",
-    "    ON mv.raw_affiliation_string = c.raw_aff_string"
-   ],
+   "execution_count": null,
    "outputs": []
   },
   {
    "cell_type": "code",
    "metadata": {},
-   "source": [
-    "%sql\n",
-    "-- Verify rebuild\n",
-    "SELECT\n",
-    "  COUNT(*) AS total_rows,\n",
-    "  COUNT(CASE WHEN SIZE(institution_ids_final) > 0 THEN 1 END) AS rows_with_institutions,\n",
-    "  ROUND(COUNT(CASE WHEN SIZE(institution_ids_final) > 0 THEN 1 END) * 100.0 / COUNT(*), 1) AS pct_with_institutions\n",
-    "FROM openalex.institutions.affiliation_strings_lookup_with_counts"
-   ],
+   "source": "-- Build staging table with content hash for change detection\nCREATE OR REPLACE TABLE openalex.institutions._ras_lookup_staging AS\nSELECT\n    mv.raw_affiliation_string,\n    mv.institution_ids AS institution_ids_final,\n    mv.model_institution_ids AS institution_ids_from_model,\n    mv.institution_ids_override,\n    mv.countries,\n    mv.source,\n    mv.created_datetime,\n    mv.updated_datetime,\n    c.works_count,\n    SHA2(TO_JSON(NAMED_STRUCT(\n        'iif', mv.institution_ids,\n        'iim', mv.model_institution_ids,\n        'iio', mv.institution_ids_override,\n        'c', mv.countries,\n        'wc', c.works_count\n    )), 256) AS content_hash\nFROM openalex.institutions.raw_affiliation_strings_institutions_mv mv\nINNER JOIN openalex.institutions.affiliation_string_works_counts c\n    ON mv.raw_affiliation_string = c.raw_aff_string",
+   "outputs": [],
+   "execution_count": null
+  },
+  {
+   "cell_type": "code",
+   "source": "-- MERGE with hash-based change detection.\n-- Only updates rows where content actually changed (new refreshed_at).\n-- Inserts new rows, deletes rows no longer in source.\n-- On first run, COALESCE(target.content_hash, '') handles NULLs from schema migration.\nMERGE INTO openalex.institutions.affiliation_strings_lookup_with_counts AS target\nUSING openalex.institutions._ras_lookup_staging AS source\nON target.raw_affiliation_string = source.raw_affiliation_string\nWHEN MATCHED AND COALESCE(target.content_hash, '') <> source.content_hash THEN\n    UPDATE SET\n        institution_ids_final = source.institution_ids_final,\n        institution_ids_from_model = source.institution_ids_from_model,\n        institution_ids_override = source.institution_ids_override,\n        countries = source.countries,\n        source = source.source,\n        created_datetime = source.created_datetime,\n        updated_datetime = source.updated_datetime,\n        works_count = source.works_count,\n        content_hash = source.content_hash,\n        refreshed_at = CURRENT_TIMESTAMP()\nWHEN NOT MATCHED THEN\n    INSERT (raw_affiliation_string, institution_ids_final, institution_ids_from_model,\n            institution_ids_override, countries, source, created_datetime, updated_datetime,\n            works_count, content_hash, refreshed_at)\n    VALUES (source.raw_affiliation_string, source.institution_ids_final, source.institution_ids_from_model,\n            source.institution_ids_override, source.countries, source.source, source.created_datetime,\n            source.updated_datetime, source.works_count, source.content_hash, CURRENT_TIMESTAMP())\nWHEN NOT MATCHED BY SOURCE THEN DELETE",
+   "metadata": {},
+   "execution_count": null,
+   "outputs": []
+  },
+  {
+   "cell_type": "code",
+   "source": "DROP TABLE IF EXISTS openalex.institutions._ras_lookup_staging",
+   "metadata": {},
+   "execution_count": null,
+   "outputs": []
+  },
+  {
+   "cell_type": "code",
+   "source": "-- Verify rebuild + change detection stats\nSELECT\n  COUNT(*) AS total_rows,\n  COUNT(CASE WHEN SIZE(institution_ids_final) > 0 THEN 1 END) AS rows_with_institutions,\n  ROUND(COUNT(CASE WHEN SIZE(institution_ids_final) > 0 THEN 1 END) * 100.0 / COUNT(*), 1) AS pct_with_institutions,\n  COUNT(CASE WHEN refreshed_at >= CURRENT_DATE() THEN 1 END) AS rows_refreshed_today,\n  MIN(refreshed_at) AS oldest_refresh,\n  MAX(refreshed_at) AS newest_refresh\nFROM openalex.institutions.affiliation_strings_lookup_with_counts",
+   "metadata": {},
+   "execution_count": null,
    "outputs": []
   }
  ],