Add interactive schedule_o notebook from Databricks

notebooks/NP01_cleaning-steps-990PF-grants.ipynb

@@ -0,0 +1,1052 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "7c432975-f5e2-406f-9e43-39aeb35c057e",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "from pyspark.sql import functions as F\n",
+    "from pyspark.sql.window import Window"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {},
+     "inputWidgets": {},
+     "nuid": "2eabbeba-4c9f-471f-a820-02a4b3a4ce89",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "source": [
+    "#Prepare 990pfp15grants3a dataset\n",
+    "\n",
+    "This notebook has some data cleaning steps for the 990pfp15grants3a dataset, which contains data from Form 990 PF, Part XIV question 3a (\"Grants and Contributions Paid During the Year\")  \n",
+    "\n",
+    "I've summarized data related to the data cleaning steps [here.](https://docs.google.com/spreadsheets/d/1FLx0uBV0T_3Vq2uVCfZH_Z1DBCQkT4-S6APtTrWyPuc/edit?gid=88674851#gid=88674851)  \n",
+    "\n",
+    "1. The majority of records are missing a valid country code but do have a valid US state code, so we join on valid US state codes to identify the country.\n",
+    "2. Filter out records where the grant value is Null or zero.\n",
+    "3. This leaves only 1% records with a Null country code. \n",
+    "4. For the remainder, I ran the place names and city codes through a geopy lookup to try and identify the country (notebook NP00_geopy-lookups). The majority of entires seem to be foreign countries, so in the below I've assumed that any of the remaining records with null country codes are foreign but more work will need to be done to validate this."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {},
+     "inputWidgets": {},
+     "nuid": "e81bcd5e-cdd2-4403-ba21-27f7f2d22b8b",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "source": [
+    "##Load data"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "882f4738-c66f-4f9c-8415-ec3eb155f562",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "grants3a = spark.table(\"prod_curated.irs.990pfpart14grants3a\")\n",
+    "# geo_data = spark.table(\"sandbox_edward.geo_data.geonames_all_countries\")\n",
+    "state_code_lookup = spark.table(\"sandbox_edward.nonprofit_mapping.state_code_lookup\")\n",
+    "geopy_place_lookup = spark.table(\"sandbox_edward.nonprofit_mapping.geopy_place_lookup\")\n",
+    "geopy_city_lookup = spark.table(\"sandbox_edward.nonprofit_mapping.geopy_city_lookup\")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "f336e511-02ad-4fed-9e2d-eb7bc908a41b",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "Clean grants3a"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "# Regex to match exactly 9 digits\n",
+    "regex_pattern_9zips = r\"^\\d{9}$\"\n",
+    "regex_pattern_pcs = r'^(\\w{3,4}) \\w{3}$' # Matches 3 or 4 characters/digits followed by 3 characters/digits\n",
+    "\n",
+    "grants3a_cleaned = (\n",
+    "    grants3a\n",
+    "    .filter(\n",
+    "        (F.col('SIGOCPYAMOUN') > 0) &\n",
+    "        (F.col('SIGOCPYAMOUN').isNotNull())\n",
+    "    ).withColumn(                                               \n",
+    "        'original_postcode',                                    # Keep original values in a separate column\n",
+    "        F.col('SIGOCPYRFAPC')\n",
+    "    ).withColumn(\n",
+    "        'SIGOCPYRFAPC',\n",
+    "        F.when(\n",
+    "            F.col(\"SIGOCPYRFAPC\").rlike(regex_pattern_9zips),   # Check if the postcode is a nine digit value\n",
+    "            F.col(\"SIGOCPYRFAPC\").substr(1, 5)                  # If it is, extract the first five digits\n",
+    "        ).otherwise(F.col(\"SIGOCPYRFAPC\"))                      # Otherwise, keep the original value\n",
+    "    ).withColumn(\n",
+    "        'SIGOCPYRFAPC',\n",
+    "        F.when(\n",
+    "            F.col(\"SIGOCPYRFAPC\").rlike(regex_pattern_pcs),                 # Check if it matches the regex\n",
+    "            F.regexp_extract(F.col('SIGOCPYRFAPC'), regex_pattern_pcs, 1)   # Extract the first 5 digits if it matches\n",
+    "        ).otherwise(F.col(\"SIGOCPYRFAPC\"))                                  # Otherwise, keep the original value\n",
+    "    ).withColumn(\n",
+    "        'SIGOCAFFRFACO',\n",
+    "        F.when(F.col('SIGOCAFFRFACO') == \"USA\", \"US\")\n",
+    "        .otherwise(F.col('SIGOCAFFRFACO'))\n",
+    "    ).join(\n",
+    "        state_code_lookup.select(\n",
+    "            F.lower(F.col('state_code')).alias('state_code_lower'),\n",
+    "            F.col('country').alias('country_from_state_code')\n",
+    "        ),\n",
+    "        on=F.lower(F.col('SIGOCPYRFAPO')) == F.col('state_code_lower'),\n",
+    "        how='left'\n",
+    "    ).join(\n",
+    "        state_code_lookup.select(\n",
+    "            F.lower(F.col('state_name')).alias('state_name_lower'),\n",
+    "            F.col('country').alias('country_from_state_name'),\n",
+    "            F.col('state_code')\n",
+    "        ),\n",
+    "        on=F.lower(F.col('SIGOCPYRFAPO')) == F.col('state_name_lower'),\n",
+    "        how='left'\n",
+    "    ).withColumn(\n",
+    "        'valid_state_code',\n",
+    "        F.upper(F.coalesce(F.col('state_code_lower'), F.col('state_code')))\n",
+    "    ).withColumn(\n",
+    "        'valid_state_code',                                                                         # If the state code is null, assume international\n",
+    "        F.when(F.col('valid_state_code').isNull(), 'INTL').otherwise(F.col('valid_state_code'))\n",
+    "    ).withColumn(\n",
+    "        'country_from_state_lookup',\n",
+    "        F.coalesce(F.col('country_from_state_code'), F.col('country_from_state_name'))\n",
+    "    ).drop(\n",
+    "        'state_name_lower',\n",
+    "        'state_code_lower',\n",
+    "        'state_code',\n",
+    "        'country_from_state_code',\n",
+    "        'country_from_state_name'\n",
+    "    ).withColumn(\n",
+    "        'country_from_state_lookup',\n",
+    "        F.coalesce(F.col('country_from_state_lookup'), F.col('SIGOCAFFRFACO'))\n",
+    "    ).withColumn(\n",
+    "        'designated_foreign_org',\n",
+    "        F.when(\n",
+    "            (F.lower(F.col('SIGOCPYRFSTA')).like('%forei%')) |\n",
+    "            F.lower(F.col('SIGOCPYRFSTA')).like('%forgn%'),\n",
+    "            1\n",
+    "        ).otherwise(0)\n",
+    "    ).withColumn(\n",
+    "        'country_from_state_lookup',\n",
+    "        F.when(F.col('country_from_state_lookup').isNull(), 'Assumed international')\n",
+    "        .otherwise(F.col('country_from_state_lookup'))\n",
+    "    )\n",
+    ")\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "bbd388ea-2a1b-471f-84f0-51e3969c4169",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "Write dataset"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "grants3a_cleaned.write.mode(\"overwrite\").format(\"delta\").saveAsTable('sandbox_edward.nonprofit_mapping.grants3a_cleaned')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "594d69d5-8ade-48de-aed5-a07c41de5038",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "display(\n",
+    "    grants3a_cleaned\n",
+    "    .groupBy(\n",
+    "        'country_from_state_lookup'\n",
+    "    ).agg(\n",
+    "        F.count('*').alias('count')\n",
+    "    )\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "a5d2c47b-d666-4e63-8726-f1aec028d81e",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "display(grants3a_cleaned)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "3152e313-b318-497f-883b-15071e21b0e8",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "no_country_no_state_cities = (\n",
+    "    grants3a_cleaned\n",
+    "    .filter(F.col('country').isNull() & F.col('SIGOCPYRFAPO').isNull())\n",
+    "    .groupBy(\n",
+    "        'SIGOCPYRFACI'\n",
+    "    ).agg(\n",
+    "        F.count('*').alias('count')\n",
+    "    ).orderBy(\n",
+    "        F.col('count').desc()\n",
+    "    )\n",
+    ")\n",
+    "\n",
+    "no_country_no_state_cities.write.format(\"delta\").saveAsTable('sandbox_edward.nonprofit_mapping.no_country_no_state_cities')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "d21e77ac-9397-4545-b2ae-fda9e2ab10b1",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "display(grants3a_cleaned.agg(F.sum('likely_foreign').alias('sum_likely_foreign')))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "c75a63c3-3b70-4cfb-9815-d30bd51892b0",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "recipient_countries"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "window_spec = Window.partitionBy().orderBy(F.col('count').desc())\n",
+    "\n",
+    "recipient_countries = (\n",
+    "    grants3a_cleaned\n",
+    "    .groupBy('SIGOCAFFRFACO')\n",
+    "    .agg(F.count('*').alias('count'))\n",
+    "    .withColumn('total', F.sum('count').over(Window.partitionBy()))\n",
+    "    .withColumn('proportion', F.col('count') / F.col('total'))\n",
+    "    .withColumn('cumulative_count', F.sum('count').over(window_spec))\n",
+    "    .withColumn('cumulative_proportion', F.col('cumulative_count') / F.col('total'))\n",
+    ")\n",
+    "\n",
+    "display(recipient_countries.orderBy(F.col('count').desc()))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "80d1d532-f6a8-4c70-ad20-7a07abff85a6",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "recipient_states"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "window_spec = Window.partitionBy().orderBy(F.col('count').desc())\n",
+    "\n",
+    "recipient_states = (\n",
+    "    grants3a_cleaned\n",
+    "    .groupBy('SIGOCPYRFAPO')\n",
+    "    .agg(F.count('*').alias('count'))\n",
+    "    .withColumn('total', F.sum('count').over(Window.partitionBy()))\n",
+    "    .withColumn('proportion', F.col('count') / F.col('total'))\n",
+    "    .withColumn('cumulative_count', F.sum('count').over(window_spec))\n",
+    "    .withColumn('cumulative_proportion', F.col('cumulative_count') / F.col('total'))\n",
+    ")\n",
+    "\n",
+    "display(recipient_states.orderBy(F.col('count').desc()))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "e4583edc-859e-4826-ad4c-f88dfa9979ae",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "window_spec = Window.partitionBy().orderBy(F.col('count').desc())\n",
+    "\n",
+    "recipient_country_states = (\n",
+    "    grants3a_cleaned\n",
+    "    .groupBy('SIGOCAFFRFACO', 'SIGOCPYRFAPO')\n",
+    "    .agg(F.count('*').alias('count'))\n",
+    "    .withColumn('total', F.sum('count').over(Window.partitionBy()))\n",
+    "    .withColumn('proportion', F.col('count') / F.col('total'))\n",
+    "    .withColumn('cumulative_count', F.sum('count').over(window_spec))\n",
+    "    .withColumn('cumulative_proportion', F.col('cumulative_count') / F.col('total'))\n",
+    ")\n",
+    "\n",
+    "display(recipient_country_states.orderBy(F.col('count').desc()))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "7abdc582-3d63-4755-80d1-cc59d05711ff",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "output = (\n",
+    "    grants3a_cleaned\n",
+    "    .withColumn(\n",
+    "        'country_identified',\n",
+    "        F.when(F.col('SIGOCAFFRFACO').isNotNull(), True)\n",
+    "        .otherwise(False)\n",
+    "    ).groupBy(\n",
+    "        'FILERNAME1', 'SIGOCPYRFAPO'\n",
+    "    ).agg(\n",
+    "        F.count()\n",
+    "    )\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "efc2b6c8-7f33-420f-966c-cc2a4edbeb7d",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "inconsistent_records"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "inconsistent_records = (\n",
+    "    grants3a\n",
+    "    .groupBy(\n",
+    "        'SIGOCPYRBNBN1', 'SIGOCAFFRFACO'\n",
+    "    ).agg(\n",
+    "        F.count('*').alias('count')\n",
+    "    ).withColumn(\n",
+    "        'id',\n",
+    "        F.row_number().over(Window.partitionBy('SIGOCPYRBNBN1').orderBy(F.col('count').desc()))\n",
+    "    ).withColumn(\n",
+    "        'duplicates',\n",
+    "        F.max('id').over(Window.partitionBy('SIGOCPYRBNBN1'))\n",
+    "    ).filter(\n",
+    "        F.col('duplicates') > 1\n",
+    "    )\n",
+    ")\n",
+    "\n",
+    "display(inconsistent_records)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {},
+     "inputWidgets": {},
+     "nuid": "9b7da0b7-7be3-4e16-ba13-cbb444bb2747",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "source": [
+    "#Old cleaning attempt - can ignore"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {},
+     "inputWidgets": {},
+     "nuid": "87d7a6d6-9944-4c69-b895-4f4923e528b0",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "source": [
+    "##Merge country information\n",
+    "\n",
+    "Column name: SIGOCPYRFACI\\\n",
+    "Description: Recipient Foreign Address - City\n",
+    "\n",
+    "Column name: SIGOCPYRFAPC\\\n",
+    "Description: Recipient Foreign Address - Postal code\n",
+    "\n",
+    "Column name: SIGOCPYRFACO\\\n",
+    "Description: Recipient Foreign Address - Country\n",
+    "\n",
+    "Column name: SIGOCPYRFAPO\\\n",
+    "Description: Province or state"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "f6c96bf0-042d-4813-b64e-e2b531ed4d05",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "# First, check the number of rows in the dataset:\n",
+    "# 13,119,448\n",
+    "# 13,119,506\n",
+    "# 13,119,506\n",
+    "# 13,132,628\n",
+    "\n",
+    "display(grants3a.count())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "896f06e6-b8e6-4c2a-bff8-7953c243c163",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "convert place information to lower case"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "grants3a = (\n",
+    "    grants3a\n",
+    "    .withColumn('SIGOCPYRFACI', F.lower(F.col('SIGOCPYRFACI'))) # City / place name\n",
+    "    .withColumn('SIGOCPYRFAPC', F.lower(F.col('SIGOCPYRFAPC'))) # Postcode\n",
+    "    .withColumn('SIGOCPYRFAPO', F.lower(F.col('SIGOCPYRFAPO'))) # State codes\n",
+    "    .withColumn('row_id', F.row_number().over(Window.orderBy(F.lit(0))))\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "4d60a9f8-a8b3-4eb8-aa99-2920450eb909",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "geo_data"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "geo_data = (\n",
+    "    geo_data\n",
+    "    .withColumn('place_name', F.lower(F.col('place_name')))\n",
+    "    .withColumn('postal_code', F.lower(F.col('postal_code')))\n",
+    "    .withColumn('admin_code1', F.lower(F.col('admin_code1')))\n",
+    "    .withColumn('admin_name1', F.lower(F.col('admin_name1')))\n",
+    "    # .withColumnRenamed('place_name', 'SIGOCPYRFACI')\n",
+    "    # .withColumnRenamed('postal_code', 'SIGOCPYRFAPC')\n",
+    "    # .withColumnRenamed('admin_code1', 'SIGOCPYRFAPO')\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "9413adb6-10ca-4cef-a6ed-267e7d71e28e",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "joined_geo"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "# First attempt to match on place name and postocde\n",
+    "\n",
+    "geo_data_1 = (\n",
+    "    geo_data\n",
+    "    .select('country_code', 'place_name', 'postal_code', 'admin_code1')\n",
+    "    .withColumnRenamed('country_code', 'country_code_nm_pc_st_mtch')\n",
+    "    .distinct()\n",
+    ")\n",
+    "\n",
+    "joined_geo = (\n",
+    "    grants3a\n",
+    "    .join(geo_data_1,\n",
+    "          on = (\n",
+    "              (F.col('SIGOCPYRFACI') == F.col('place_name')) &\n",
+    "              (F.col('SIGOCPYRFAPC') == F.col('postal_code')) &\n",
+    "              (F.col('SIGOCPYRFAPO') == F.col('admin_code1')) &\n",
+    "              (F.col('SIGOCAFFRFACO').isNull())\n",
+    "              ),\n",
+    "          how='left')\n",
+    "    .drop('place_name', 'postal_code', 'admin_code1')\n",
+    ")\n",
+    "\n",
+    "display(joined_geo.count())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "38b09e83-db48-4e2a-b509-a12ba5e0416f",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "joined_geo2"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "# First attempt to match on place name and postocde\n",
+    "\n",
+    "geo_data_2 = (\n",
+    "    geo_data\n",
+    "    .select('country_code', 'place_name', 'postal_code')\n",
+    "    .withColumnRenamed('country_code', 'country_code_nm_pc_mtch')\n",
+    "    .distinct()\n",
+    ")\n",
+    "\n",
+    "joined_geo2 = (\n",
+    "    joined_geo\n",
+    "    .join(geo_data_2,\n",
+    "          on = (\n",
+    "              (F.col('SIGOCPYRFACI') == F.col('place_name')) &\n",
+    "              (F.col('SIGOCPYRFAPC') == F.col('postal_code')) &\n",
+    "              (F.col('SIGOCAFFRFACO').isNull()) &\n",
+    "              (F.col('country_code_nm_pc_st_mtch').isNull())\n",
+    "              ),\n",
+    "          how='left')\n",
+    "    .drop('place_name', 'postal_code')\n",
+    ")\n",
+    "\n",
+    "display(joined_geo2.count())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "9c68184d-2476-4bc6-8d53-7262785e8b6c",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "joined_geo3"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "# Second, attempt to match on state code and postocde\n",
+    "\n",
+    "geo_data_3 = (\n",
+    "    geo_data\n",
+    "    .select('country_code', 'admin_code1', 'postal_code')\n",
+    "    .withColumnRenamed('country_code', 'country_code_st_pc_mtch')\n",
+    "    .distinct()\n",
+    ")\n",
+    "\n",
+    "# Perform the join with additional conditions in the \"on\" clause\n",
+    "joined_geo3 = (\n",
+    "    joined_geo2\n",
+    "    .join(\n",
+    "        geo_data_3,\n",
+    "        on=(\n",
+    "            (F.col('SIGOCPYRFAPO') == F.col('admin_code1')) &  # Match postocode\n",
+    "            (F.col('SIGOCPYRFAPC') == F.col('postal_code')) &  # Match state/province\n",
+    "            (F.col('SIGOCAFFRFACO').isNull()) &\n",
+    "            (F.col('country_code_nm_pc_st_mtch').isNull()) &        # Only join when place name & postcode match did not find a match\n",
+    "            (F.col('country_code_nm_pc_mtch').isNull())\n",
+    "        ),\n",
+    "        how='left'\n",
+    "    )\n",
+    "    .drop('admin_code1', 'postal_code')\n",
+    ")\n",
+    "\n",
+    "display(joined_geo3.count())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "995292e5-8832-4728-b891-750a033e025d",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "joined_geo4"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "# Third, attempt to match on place name and state code\n",
+    "\n",
+    "geo_data_4 = (\n",
+    "    geo_data\n",
+    "    .select('country_code', 'admin_code1', 'place_name')\n",
+    "    .withColumnRenamed('country_code', 'country_code_st_nm_mtch')\n",
+    "    .distinct()\n",
+    ")\n",
+    "\n",
+    "# Perform the join with additional conditions in the \"on\" clause\n",
+    "joined_geo4 = (\n",
+    "    joined_geo3\n",
+    "    .withColumn('SIGOCPYRFAPO', F.lower(F.col('SIGOCPYRFAPO')))\n",
+    "    .join(\n",
+    "        geo_data_4,\n",
+    "        on=(\n",
+    "            (F.col('SIGOCPYRFACI') == F.col('place_name')) &  # Match SIGOCPYRFAPC\n",
+    "            (F.col('SIGOCPYRFAPO') == F.col('admin_code1')) &  # Match SIGOCPYRFAPO\n",
+    "            (F.col('SIGOCAFFRFACO').isNull()) &\n",
+    "            (F.col('country_code_nm_pc_st_mtch').isNull()) &\n",
+    "            (F.col('country_code_nm_pc_mtch').isNull()) & \n",
+    "            (F.col('country_code_st_pc_mtch').isNull()) # Only match in instances where previous matches have failed\n",
+    "        ),\n",
+    "        how='left'\n",
+    "    )\n",
+    "    .drop('admin_code1', 'place_name')\n",
+    ")\n",
+    "\n",
+    "display(joined_geo4.count())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "6ea92529-581a-4ad9-98f1-aa5c347422f6",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "join duplicate check"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "duplicate_window = Window.partitionBy('row_id').orderBy(F.lit(0))\n",
+    "\n",
+    "joined_geo4 = (\n",
+    "    joined_geo3\n",
+    "    .withColumn('repeat_count', F.sum(F.lit(1)).over(duplicate_window))\n",
+    ")\n",
+    "\n",
+    "display(joined_geo4.filter(F.col('repeat_count') > 1))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "85f9154a-4e4b-4802-a9b8-a71096993ead",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "recipient_countries"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "window_spec = Window.partitionBy().orderBy(F.col('count').desc())\n",
+    "\n",
+    "recipient_countries = (\n",
+    "    joined_geo4\n",
+    "    .groupBy('SIGOCAFFRFACO', \n",
+    "             'country_code_nm_pc_st_mtch',\n",
+    "             'country_code_nm_pc_mtch',\n",
+    "             'country_code_st_pc_mtch',\n",
+    "             'country_code_st_nm_mtch'\n",
+    "             )\n",
+    "    .agg(F.count('*').alias('count'))\n",
+    "    .withColumn('total', F.sum('count').over(Window.partitionBy()))\n",
+    "    .withColumn('proportion', F.col('count') / F.col('total'))\n",
+    "    .withColumn('cumulative_count', F.sum('count').over(window_spec))\n",
+    "    .withColumn('cumulative_proportion', F.col('cumulative_count') / F.col('total'))\n",
+    ")\n",
+    "\n",
+    "display(recipient_countries.orderBy(F.col('count').desc()))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "70b84b6b-b039-41b9-a8b4-3ae0818520d7",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "missing_countries"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "missing_countries = (\n",
+    "    joined_geo4\n",
+    "    .filter(\n",
+    "        (F.col('SIGOCAFFRFACO').isNull()) &\n",
+    "        (F.col('country_code_nm_pc_st_mtch').isNull()) &\n",
+    "        (F.col('country_code_nm_pc_mtch').isNull()) &\n",
+    "        (F.col('country_code_st_pc_mtch').isNull()) &\n",
+    "        (F.col('country_code_st_nm_mtch').isNull()) &\n",
+    "        (F.col('SIGOCPYAMOUN') > 0)\n",
+    "        )\n",
+    "    )"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "f8ffca0f-f043-41df-af46-7f225eee6cc6",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "missing_countries"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "display(missing_countries.filter(F.col('SIGOCPYRFAPO') == 'ontario'))"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "a4bcbfa0-4c19-4377-981d-bd9d72a8fd0f",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "missing_countries"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "display(\n",
+    "    missing_countries\n",
+    "    .groupBy(\n",
+    "        'SIGOCPYRFAPO'\n",
+    "    ).agg(\n",
+    "        F.count('*').alias('count')\n",
+    "    ).orderBy(\n",
+    "        'count',\n",
+    "        ascending=False\n",
+    "    )\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "79c3bb18-50f8-4538-89d7-92288d143cb5",
+     "showTitle": true,
+     "tableResultSettingsMap": {},
+     "title": "missing_countries1"
+    }
+   },
+   "outputs": [],
+   "source": [
+    "geo_data_5 = (\n",
+    "    geo_data\n",
+    "    .select('country_code', 'admin_name1', 'place_name')\n",
+    "    .withColumnRenamed('country_code', 'country_code_stnm_nm_mtch')\n",
+    "    .distinct()\n",
+    ")\n",
+    "\n",
+    "# Perform the join with additional conditions in the \"on\" clause\n",
+    "missing_countries1 = (\n",
+    "    missing_countries\n",
+    "    .withColumn('SIGOCPYRFAPO', F.lower(F.col('SIGOCPYRFAPO')))\n",
+    "    .join(\n",
+    "        geo_data_5,\n",
+    "        on=(\n",
+    "            (F.col('SIGOCPYRFACI') == F.col('place_name')) &  # Match SIGOCPYRFAPC\n",
+    "            (F.col('SIGOCPYRFAPO') == F.col('admin_name1')) &  # Match SIGOCPYRFAPO\n",
+    "            (F.col('SIGOCAFFRFACO').isNull()) &\n",
+    "            (F.col('country_code_nm_pc_st_mtch').isNull()) &\n",
+    "            (F.col('country_code_nm_pc_mtch').isNull()) & \n",
+    "            (F.col('country_code_st_pc_mtch').isNull()) &\n",
+    "            (F.col('country_code_st_nm_mtch').isNull())\n",
+    "        ),\n",
+    "        how='left'\n",
+    "    )\n",
+    "    .drop('admin_name1', 'place_name')\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "application/vnd.databricks.v1+cell": {
+     "cellMetadata": {
+      "byteLimit": 2048000,
+      "rowLimit": 10000
+     },
+     "inputWidgets": {},
+     "nuid": "2586ab16-8bed-4ef0-8e2c-428702bdb833",
+     "showTitle": false,
+     "tableResultSettingsMap": {},
+     "title": ""
+    }
+   },
+   "outputs": [],
+   "source": [
+    "display(\n",
+    "joined_geo4.filter(\n",
+    "    (F.col('SIGOCAFFRFACO').isNull()) &\n",
+    "    (F.col('country_code_nm_pc_st_mtch').isNull()) &\n",
+    "    (F.col('country_code_nm_pc_mtch').isNull()) &\n",
+    "    (F.col('country_code_st_pc_mtch').isNull()) &\n",
+    "    (F.col('country_code_st_nm_mtch').isNull()) &\n",
+    "    (F.col('SIGOCPYAMOUN') > 0)\n",
+    ")\n",
+    ")"
+   ]
+  }
+ ],
+ "metadata": {
+  "application/vnd.databricks.v1+notebook": {
+   "computePreferences": {
+    "hardware": {
+     "accelerator": null,
+     "gpuPoolId": null,
+     "memory": null
+    }
+   },
+   "dashboards": [],
+   "environmentMetadata": {
+    "base_environment": "",
+    "dependencies": [
+     "pgeocode",
+     "geopy"
+    ],
+    "environment_version": "1"
+   },
+   "inputWidgetPreferences": null,
+   "language": "python",
+   "notebookMetadata": {
+    "mostRecentlyExecutedCommandWithImplicitDF": {
+     "commandId": 567296865538331,
+     "dataframes": [
+      "_sqldf"
+     ]
+    },
+    "pythonIndentUnit": 4
+   },
+   "notebookName": "hhaq_NP01_cleaning-steps-990PF-grants",
+   "widgets": {}
+  },
+  "language_info": {
+   "name": "python"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 0
+}