Merge pull request #50 from digital-land/upload_parquet_to_S3

Upload parquet to s3

Author: ssadhu-sl

.github/workflows/run.yml

@@ -63,6 +63,12 @@ jobs:
         HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
       run: make save-dataset
 
+    - name: Save Parquet files to Development S3
+      env:
+        COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
+        HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
+      run: make save-tables-to-parquet
+
     # Staging
     - name: Configure Staging AWS Credentials
       uses: aws-actions/configure-aws-credentials@v1-node16
@@ -76,6 +82,12 @@ jobs:
         COLLECTION_DATASET_BUCKET_NAME: ${{secrets.STAGING_DATA_S3_BUCKET}}
         HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.STAGING_DATA_S3_BUCKET}}
       run: make save-dataset
+
+    - name: Save Parquet files to Staging S3
+      env:
+        COLLECTION_DATASET_BUCKET_NAME: ${{secrets.STAGING_DATA_S3_BUCKET}}
+        HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.STAGING_DATA_S3_BUCKET}}
+      run: make save-tables-to-parquet
 
     # Production
     - name: Configure Production AWS Credentials
@@ -90,6 +102,12 @@ jobs:
         COLLECTION_DATASET_BUCKET_NAME: ${{secrets.PRODUCTION_DATA_S3_BUCKET}}
         HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.PRODUCTION_DATA_S3_BUCKET}}
       run: make save-dataset
+    
+    - name: Save Parquet files to Prod S3
+      env:
+        COLLECTION_DATASET_BUCKET_NAME: ${{secrets.PRODUCTION_DATA_S3_BUCKET}}
+        HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.PRODUCTION_DATA_S3_BUCKET}}
+      run: make save-tables-to-parquet
 
   check-digital-land-builder-errors:
     runs-on: ubuntu-latest

.github/workflows/run_dev.yml

@@ -7,7 +7,7 @@ env:
   DLB_BOT_USERNAME: ${{ secrets.DLB_BOT_USERNAME }}
 jobs:
   build:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
     steps:
 
     - name: Free up disk space
@@ -61,4 +61,10 @@ jobs:
         HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
       run: make save-dataset
 
+    - name: Save Parquet files to Development S3
+      env:
+        COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
+        HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
+      run: make save-tables-to-parquet
+
 

.github/workflows/run_performance.yml

@@ -63,7 +63,13 @@ jobs:
         HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
       run: make save-dataset
 
-      # Staging
+    - name: Save Parquet files to Development S3
+      env:
+        COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
+        HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
+      run: make save-tables-to-parquet
+
+   # Staging
     - name: Configure Staging AWS Credentials
       uses: aws-actions/configure-aws-credentials@v1-node16
       with:
@@ -76,6 +82,12 @@ jobs:
         COLLECTION_DATASET_BUCKET_NAME: ${{secrets.STAGING_DATA_S3_BUCKET}}
         HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.STAGING_DATA_S3_BUCKET}}
       run: make save-dataset
+
+    - name: Save Parquet files to Staging S3
+      env:
+        COLLECTION_DATASET_BUCKET_NAME: ${{secrets.STAGING_DATA_S3_BUCKET}}
+        HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.STAGING_DATA_S3_BUCKET}}
+      run: make save-tables-to-parquet
 
     # Production
     - name: Configure Production AWS Credentials
@@ -90,3 +102,10 @@ jobs:
         COLLECTION_DATASET_BUCKET_NAME: ${{secrets.PRODUCTION_DATA_S3_BUCKET}}
         HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.PRODUCTION_DATA_S3_BUCKET}}
       run: make save-dataset
+
+    - name: Save Parquet files to Prod S3
+      env:
+        COLLECTION_DATASET_BUCKET_NAME: ${{secrets.PRODUCTION_DATA_S3_BUCKET}}
+        HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.PRODUCTION_DATA_S3_BUCKET}}
+      run: make save-tables-to-parquet
+

.github/workflows/run_performance_dev.yml

@@ -63,4 +63,10 @@ jobs:
         HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
       run: make save-dataset
 
+    - name: Save Parquet files to Development S3
+      env:
+        COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
+        HOISTED_COLLECTION_DATASET_BUCKET_NAME: ${{secrets.DEVELOPMENT_DATA_S3_BUCKET}}
+      run: make save-tables-to-parquet
+
 

Makefile

@@ -11,6 +11,18 @@ include makerules/development.mk
 
 DB=dataset/digital-land.sqlite3
 DB_PERF = dataset/performance.sqlite3
+
+ifeq ($(PARQUET_DIR),)
+PARQUET_DIR=data/
+endif
+ifeq ($(PARQUET_SPECIFICATION_DIR),)
+export PARQUET_SPECIFICATION_DIR=$(PARQUET_DIR)specification/
+endif
+ifeq ($(PARQUET_PERFORMANCE_DIR),)
+export PARQUET_PERFORMANCE_DIR=$(PARQUET_DIR)performance/
+endif
+
+
 DATASTORE_URL = https://files.planning.data.gov.uk/
 
 first-pass::
@@ -35,11 +47,13 @@ third-pass:: $(DB_PERF)
 
 $(DB):	bin/load.py
 	@rm -f $@
+	mkdir -p $(PARQUET_SPECIFICATION_DIR)
 	python3 bin/load.py $@
 
 $(DB_PERF): bin/load_reporting_tables.py bin/load_performance.py
 	bin/download-digital-land.sh
 	@rm -f $@  
+	mkdir -p $(PARQUET_PERFORMANCE_DIR)
 	python3 bin/load_reporting_tables.py $@ $(DB)
 	python3 bin/load_performance.py $@ $(DB)
 
@@ -52,6 +66,9 @@ clobber::
 	rm -rf dataset/
 	rm -rf $(DB)
 	rm -rf $(DB_PERF)
+	rm -rf $(PARQUET_SPECIFICATION_DIR)
+	rm -rf $(PARQUET_PERFORMANCE_DIR)
+	rm -rf $(PARQUET_DIR)
 
 clobber-performance::
 	rm -rf $(DB_PERF)
@@ -62,7 +79,7 @@ aws-build::
 push::
 	aws s3 cp $(DB) s3://digital-land-collection/digital-land.sqlite3
 	aws s3 cp $(DB_PERF) s3://digital-land-collection/performance.sqlite3
-
+	
 specification::
 	# additional
 	curl -qfsL '$(SOURCE_URL)/specification/main/specification/issue-type.csv' > specification/issue-type.csv

bin/load.py

@@ -11,6 +11,8 @@
 import pandas as pd
 from digital_land.package.sqlite import SqlitePackage
 
+PARQUET_SPECIFICATION_DIR = os.getenv("PARQUET_SPECIFICATION_DIR")
+
 
 tables = {
     "organisation": "var/cache",
@@ -79,6 +81,11 @@
 }
 
 
+def create_parquet_from_table(df, name, output_dir):
+    parquet_file_path = os.path.join(output_dir, f"{name}.parquet")
+    df.to_parquet(parquet_file_path, engine="pyarrow")
+
+
 if __name__ == "__main__":
     level = logging.INFO
     logging.basicConfig(level=level, format="%(asctime)s %(levelname)s %(message)s")
@@ -88,6 +95,8 @@
 
     conn = sqlite3.connect(path)
 
+    specification_df = pd.read_sql_query("SELECT * FROM specification", conn)
+    create_parquet_from_table(specification_df, "specification", PARQUET_SPECIFICATION_DIR)
     operational_issue_log = pd.read_csv("performance/operational_issue/operational-issue.csv")
     operational_issue_log.to_sql("operational_issue", conn, if_exists="replace", index=False)
 
@@ -197,3 +206,4 @@
         where t1.rn = 1              
     """)
     conn.close()
+

bin/load_performance.py

@@ -6,16 +6,19 @@
 import logging
 import sqlite3
 import pandas as pd
+import os
 
 indexes = {
     "provision_summary": ["organisation", "organisation_name", "dataset"]
 }
 
+PARQUET_PERFORMANCE_DIR = os.getenv("PARQUET_PERFORMANCE_DIR")
+
 
 def fetch_provision_data(db_path):
     conn = sqlite3.connect(db_path)
     query = """
-        select p.organisation, o.name as organisation_name, p.cohort, p.dataset from provision p
+        select p.organisation, o.name as organisation_name, p.cohort, p.dataset,p.provision_reason from provision p
         inner join organisation o on o.organisation = p.organisation
         order by p.organisation
     """
@@ -121,7 +124,7 @@ def create_performance_tables(merged_data, cf_merged_data, endpoint_summary_data
         endpoint_summary_table_name, conn, if_exists='replace', index=False)
 
     # Filter out endpoints with an end date as we don't want to count them in provision summary
-    final_result = merged_data.groupby(['organisation', 'organisation_name', 'dataset']).agg(
+    final_result = merged_data.groupby(['organisation', 'organisation_name', 'dataset', 'provision_reason']).agg(
         active_endpoint_count=pd.NamedAgg(
             column='endpoint',
             aggfunc=lambda x: x[(merged_data.loc[x.index,
@@ -191,6 +194,7 @@ def create_performance_tables(merged_data, cf_merged_data, endpoint_summary_data
     })
 
     provision_table_name = "provision_summary"
+    final_result.to_parquet(os.path.join(PARQUET_PERFORMANCE_DIR,"provision_summary.parquet"), engine="pyarrow")
     final_result.to_sql(provision_table_name, conn,
                         if_exists='replace', index=False)
     conn.close()

requirements.txt

@@ -7,4 +7,5 @@ chardet
 PyPDF2
 pandas
 tqdm
+pyarrow
 -e git+https://github.com/digital-land/digital-land-python.git#egg=digital-land