Merge pull request #61 from digital-land/save_perf_tables_to_parquet

Save perf tables to parquet

Author: ssadhu-sl

bin/concat-column-field.py

@@ -44,4 +44,4 @@ def process_column_fields(column_field_dir, input_dir):
                         w.writerow(row)
 
 if __name__ == "__main__":
-    process_column_fields()
+    process_column_fields()

bin/load_performance.py

@@ -31,12 +31,12 @@ def fetch_issue_data(db_path):
     query = """
         select  
         count(*) as count_issues, strftime('%d-%m-%Y', 'now') as date,
-        i.issue_type as issue_type, it.severity, it.responsibility, i.dataset, i.resource, GROUP_CONCAT(DISTINCT i.field) as fields
+        i.issue_type as issue_type, it.severity, it.responsibility, i.dataset, i.resource, i.field
         from issue i
         inner join resource r on i.resource = r.resource
         inner join issue_type it on i.issue_type = it.issue_type
         where r.end_date = ''
-        group by i.dataset,i.resource,i.issue_type
+        group by i.dataset,i.resource,i.issue_type,i.field
     """
     df_issue = pd.read_sql_query(query, conn)
     return df_issue
@@ -82,6 +82,7 @@ def fetch_endpoint_summary(perf_path):
     dataset,
     endpoint,
     endpoint_url, 
+    documentation_url,
     resource,
     latest_status,
     latest_exception,
@@ -107,19 +108,25 @@ def create_performance_tables(merged_data, cf_merged_data, endpoint_summary_data
     cf_merged_data_filtered = cf_merged_data[cf_merged_data['resource'] != ""]
     cf_merged_data_filtered = cf_merged_data_filtered[cf_merged_data_filtered['endpoint'].notna(
     )]
+    cf_merged_data_filtered[column_field_table_fields].to_parquet(os.path.join(PARQUET_PERFORMANCE_DIR,
+                                        "endpoint_dataset_resource_summary.parquet"), engine="pyarrow")
     cf_merged_data_filtered[column_field_table_fields].to_sql(
         column_field_table_name, conn, if_exists="replace", index=False)
 
     issue_table_name = "endpoint_dataset_issue_type_summary"
     issue_table_fields = ["organisation", "organisation_name", "cohort", "dataset", "collection", "pipeline", "endpoint", "endpoint_url", "resource", "resource_start_date",
-                          "resource_end_date", "latest_log_entry_date", "count_issues", "date", "issue_type", "severity", "responsibility", "fields"]
+                          "resource_end_date", "latest_log_entry_date", "count_issues", "date", "issue_type", "severity", "responsibility", "field"]
     issue_data_filtered = merged_data[merged_data['resource'] != ""]
     issue_data_filtered = issue_data_filtered[issue_data_filtered['endpoint'].notna(
     )]
+    issue_data_filtered[issue_table_fields].to_parquet(os.path.join(PARQUET_PERFORMANCE_DIR,
+                                        "endpoint_dataset_issue_type_summary.parquet"), engine="pyarrow")
     issue_data_filtered[issue_table_fields].to_sql(issue_table_name, conn, if_exists='replace', index=False, dtype={
         'count_issues': 'INTEGER'})
 
     endpoint_summary_table_name = "endpoint_dataset_summary"
+    endpoint_summary_data.to_parquet(os.path.join(PARQUET_PERFORMANCE_DIR,
+                                        "endpoint_dataset_summary.parquet"), engine="pyarrow")
     endpoint_summary_data.to_sql(
         endpoint_summary_table_name, conn, if_exists='replace', index=False)
 

bin/load_reporting_tables.py

@@ -26,6 +26,7 @@ def fetch_historic_endpoints_data_from_dl(db_path):
         sp.pipeline,
         l.endpoint,
         e.endpoint_url,
+        s.documentation_url,
         s.licence,
         l.status as latest_status,
         l.exception as latest_exception,
@@ -46,7 +47,7 @@ def fetch_historic_endpoints_data_from_dl(db_path):
         LEFT JOIN resource r on l.resource = r.resource
 
     GROUP BY
-        1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12
+        1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13
 
     ORDER BY
         s.organisation, o.name, o.dataset, s.collection, sp.pipeline, latest_log_entry_date DESC
@@ -134,6 +135,7 @@ def create_reporting_tables(historic_endpoints_data, latest_endpoint_data, perfo
             pipeline TEXT,
             endpoint TEXT,
             endpoint_url TEXT,
+            documentation_url TEXT,       
             licence TEXT,
             latest_status TEXT,
             latest_exception TEXT,
@@ -147,9 +149,9 @@ def create_reporting_tables(historic_endpoints_data, latest_endpoint_data, perfo
     """)
     cursor.executemany("""
             INSERT INTO reporting_historic_endpoints (
-                organisation, name, organisation_name, dataset, collection, pipeline, endpoint, endpoint_url, licence, latest_status, latest_exception, resource, 
+                organisation, name, organisation_name, dataset, collection, pipeline, endpoint, endpoint_url, documentation_url, licence, latest_status, latest_exception, resource, 
                 latest_log_entry_date, endpoint_entry_date, endpoint_end_date, resource_start_date, resource_end_date
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """, historic_endpoints_data)
 
      # Create the reporting_latest_endpoints table

check/performance.py

@@ -7,11 +7,11 @@
 logger = logging.getLogger("__name__")
 
 EXPECTED = {
-    "endpoint_dataset_issue_type_summary": ['organisation', 'organisation_name', 'cohort', 'dataset', 'collection', 'pipeline', 'endpoint', 'endpoint_url', 'resource', 'resource_start_date', 'resource_end_date', 'latest_log_entry_date', 'count_issues', 'date', 'issue_type', 'severity', 'responsibility', 'fields'],
+    "endpoint_dataset_issue_type_summary": ['organisation', 'organisation_name', 'cohort', 'dataset', 'collection', 'pipeline', 'endpoint', 'endpoint_url', 'resource', 'resource_start_date', 'resource_end_date', 'latest_log_entry_date', 'count_issues', 'date', 'issue_type', 'severity', 'responsibility', 'field'],
     "endpoint_dataset_resource_summary": ['organisation', 'organisation_name', 'cohort', 'dataset', 'collection', 'pipeline', 'endpoint', 'endpoint_url', 'resource', 'resource_start_date', 'resource_end_date', 'latest_log_entry_date', 'mapping_field', 'non_mapping_field'],
-    "endpoint_dataset_summary": ['organisation', 'dataset', 'endpoint', 'endpoint_url', 'resource', 'latest_status', 'latest_exception', 'latest_log_entry_date', 'entry_date', 'end_date', 'latest_resource_start_date', 'resource_end_date'],
+    "endpoint_dataset_summary": ['organisation', 'dataset', 'endpoint', 'endpoint_url', 'documentation_url', 'resource', 'latest_status', 'latest_exception', 'latest_log_entry_date', 'entry_date', 'end_date', 'latest_resource_start_date', 'resource_end_date'],
     "provision_summary": ['organisation', 'organisation_name', 'dataset', 'provision_reason', 'active_endpoint_count', 'error_endpoint_count', 'count_issue_error_internal', 'count_issue_error_external', 'count_issue_warning_internal', 'count_issue_warning_external', 'count_issue_notice_internal', 'count_issue_notice_external'],
-    "reporting_historic_endpoints": ['organisation', 'name', 'organisation_name', 'dataset', 'collection', 'pipeline', 'endpoint', 'endpoint_url', 'licence', 'latest_status', 'latest_exception', 'resource', 'latest_log_entry_date', 'endpoint_entry_date', 'endpoint_end_date', 'resource_start_date', 'resource_end_date'],
+    "reporting_historic_endpoints": ['organisation', 'name', 'organisation_name', 'dataset', 'collection', 'pipeline', 'endpoint', 'endpoint_url', 'documentation_url', 'licence', 'latest_status', 'latest_exception', 'resource', 'latest_log_entry_date', 'endpoint_entry_date', 'endpoint_end_date', 'resource_start_date', 'resource_end_date'],
     "reporting_latest_endpoints": ['organisation', 'name', 'organisation_name', 'dataset', 'collection', 'pipeline', 'endpoint', 'endpoint_url', 'licence', 'latest_status', 'days_since_200', 'latest_exception', 'resource', 'latest_log_entry_date', 'endpoint_entry_date', 'endpoint_end_date', 'resource_start_date', 'resource_end_date', 'rn']
 }