Merge branch 'main' into CE-2642-adding-s3-bucket-for-mwaa-etl-scripts

Author: LBH-wgreeff

scripts/jobs/parking/parking_permit_denormalised_gds_street_llpg.py

@@ -1,18 +1,21 @@
-from calendar import c
 import sys
+from calendar import c
+
+from awsglue import DynamicFrame
+from awsglue.context import GlueContext
+from awsglue.job import Job
 from awsglue.transforms import *
 from awsglue.utils import getResolvedOptions
 from pyspark.context import SparkContext
-from awsglue.context import GlueContext
-from awsglue.job import Job
-from awsglue import DynamicFrame
+
 from scripts.helpers.helpers import (
-    get_glue_env_var,
-    get_latest_partitions,
     PARTITION_KEYS,
     create_pushdown_predicate,
+    get_glue_env_var,
+    get_latest_partitions,
 )
 
+
 def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFrame:
     for alias, frame in mapping.items():
         frame.toDF().createOrReplaceTempView(alias)
@@ -29,20 +32,18 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
 environment = get_glue_env_var("environment")
 
 # Script generated for node S3 bucket - refined - parking_permit_denormalised_data
-S3bucketrefinedparking_permit_denormalised_data_node1 = (
-    glueContext.create_dynamic_frame.from_catalog(
-        database="dataplatform-"+environment+"-liberator-refined-zone",
-        table_name="parking_permit_denormalised_data",
-        transformation_ctx="S3bucketrefinedparking_permit_denormalised_data_node1",
-        #teporarily removed while table partitions are fixed
-        #push_down_predicate=create_pushdown_predicate("import_date", 7),
-    )
+S3bucketrefinedparking_permit_denormalised_data_node1 = glueContext.create_dynamic_frame.from_catalog(
+    database="dataplatform-" + environment + "-liberator-refined-zone",
+    table_name="parking_permit_denormalised_data",
+    transformation_ctx="S3bucketrefinedparking_permit_denormalised_data_node1",
+    # teporarily removed while table partitions are fixed
+    # push_down_predicate=create_pushdown_predicate("import_date", 7),
 )
 
 # Script generated for node Amazon S3 - raw - liberator_permit_llpg
 AmazonS3rawliberator_permit_llpg_node1657535904691 = (
     glueContext.create_dynamic_frame.from_catalog(
-        database="dataplatform-"+environment+"-liberator-raw-zone",
+        database="dataplatform-" + environment + "-liberator-raw-zone",
         table_name="liberator_permit_llpg",
         transformation_ctx="AmazonS3rawliberator_permit_llpg_node1657535904691",
         push_down_predicate=create_pushdown_predicate("import_date", 7),
@@ -51,10 +52,10 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
 
 # Script generated for node Amazon S3 - unrestricted_address_api_dbo_hackney_address
 AmazonS3unrestricted_address_api_dbo_hackney_address_node1657535910004 = glueContext.create_dynamic_frame.from_catalog(
-    database="dataplatform-"+environment+"-raw-zone-unrestricted-address-api",
+    database="dataplatform-" + environment + "-raw-zone-unrestricted-address-api",
     table_name="unrestricted_address_api_dbo_hackney_address",
     transformation_ctx="AmazonS3unrestricted_address_api_dbo_hackney_address_node1657535910004",
-    push_down_predicate=create_pushdown_predicate("import_date", 7),
+    push_down_predicate=create_pushdown_predicate("import_date", 30),
 )
 
 # Script generated for node Amazon S3 - parking raw - ltn_london_fields
@@ -69,7 +70,7 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
 # Script generated for node Apply Mapping - New
 SqlQuery0 = """
 /*08/04/2022 - added ,case when latest_permit_status in('Approved','Renewed','Created','ORDER_APPROVED','PENDING_VRM_CHANGE','RENEW_EVID','PENDING_ADDR_CHANGE') and live_permit_flag = 1 then 1 else 0 end as live_flag */
-with street as (select  
+with street as (select
     UPRN as SR_UPRN,
 	ADDRESS1 as SR_ADDRESS1,
 	ADDRESS2 as SR_ADDRESS2,
@@ -96,27 +97,27 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
 	STREET_END_X as SR_STREET_END_X,
 	STREET_END_Y as SR_STREET_END_Y,
 	WARD_CODE as SR_WARD_CODE,
-if(WARD_CODE = 'E05009367',    'BROWNSWOOD WARD',    
-if(WARD_CODE = 'E05009368',    'CAZENOVE WARD',    
-if(WARD_CODE = 'E05009369',    'CLISSOLD WARD',    
-if(WARD_CODE = 'E05009370',    'DALSTON WARD',    
-if(WARD_CODE = 'E05009371',    'DE BEAUVOIR WARD',    
-if(WARD_CODE = 'E05009372',    'HACKNEY CENTRAL WARD',    
-if(WARD_CODE = 'E05009373',    'HACKNEY DOWNS WARD',    
-if(WARD_CODE = 'E05009374',    'HACKNEY WICK WARD',    
-if(WARD_CODE = 'E05009375',    'HAGGERSTON WARD',    
-if(WARD_CODE = 'E05009376',    'HOMERTON WARD',    
-if(WARD_CODE = 'E05009377',    'HOXTON EAST AND SHOREDITCH WARD',    
-if(WARD_CODE = 'E05009378',    'HOXTON WEST WARD',    
-if(WARD_CODE = 'E05009379',    'KINGS PARK WARD',    
-if(WARD_CODE = 'E05009380',    'LEA BRIDGE WARD',    
-if(WARD_CODE = 'E05009381',    'LONDON FIELDS WARD',    
-if(WARD_CODE = 'E05009382',    'SHACKLEWELL WARD',    
-if(WARD_CODE = 'E05009383',    'SPRINGFIELD WARD',    
-if(WARD_CODE = 'E05009384',    'STAMFORD HILL WEST',    
-if(WARD_CODE = 'E05009385',    'STOKE NEWINGTON WARD',    
-if(WARD_CODE = 'E05009386',    'VICTORIA WARD',    
-if(WARD_CODE = 'E05009387',    'WOODBERRY DOWN WARD',WARD_CODE))))))))))))))))))))) as SR_ward_name, 	 
+if(WARD_CODE = 'E05009367',    'BROWNSWOOD WARD',
+if(WARD_CODE = 'E05009368',    'CAZENOVE WARD',
+if(WARD_CODE = 'E05009369',    'CLISSOLD WARD',
+if(WARD_CODE = 'E05009370',    'DALSTON WARD',
+if(WARD_CODE = 'E05009371',    'DE BEAUVOIR WARD',
+if(WARD_CODE = 'E05009372',    'HACKNEY CENTRAL WARD',
+if(WARD_CODE = 'E05009373',    'HACKNEY DOWNS WARD',
+if(WARD_CODE = 'E05009374',    'HACKNEY WICK WARD',
+if(WARD_CODE = 'E05009375',    'HAGGERSTON WARD',
+if(WARD_CODE = 'E05009376',    'HOMERTON WARD',
+if(WARD_CODE = 'E05009377',    'HOXTON EAST AND SHOREDITCH WARD',
+if(WARD_CODE = 'E05009378',    'HOXTON WEST WARD',
+if(WARD_CODE = 'E05009379',    'KINGS PARK WARD',
+if(WARD_CODE = 'E05009380',    'LEA BRIDGE WARD',
+if(WARD_CODE = 'E05009381',    'LONDON FIELDS WARD',
+if(WARD_CODE = 'E05009382',    'SHACKLEWELL WARD',
+if(WARD_CODE = 'E05009383',    'SPRINGFIELD WARD',
+if(WARD_CODE = 'E05009384',    'STAMFORD HILL WEST',
+if(WARD_CODE = 'E05009385',    'STOKE NEWINGTON WARD',
+if(WARD_CODE = 'E05009386',    'VICTORIA WARD',
+if(WARD_CODE = 'E05009387',    'WOODBERRY DOWN WARD',WARD_CODE))))))))))))))))))))) as SR_ward_name,
 	PARISH_CODE as SR_PARISH_CODE,
 	PARENT_UPRN as SR_PARENT_UPRN,
 	PAO_START as SR_PAO_START,
@@ -142,14 +143,14 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
   SELECT * FROM unrestricted_address_api_dbo_hackney_address where unrestricted_address_api_dbo_hackney_address.import_date = (SELECT max(unrestricted_address_api_dbo_hackney_address.import_date) FROM unrestricted_address_api_dbo_hackney_address) and lpi_logical_status like 'Approved Preferred'
   )
 SELECT street.usrn as sr_usrn, SR_ADDRESS1, SR_ADDRESS2, llpg.street_description, SR_WARD_CODE, SR_ward_name, llpg.property_shell, llpg.blpu_class, llpg.usage_primary, llpg.usage_description
-,concat(cast(street.usrn as string),' - ', llpg.street_description) as street, concat(cast(llpg.uprn as string),' - ',llpg.usage_description) as add_type , concat(llpg.blpu_class,' - ',llpg.planning_use_class ) as add_class, 
+,concat(cast(street.usrn as string),' - ', llpg.street_description) as street, concat(cast(llpg.uprn as string),' - ',llpg.usage_description) as add_type , concat(llpg.blpu_class,' - ',llpg.planning_use_class ) as add_class,
 
 case when cpz !='' and cpz_name != '' then concat(cpz,' - ', cpz_name)
 when cpz !='' and cpz_name = '' then concat(cpz)
 when cpz ='' and cpz_name != '' then concat(cpz_name)
 else 'NONE'
-end as zone_name, 
-case 
+end as zone_name,
+case
 
 when address_line_2 ='' and business_name ='' and hasc_organisation_name ='' and  doctors_surgery_name ='' then concat(permit_reference,' - ',address_line_1,', ',parking_permit_denormalised_data.postcode,' - ',email_address_of_applicant)
 when business_name !='' and hasc_organisation_name !='' and address_line_2 ='' then concat(permit_reference,' - ',business_name,' - ',hasc_organisation_name,' - ',address_line_1,', ',parking_permit_denormalised_data.postcode,' - ',email_address_of_applicant)
@@ -167,37 +168,37 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
 
 
 when business_name !='' then concat(permit_reference,' - ',business_name,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode,' - ',email_address_of_applicant)
-when hasc_organisation_name !='' then concat(permit_reference,' - ',hasc_organisation_name,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode,' - ',email_address_of_applicant) 
+when hasc_organisation_name !='' then concat(permit_reference,' - ',hasc_organisation_name,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode,' - ',email_address_of_applicant)
 when doctors_surgery_name !='' then concat(permit_reference,' - ',doctors_surgery_name,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode,' - ',email_address_of_applicant)
 
-else concat(permit_reference,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode,' - ',email_address_of_applicant) 
+else concat(permit_reference,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode,' - ',email_address_of_applicant)
 end as permit_summary
 
 
-,case 
+,case
 when address_line_2 ='' then concat(permit_reference,' - ',address_line_1,', ',parking_permit_denormalised_data.postcode)
-when address_line_3 ='' then concat(permit_reference,' - ',address_line_1,', ',address_line_2,', ',parking_permit_denormalised_data.postcode) 
-else concat(permit_reference,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode) 
+when address_line_3 ='' then concat(permit_reference,' - ',address_line_1,', ',address_line_2,', ',parking_permit_denormalised_data.postcode)
+else concat(permit_reference,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode)
 end as permit_full_address
-,case 
+,case
 when address_line_2 ='' then concat(address_line_1,', ',parking_permit_denormalised_data.postcode)
-when address_line_3 ='' then concat(address_line_1,', ',address_line_2,', ',parking_permit_denormalised_data.postcode) 
-else concat(address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode) 
+when address_line_3 ='' then concat(address_line_1,', ',address_line_2,', ',parking_permit_denormalised_data.postcode)
+else concat(address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode)
 end as full_address
 
-,case 
-when address_line_2 ='' then concat(permit_reference,' - ',usage_primary,' - ',address_line_1,', ',parking_permit_denormalised_data.postcode) 
-when address_line_3 ='' then concat(permit_reference,' - ',usage_primary,' - ',address_line_1,', ',address_line_2,', ',parking_permit_denormalised_data.postcode) 
-else concat(permit_reference,' - ',usage_primary,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode) 
+,case
+when address_line_2 ='' then concat(permit_reference,' - ',usage_primary,' - ',address_line_1,', ',parking_permit_denormalised_data.postcode)
+when address_line_3 ='' then concat(permit_reference,' - ',usage_primary,' - ',address_line_1,', ',address_line_2,', ',parking_permit_denormalised_data.postcode)
+else concat(permit_reference,' - ',usage_primary,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode)
 end as permit_full_address_type
 
-,case 
-when address_line_2 ='' then concat(usage_primary,' - ',address_line_1,', ',parking_permit_denormalised_data.postcode) 
-when address_line_3 ='' then concat(usage_primary,' - ',address_line_1,', ',address_line_2,', ',parking_permit_denormalised_data.postcode) 
-else concat(usage_primary,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode) 
+,case
+when address_line_2 ='' then concat(usage_primary,' - ',address_line_1,', ',parking_permit_denormalised_data.postcode)
+when address_line_3 ='' then concat(usage_primary,' - ',address_line_1,', ',address_line_2,', ',parking_permit_denormalised_data.postcode)
+else concat(usage_primary,' - ',address_line_1,', ',address_line_2,', ',address_line_3,', ',parking_permit_denormalised_data.postcode)
 end as full_address_type
 
-,case when latest_permit_status in('Approved','Renewed','Created','ORDER_APPROVED','PENDING_VRM_CHANGE','RENEW_EVID','PENDING_ADDR_CHANGE') and live_permit_flag = 1 then 1 else 0 end as live_flag 
+,case when latest_permit_status in('Approved','Renewed','Created','ORDER_APPROVED','PENDING_VRM_CHANGE','RENEW_EVID','PENDING_ADDR_CHANGE') and live_permit_flag = 1 then 1 else 0 end as live_flag
 
 ,Case when live_permit_flag = 1 and blue_badge_number !='' and permit_type like 'Estate Resident' and (amount like '0.00' or amount like '0.000'  or amount like '') then 1 else 0 end as flag_lp_est_bb_zero
 ,Case when live_permit_flag = 1 and blue_badge_number !='' and permit_type != 'Estate Resident' then 1 else 0 end as flag_lp_bb_onst
@@ -220,7 +221,7 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
 
 , parking_permit_denormalised_data.*, llpg.planning_use_class, llpg.longitude, llpg.latitude  FROM parking_permit_denormalised_data
 
-left join llpg on cast(llpg.uprn as string) = cast(parking_permit_denormalised_data.uprn as string) /*and 
+left join llpg on cast(llpg.uprn as string) = cast(parking_permit_denormalised_data.uprn as string) /*and
 cast(concat(parking_permit_denormalised_data.import_year,parking_permit_denormalised_data.import_month,parking_permit_denormalised_data.import_day) as string
 ) = llpg.import_date*/
 
@@ -247,7 +248,9 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
 
 # Script generated for node Amazon S3
 AmazonS3_node1657709892303 = glueContext.getSink(
-    path="s3://dataplatform-"+environment+"-refined-zone/parking/liberator/parking_permit_denormalised_gds_street_llpg/",
+    path="s3://dataplatform-"
+    + environment
+    + "-refined-zone/parking/liberator/parking_permit_denormalised_gds_street_llpg/",
     connection_type="s3",
     updateBehavior="UPDATE_IN_DATABASE",
     partitionKeys=["import_year", "import_month", "import_day", "import_date"],
@@ -256,7 +259,7 @@ def sparkSqlQuery(glueContext, query, mapping, transformation_ctx) -> DynamicFra
     transformation_ctx="AmazonS3_node1657709892303",
 )
 AmazonS3_node1657709892303.setCatalogInfo(
-    catalogDatabase="dataplatform-"+environment+"-liberator-refined-zone",
+    catalogDatabase="dataplatform-" + environment + "-liberator-refined-zone",
     catalogTableName="parking_permit_denormalised_gds_street_llpg",
 )
 AmazonS3_node1657709892303.setFormat("glueparquet")