Merge pull request #1967 from LBHackney-IT/di-471-improvements-to-dq-outputs

annajgibson · web-flow · commit c5ded7258378 · 2024-11-06T16:22:44.000Z
Improvements to data quality results output for improved analysis
diff --git a/scripts/helpers/housing_gx_dq_inputs.py b/scripts/helpers/housing_gx_dq_inputs.py
@@ -1,6 +1,6 @@
 sql_config = {'person_reshape': {
     'sql': """SELECT *, substr(startdate, 1, 10) as startdate_parsed, substr(enddate, 1, 10) as enddate_parsed,
-    substr(dateofbirth, 1, 10) as dateofbirth_parsed FROM "housing-refined-zone"."person_reshape" WHERE import_date = (SELECT max(import_date) FROM "housing-refined-zone"."person_reshape") AND enddate IS NULL AND type IN ('Secure', 'Introductory')	and substr(dateofbirth, 1, 10) between '1850-01-01' and '2100-01-01' 	and substr(startdate, 1, 10) between '1900-01-01' and '2100-01-01'""",
+    substr(dateofbirth, 1, 10) as dateofbirth_parsed FROM "housing-refined-zone"."person_reshape" WHERE import_date = (SELECT max(import_date) FROM "housing-refined-zone"."person_reshape") AND enddate IS NULL AND type IN ('Secure', 'Introductory')""",
     'id_field': 'person_id'},
     'tenure_reshape': {
         'sql': """SELECT * FROM "housing-refined-zone"."tenure_reshape" where import_date>'20240412' and import_date=(select max(import_date) from "housing-refined-zone"."tenure_reshape" where import_date>'20240412') and isterminated=False and description in ('Secure', 'Introductory')""",
@@ -24,11 +24,18 @@
 
 partition_keys = ['import_year', 'import_month', 'import_day', 'import_date']
 
-dq_dimensions_map = {'expect_column_value_lengths_to_be_between': 'ACCURACY',
+dq_dimensions_map = {'expect_column_value_lengths_to_be_between': 'VALIDITY',
+                     'expect_first_name_column_value_lengths': 'VALIDITY',
+                     'expect_surname_column_value_lengths': 'VALIDITY',
                      'expect_column_values_to_be_unique': 'UNIQUENESS',
                      'expect_column_values_to_match_regex': 'VALIDITY',
+                     'expect_uprn_column_values_to_match_regex': 'VALIDITY',
                      'expect_column_values_to_be_in_set': 'CONSISTENCY',
+                     'expect_person_type_values_to_be_in_set': 'CONSISTENCY',
+                     'expect_preferred_title_values_to_be_in_set': 'CONSISTENCY',
                      'expect_select_column_values_to_be_unique_within_record': 'UNIQUENESS',
                      'expect_column_values_to_not_be_null': 'COMPLETENESS',
-                     'expect_column_values_to_be_between': 'VALIDITY'
+                     'expect_uprn_not_to_be_null': 'COMPLETENESS',
+                     'expect_column_values_to_be_between': 'VALIDITY',
+                     'expect_date_of_birth_to_be_between': 'VALIDITY'
                      }
diff --git a/scripts/jobs/housing/housing_apply_gx_dq_tests.py b/scripts/jobs/housing/housing_apply_gx_dq_tests.py
@@ -105,12 +105,8 @@ def main():
         )
 
         checkpoint_result = checkpoint.run(batch_parameters=batch_parameters)
-        results_dict = checkpoint_result.describe_dict()
-        # Serialize the result to handle any datetime objects
-        json_results = json.dumps(results_dict, default=json_serial)
-        logger.info(f"json_results: {json_results}")
-        results = json.loads(json_results)
-        table_results_df = pd.json_normalize(results['validation_results'][0]['expectations'])
+        results_dict = list(checkpoint_result.run_results.values())[0].to_json_dict()
+        table_results_df = pd.json_normalize(results_dict['results'])
         table_results_df_list.append(table_results_df)
 
         # generate id lists for each unexpected result set
@@ -125,33 +121,41 @@ def main():
     results_df = pd.concat(table_results_df_list)
 
     # map DQ dimension type
-    results_df['dq_dimension_type'] = results_df['expectation_type'].map(dq_dimensions_map)
+    results_df['dq_dimension_type'] = results_df['expectation_config.type'].map(dq_dimensions_map)
 
     # add clean dataset name
-    results_df['table_name'] = results_df['kwargs.batch_id'].map(
+    results_df['dataset_name'] = results_df['expectation_config.kwargs.batch_id'].map(
         lambda x: x.removeprefix('pandas-').removesuffix('_df_asset'))
 
+    # add composite key for each specific test (so can be tracked over time)
+    results_df.insert(loc=0, column='expectation_key',
+                      value=results_df.set_index(['expectation_config.type', 'dataset_name']).index.factorize()[0] + 1)
+    results_df['expectation_id'] = results_df['expectation_config.type'] + "_" + results_df['dataset_name']
+
     results_df['import_year'] = datetime.today().year
     results_df['import_month'] = datetime.today().month
     results_df['import_day'] = datetime.today().day
     results_df['import_date'] = datetime.today().strftime('%Y%m%d')
 
     # set dtypes for Athena
-    dtype_dict = {'expectation_type': 'string',
-                  'kwargs.batch_id': 'string',
-                  'kwargs.column': 'string',
-                  'kwargs.min_value': 'string',
-                  'kwargs.max_value': 'string',
+    dtype_dict = {'expectation_config.type': 'string',
+                  'expectation_config.kwargs.batch_id': 'string',
+                  'expectation_config.kwargs.column': 'string',
+                  'expectation_config.kwargs.min_value': 'string',
+                  'expectation_config.kwargs.max_value': 'string',
                   'result.element_count': 'bigint',
                   'result.unexpected_count': 'bigint',
                   'result.missing_count': 'bigint',
                   'result.partial_unexpected_list': 'array<string>',
                   'result.unexpected_list': 'array<string>',
                   'result.unexpected_index_list': 'array<bigint>',
                   'result.unexpected_index_query': 'string',
-                  'kwargs.regex': 'string',
-                  'kwargs.value_set': 'string',
-                  'kwargs.column_list': 'string',
+                  'expectation_config.kwargs.regex': 'string',
+                  'expectation_config.kwargs.value_set': 'string',
+                  'expectation_config.kwargs.column_list': 'string',
+                  'exception_info.raised_exception': 'string',
+                  'exception_info.exception_traceback': 'string',
+                  'exception_info.exception_message': 'string',
                   'import_year': 'string',
                   'import_month': 'string',
                   'import_day': 'string',
@@ -164,7 +168,7 @@ def main():
         dataset=True,
         database=target_database,
         table=target_table,
-        mode="overwrite_partitions",
+        mode="overwrite",
         partition_cols=partition_keys,
         dtype=dtype_dict
     )
diff --git a/scripts/jobs/housing/housing_person_reshape_gx_suite.py b/scripts/jobs/housing/housing_person_reshape_gx_suite.py
@@ -10,89 +10,114 @@
 args = getResolvedOptions(sys.argv, arg_key)
 locals().update(args)
 
+
+class ExpectFirstNameColumnValueLength(gxe.ExpectColumnValueLengthsToBeBetween):
+    column: str = "firstname"
+    min_value: int = 1
+    description: str = "Expect first name to be at least 1 character length"
+
+
+class ExpectSurnameColumnValueLength(gxe.ExpectColumnValueLengthsToBeBetween):
+    column: str = "surname"
+    min_value: int = 1
+    description: str = "Expect surname to be at least 1 character length"
+
+
+class ExpectUPRNColumnValueLengthsBetween(gxe.ExpectColumnValueLengthsToBeBetween):
+    column: str = "uprn"
+    min_value: int = 11
+    max_value: int = 12
+    description: str = "Expect UPRN to be between 11 and 12 characters length inclusive"
+
+
+class ExpectUPRNColumnValuesToMatchRegex(gxe.ExpectColumnValuesToMatchRegex):
+    column: str = "uprn"
+    regex: str = r"^[1-9]\d{10,11}"
+    description: str = "Expect UPRN to match regex ^[1-9]\d{10,11} (starting with digit 1-9, followed by 10 or 11 digits"
+
+
+class ExpectUPRNNotToBeNull(gxe.ExpectColumnValuesToNotBeNull):
+    column: str = "uprn"
+    description: str = "Expect UPRN column to be complete with no missing values"
+
+
+class ExpectPersonTypeValuesToBeInSet(gxe.ExpectColumnValuesToBeInSet):
+    column: str = 'person_type'
+    value_set: list = ['Tenant', 'HouseholdMember', 'Leaseholder', 'Freeholder', 'Occupant', 'HousingOfficer',
+                       'HousingAreaManager']
+    description: str = "Expect person types values to contain one of Tenant, HouseholdMember, Leaseholder, Freeholder, Occupant HousingOfficer, HousingAreaManager"
+
+
+class ExpectPreferredTitleValuesToBeInSet(gxe.ExpectColumnValuesToBeInSet):
+    column: str = 'preferredtitle'
+    value_set: list = ['Dr', 'Master', 'Miss', 'Mr', 'Mrs', 'Ms', 'Rabbi', 'Reverend', 'Mx']
+    description: str = "Expect preferred titles to be one of Dr, Master, Miss, Mr, Mrs, Ms, Mx, Rabbi, Reverend"
+
+
+class ExpectPersonIDColumnValuesToBeUnique(gxe.ExpectColumnValuesToBeUnique):
+    column: str = 'person_id'
+    description: str = "Expect Person ID to be unique within dataset"
+
+
+class ExpectPersonIDColumnValuesToNotBeNull(gxe.ExpectColumnValuesToNotBeNull):
+    column: str = 'person_id'
+    description: str = "Expect Person ID be complete with no missing values"
+
+
+class ExpectPersonIDAndPropertyReferenceColumnValuesToBeUniqueWithinRecord(
+    gxe.ExpectSelectColumnValuesToBeUniqueWithinRecord):
+    column_list: list = ['person_id', 'propertyreference']
+    description: str = "Expect Person ID and Property Reference to be unique within dataset"
+
+
+class ExpectPropertyRefColumnValuesToNotBeNull(gxe.ExpectColumnValuesToNotBeNull):
+    column: str = 'propertyreference'
+    description: str = "Expect Property Reference be complete with no missing values"
+
+
+class ExpectPersonIDAndPaymentReferenceColumnValuesToBeUniqueWithinRecord(
+    gxe.ExpectSelectColumnValuesToBeUniqueWithinRecord):
+    column_list: list = ['person_id', 'paymentreference']
+    description: str = "Expect Person ID and Payment Reference to be unique within dataset"
+
+
+class ExpectUPRNColumnValuesToNotBeNull(gxe.ExpectColumnValuesToNotBeNull):
+    column: str = 'uprn'
+    description: str = "Expect UPRN be complete with no missing values"
+
+
+class ExpectDateOfBirthColumnValuesToNotBeNull(gxe.ExpectColumnValuesToNotBeNull):
+    column: str = 'dateofbirth_parsed'
+    description: str = "Expect dateofbirth_parsed be complete with no missing values"
+
+
+class ExpectDateOfBirthToBeBetween(gxe.ExpectColumnValuesToBeBetween):
+    column: str = 'dateofbirth_parsed'
+    min_value: str = datetime(1900, 1, 1, 0, 0, 0).isoformat()
+    max_value: str = datetime.today().isoformat()
+    condition_parser: str = "pandas"
+    row_condition: str = 'df["dateofbirth_parsed"].str[:10]  >= "1850-01-01" and df["dateofbirth_parsed"].str[:10] < "2025-01-01" and df["startdate_parsed"].str[:10] > "1900-01-01" and df["startdate_parsed"].str[:10] < "2100-01-01"'
+    description: str = "Expect dateofbirth_parsed be complete with no missing values"
+
+
 # add to GX context
 context = gx.get_context(mode="file", project_root_dir=s3_target_location)
 
 suite = gx.ExpectationSuite(name='person_reshape_suite')
-suite.add_expectation(
-    gxe.ExpectColumnValueLengthsToBeBetween(
-        column="firstname",
-        min_value=1)
-)
-suite.add_expectation(
-    gxe.ExpectColumnValueLengthsToBeBetween(
-        column="surname",
-        min_value=1)
-)
-suite.add_expectation(
-    gxe.ExpectColumnValueLengthsToBeBetween(
-        column="uprn",
-        min_value=11,
-        max_value=12)
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToMatchRegex(
-        column="uprn",
-        regex=r"^[1-9]\d{10,11}")
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToNotBeNull(
-        column='uprn')
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToBeInSet(
-        column='type',
-        value_set=['Asylum Seeker', 'Commercial Let', 'Temp Decant', 'Freehold', 'Freehold (Serv)', 'Introductory',
-                   'Leasehold (RTB)', 'Lse 100% Stair', 'License Temp Ac', 'Mesne Profit Ac', 'Non-Secure',
-                   'Private Garage', 'Registered Social Landlord', 'RenttoMortgage', 'Secure', 'Shared Owners',
-                   'Short Life Lse', 'Private Sale LH', 'Shared Equity', 'Tenant Acc Flat', 'Temp B&B', 'Tenant Garage',
-                   'Temp Hostel Lse', 'Temp Hostel', 'Temp Annex', 'Temp Private Lt', 'Temp Traveller'])
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToBeInSet(
-        column='person_type',
-        value_set=['Tenant', 'HouseholdMember', 'Leaseholder', 'Freeholder', 'Occupant', 'HousingOfficer',
-                   'HousingAreaManager'])
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToBeInSet(
-        column='preferredtitle',
-        value_set=['Dr', 'Master', 'Miss', 'Mr', 'Mrs', 'Ms', 'Other', 'Rabbi', 'Reverend'])
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToBeUnique(
-        column='person_id')
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToNotBeNull(
-        column='person_id')
-)
-suite.add_expectation(
-    gxe.ExpectSelectColumnValuesToBeUniqueWithinRecord(
-        column_list=['person_id', 'propertyreference'])
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToNotBeNull(
-        column='propertyreference')
-)
-suite.add_expectation(
-    gxe.ExpectSelectColumnValuesToBeUniqueWithinRecord(
-        column_list=['person_id', 'paymentreference'])
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToNotBeNull(
-        column='uprn')
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToNotBeNull(
-        column='dateofbirth_parsed')
-)
-suite.add_expectation(
-    gxe.ExpectColumnValuesToBeBetween(
-        column='dateofbirth_parsed',
-        min_value=datetime(1900, 1, 1, 0, 0, 0).isoformat(),
-        max_value=datetime.today().isoformat()
-    )
-)
+suite.add_expectation(ExpectFirstNameColumnValueLength())
+suite.add_expectation(ExpectSurnameColumnValueLength())
+suite.add_expectation(ExpectUPRNColumnValueLengthsBetween())
+suite.add_expectation(ExpectUPRNColumnValuesToMatchRegex())
+suite.add_expectation(ExpectUPRNNotToBeNull())
+suite.add_expectation(ExpectPersonTypeValuesToBeInSet())
+suite.add_expectation(ExpectPreferredTitleValuesToBeInSet())
+suite.add_expectation(ExpectPersonIDColumnValuesToBeUnique())
+suite.add_expectation(ExpectPersonIDColumnValuesToNotBeNull())
+suite.add_expectation(ExpectPersonIDAndPropertyReferenceColumnValuesToBeUniqueWithinRecord())
+suite.add_expectation(ExpectPropertyRefColumnValuesToNotBeNull())
+suite.add_expectation(ExpectPersonIDAndPaymentReferenceColumnValuesToBeUniqueWithinRecord())
+suite.add_expectation(ExpectUPRNColumnValuesToNotBeNull())
+suite.add_expectation(ExpectDateOfBirthColumnValuesToNotBeNull())
+suite.add_expectation(ExpectDateOfBirthToBeBetween())
 
 suite = context.suites.add(suite)
diff --git a/terraform/etl/54-aws-glue-housing-apply-gx-dq-tests.tf b/terraform/etl/54-aws-glue-housing-apply-gx-dq-tests.tf
@@ -20,7 +20,7 @@ module "housing_apply_gx_dq_tests" {
     "--job-bookmark-option"              = "job-bookmark-enable"
     "--enable-glue-datacatalog"          = "true"
     "--enable-continuous-cloudwatch-log" = "true"
-    "--additional-python-modules"        = "great_expectations==1.2.0,PyAthena,awswrangler"
+    "--additional-python-modules"        = "great_expectations==1.2.1,PyAthena,numpy==1.26.1,awswrangler==3.10.0"
     "--region_name"                      = data.aws_region.current.name
     "--s3_endpoint"                      = "https://s3.${data.aws_region.current.name}.amazonaws.com"
     "--s3_target_location"               = "s3://${module.raw_zone_data_source.bucket_id}/housing/data-quality-tests/"
