Merge pull request #4536 from fedspendingtransparency/tst/emr-pipeline-testing

Downmerge EMR cutover branch into QAT

Author: sethstoudenmier

.env.template

@@ -39,8 +39,8 @@ USASPENDING_DB_PASSWORD=usaspender
 
 # The Broker configuration below supports tests creating a Broker DB on the usaspending-db
 # container as part of standing up the test suite.
-# All values of BROKER_DB_* must match what is in DATA_BROKER_DATABASE_URL if BOTH are given
-DATA_BROKER_DATABASE_URL=postgres://usaspending:usaspender@usaspending-db:5432/data_broker
+# All values of BROKER_DB_* must match what is in BROKER_DB if BOTH are given
+BROKER_DB=postgres://usaspending:usaspender@usaspending-db:5432/data_broker
 # Configuration values for a connection string to a Broker database
 #    Only necessary for some management commands
 BROKER_DB_HOST=usaspending-db

.github/actions/init-test-environment/action.yaml

@@ -17,7 +17,7 @@ runs:
     - name: Set combined ENV
       shell: bash
       run: |
-        echo "DATA_BROKER_DATABASE_URL=postgres://$BROKER_DB_USER:$BROKER_DB_PASSWORD@$BROKER_DB_HOST:$BROKER_DB_PORT/$BROKER_DB_NAME" >> $GITHUB_ENV
+        echo "BROKER_DB=postgres://$BROKER_DB_USER:$BROKER_DB_PASSWORD@$BROKER_DB_HOST:$BROKER_DB_PORT/$BROKER_DB_NAME" >> $GITHUB_ENV
         echo "DATABASE_URL=postgres://$USASPENDING_DB_USER:$USASPENDING_DB_PASSWORD@$USASPENDING_DB_HOST:$USASPENDING_DB_PORT/$USASPENDING_DB_NAME" >> $GITHUB_ENV
         echo "DOWNLOAD_DATABASE_URL=postgres://$USASPENDING_DB_USER:$USASPENDING_DB_PASSWORD@$USASPENDING_DB_HOST:$USASPENDING_DB_PORT/$USASPENDING_DB_NAME" >> $GITHUB_ENV
         echo "ES_HOSTNAME=$ES_SCHEME://$ES_HOST:$ES_PORT" >> $GITHUB_ENV

README.md

@@ -60,7 +60,7 @@ Create a `.envrc` file in the repo root, which will be ignored by git. Change cr
 ```shell
 export DATABASE_URL=postgres://usaspending:usaspender@localhost:5432/data_store_api
 export ES_HOSTNAME=http://localhost:9200
-export DATA_BROKER_DATABASE_URL=postgres://admin:root@localhost:5435/data_broker
+export BROKER_DB=postgres://admin:root@localhost:5435/data_broker
 ```
 
 If `direnv` does not pick this up after saving the file, type
@@ -220,10 +220,10 @@ Deployed production API endpoints and docs are found by following links here: `h
 
 3. To run all USAspending tests in the docker services run
     ```shell
-    docker compose run --rm -e DATA_BROKER_DATABASE_URL='' usaspending-test
+    docker compose run --rm -e BROKER_DB='' usaspending-test
     ```
 
-_**NOTE**: If an env var named `DATA_BROKER_DATABASE_URL` is set, Broker Integration tests will attempt to be run as well. If doing so, Broker dependencies must be met (see below) or ALL tests will fail hard. Running the above command with `-e DATA_BROKER_DATABASE_URL=''` is a precaution to keep them excluded, unless you really want them (see below if so)._
+_**NOTE**: If an env var named `BROKER_DB` is set, Broker Integration tests will attempt to be run as well. If doing so, Broker dependencies must be met (see below) or ALL tests will fail hard. Running the above command with `-e BROKER_DB=''` is a precaution to keep them excluded, unless you really want them (see below if so)._
 
 To run tests locally and not in the docker services, you need:
 
@@ -273,7 +273,7 @@ To satisfy these dependencies and include execution of these tests, do the follo
     ```shell
     docker build -t dataact-broker-backend ../data-act-broker-backend
     ```
-1. Ensure you have the `DATA_BROKER_DATABASE_URL` environment variable set, and it points to what will be a live PostgreSQL server (no database required) at the time tests are run.
+1. Ensure you have the `BROKER_DB` environment variable set, and it points to what will be a live PostgreSQL server (no database required) at the time tests are run.
     1. _WARNING: If this is set at all, then ALL above dependencies must be met or ALL tests will fail (Django will try this connection on ALL tests' run)_
     1. This DB could be one you always have running in a local Postgres instance, or one you spin up in a Docker container just before tests are run
 1. If invoking `pytest` within a docker container (e.g. using the `usaspending-test` container), you _must_ mount the host's docker socket. This is declared already in the `docker-compose.yml` file services, but would be done manually with: `-v /var/run/docker.sock:/var/run/docker.sock`
@@ -286,15 +286,15 @@ Re-running the test suite using `pytest -rs` with these dependencies satisfied s
 
 _From within a container_
 
-_**NOTE**: `DATA_BROKER_DATABASE_URL` is set in the `docker-compose.yml` file (and could pick up `.env` values, if set)_
+_**NOTE**: `BROKER_DB` is set in the `docker-compose.yml` file (and could pick up `.env` values, if set)_
 
 ```shell
 docker compose run --rm usaspending-test pytest --capture=no --verbose --tb=auto --no-cov --log-cli-level=INFO -k test_broker_integration
 ```
 
 _From Developer Desktop_
 
-_**NOTE**: `DATA_BROKER_DATABASE_URL` is set in the `.envrc` file and available in the shell_
+_**NOTE**: `BROKER_DB` is set in the `.envrc` file and available in the shell_
 ```shell
 pytest --capture=no --verbose --tb=auto --no-cov --log-cli-level=INFO -k test_broker_integration
 ```

docker-compose.yml

@@ -44,7 +44,7 @@ services:
       DJANGO_DEBUG: ${DJANGO_DEBUG}
       DATABASE_URL: postgres://${USASPENDING_DB_USER}:${USASPENDING_DB_PASSWORD}@${USASPENDING_DB_HOST}:${USASPENDING_DB_PORT}/data_store_api
       ES_HOSTNAME: ${ES_HOSTNAME}
-      DATA_BROKER_DATABASE_URL: postgresql://${BROKER_DB_USER}:${BROKER_DB_PASSWORD}@${BROKER_DB_HOST}:${BROKER_DB_PORT}/data_broker
+      BROKER_DB: postgresql://${BROKER_DB_USER}:${BROKER_DB_PASSWORD}@${BROKER_DB_HOST}:${BROKER_DB_PORT}/data_broker
 
   usaspending-test:
     profiles:
@@ -68,7 +68,7 @@ services:
       DATABASE_URL: postgres://${USASPENDING_DB_USER}:${USASPENDING_DB_PASSWORD}@${USASPENDING_DB_HOST}:${USASPENDING_DB_PORT}/data_store_api
       ES_HOST: ${ES_HOST}
       ES_HOSTNAME: ${ES_HOSTNAME}
-      DATA_BROKER_DATABASE_URL: postgresql://${BROKER_DB_USER}:${BROKER_DB_PASSWORD}@${BROKER_DB_HOST}:${BROKER_DB_PORT}/data_broker
+      BROKER_DB: postgresql://${BROKER_DB_USER}:${BROKER_DB_PASSWORD}@${BROKER_DB_HOST}:${BROKER_DB_PORT}/data_broker
       MINIO_HOST: ${MINIO_HOST}
       DOWNLOAD_DATABASE_URL: postgres://${USASPENDING_DB_USER}:${USASPENDING_DB_PASSWORD}@${USASPENDING_DB_HOST}:${USASPENDING_DB_PORT}/data_store_api
       # Location in host machine where broker src code root can be found
@@ -107,7 +107,7 @@ services:
     environment:
       DATABASE_URL: postgres://${USASPENDING_DB_USER}:${USASPENDING_DB_PASSWORD}@${USASPENDING_DB_HOST}:${USASPENDING_DB_PORT}/data_store_api
       ES_HOSTNAME: ${ES_HOSTNAME}
-      DATA_BROKER_DATABASE_URL: postgresql://${BROKER_DB_USER}:${BROKER_DB_PASSWORD}@${BROKER_DB_HOST}:${BROKER_DB_PORT}/data_broker
+      BROKER_DB: postgresql://${BROKER_DB_USER}:${BROKER_DB_PASSWORD}@${BROKER_DB_HOST}:${BROKER_DB_PORT}/data_broker
       # Location in host machine where broker src code root can be found
       DATA_BROKER_SRC_PATH: "${PWD}/../data-act-broker-backend"
 
@@ -233,7 +233,11 @@ services:
         mkdir -p data/dti-da-public-files-nonprod/user_reference_docs
         # Create the bucket within MinIO used for endpoints that list generated downloads
         mkdir -p data/bulk-download
+        # Create the bucket for MinIO used for Spark
+        mkdir -p data/data/files
+        # Populate initial files in buckets
         cp dockermount/usaspending_api/data/Data_Dictionary_Crosswalk.xlsx data/dti-da-public-files-nonprod/user_reference_docs/Data_Dictionary_Crosswalk.xlsx
+        cp dockermount/usaspending_api/data/COVID-19_download_readme.txt data/data/files/COVID-19_download_readme.txt
         minio server --address ":10001" --console-address ":10002" /data
       "
     healthcheck:

loading_data.md

@@ -35,7 +35,7 @@ To load in the reference data, from the same directory as manage.py:
 
 To load certified submission data from the broker, you will need a read-only (or higher) connection string to the broker PostgreSQL database. If not running locally, you will also need to ensure your IP address has been whitelisted in the appropriate AWS Security Groups. Set this environment variable before running the **load_submission** command:
 
-    DATA_BROKER_DATABASE_URL=postgres://user:password@url:5432/data_broker
+    BROKER_DB=postgres://user:password@url:5432/data_broker
 
 To load a submission from data broker database:
 

usaspending_api/awards/delta_models/award_id_lookup.py

@@ -0,0 +1,11 @@
+from pyspark.sql.types import LongType, StringType, StructField, StructType, BooleanType
+
+
+AWARD_ID_LOOKUP_SCHEMA = StructType(
+    [
+        StructField("award_id", LongType(), False),
+        StructField("is_fpds", BooleanType(), False),
+        StructField("transaction_unique_id", StringType(), False),
+        StructField("generated_unique_award_id", StringType(), False),
+    ]
+)

usaspending_api/awards/management/commands/generate_unlinked_awards_download.py

@@ -16,6 +16,7 @@
 from usaspending_api.awards.management.sql.spark.unlinked_awards_summary_file import summary_file
 from usaspending_api.awards.management.sql.spark.unlinked_assistance_file_d2 import file_d2_sql_string
 from usaspending_api.awards.management.sql.spark.unlinked_accounts_file_c import file_c_sql_string
+from usaspending_api.config import CONFIG
 from usaspending_api.download.filestreaming.file_description import build_file_description, save_file_description
 from usaspending_api.download.filestreaming.zip_file import append_files_to_zip_file
 from usaspending_api.references.models.toptier_agency import ToptierAgency
@@ -108,9 +109,10 @@ def handle(self, *args, **options):
         # Save queries as delta tables for efficiency
         for delta_table_name, sql_file, final_name in self.download_file_list:
             df = self.spark.sql(sql_file)
-            df.write.format(source="delta").mode(saveMode="overwrite").option("overwriteSchema", "True").saveAsTable(
-                name=delta_table_name
-            )
+            df.write.format(source="delta").mode(saveMode="overwrite").options(
+                overwriteSchema=True,
+                path=f"s3a://{CONFIG.SPARK_S3_BUCKET}/{CONFIG.DELTA_LAKE_S3_PATH}/temp/{delta_table_name}",
+            ).saveAsTable(name=f"temp.{delta_table_name}")
 
         for agency in toptier_agencies:
             agency_name = agency["name"]
@@ -140,7 +142,7 @@ def process_data_copy_jobs(self, zip_file_path):
         self.filepaths_to_delete.append(zip_file_path)
 
         for delta_table_name, sql_file, final_name in self.download_file_list:
-            df = self.spark.sql(f"select * from {delta_table_name} where toptier_code = '{self._toptier_code}'")
+            df = self.spark.sql(f"select * from temp.{delta_table_name} where toptier_code = '{self._toptier_code}'")
             sql_file = None
             final_path = self._create_data_csv_dest_path(final_name)
             intermediate_data_file_path = final_path.parent / (final_path.name + "_temp")

usaspending_api/broker/helpers/delete_fabs_transactions.py

@@ -43,7 +43,7 @@ def get_delete_pks_for_afa_keys(afa_ids_to_delete):
             is_active is not true
     """
 
-    with connections[settings.DATA_BROKER_DB_ALIAS].cursor() as cursor:
+    with connections[settings.BROKER_DB_ALIAS].cursor() as cursor:
         cursor.execute(sql, [uppercased])
         rows = cursor.fetchall()
 

usaspending_api/broker/management/commands/derive_office_names.py

@@ -16,16 +16,16 @@ class Command(load_base.Command):
     """
 
     help = "Derives all FABS office names from the office codes in the Office table in Data broker. The \
-                DATA_BROKER_DATABASE_URL environment variable must set so we can pull Office data from their db."
+                BROKER_DB environment variable must set so we can pull Office data from their db."
 
     def handle(self, *args, **options):
         # Grab data broker database connections
         if not options["test"]:
             try:
-                db_conn = connections[settings.DATA_BROKER_DB_ALIAS]
+                db_conn = connections[settings.BROKER_DB_ALIAS]
                 db_cursor = db_conn.cursor()
             except Exception as err:
-                logger.critical("Could not connect to database. Is DATA_BROKER_DATABASE_URL set?")
+                logger.critical("Could not connect to database. Is BROKER_DB set?")
                 logger.critical(print(err))
                 raise
         else:

usaspending_api/broker/management/commands/load_broker_table.py

@@ -63,7 +63,7 @@ def handle(self, *args, **options):
             f'Copying "{broker_schema_name}"."{broker_table_name}" from Broker to '
             f'"{usas_schema_name}"."{usas_table_name}" in USAspending.'
         )
-        broker_conn = connections[settings.DATA_BROKER_DB_ALIAS]
+        broker_conn = connections[settings.BROKER_DB_ALIAS]
         usas_conn = connections[settings.DEFAULT_DB_ALIAS]
         table_exists_query = f"""
             SELECT EXISTS (