Adding support for workflow cron expressions. Some changes to how coarse IAM permissions are granted to reduce LakeFormation location permisison redundancy.

Author: EC2 Default User

baseline_binaries/ojdbc8.jar

@@ -0,0 +1 @@
+YOU NEED TO DOWNLOAD ojdbc8.jar FROM https://www.oracle.com/database/technologies/appdev/jdbc-ucp-19c-downloads.html and replace this file with the file you downloaded. 

bin/aws.ts

@@ -50,6 +50,7 @@ const exisitingResourceImportStack = new cdk.Stack(app, 'resourceImportStack', {
 
 
 
+
 // Grant permissions:
 
 // const exampleExistingIamUser = iam.User.fromUserName(exisitingResourceImportStack, 'exampleUserGrantee', '--- EXISTING IAM USERNAME GOES HERE --' );
@@ -59,6 +60,14 @@ const exisitingResourceImportStack = new cdk.Stack(app, 'resourceImportStack', {
 // exampleS3DataSet.grantIamRead(exampleExistingIamRole);
 
 
+// chemblStack.grantIamRead(analyticsStack.NotebookRole);
+// openTargetsStack.grantIamRead(analyticsStack.NotebookRole);
+// bindingDBStack.grantIamRead(analyticsStack.NotebookRole);
+
+
+// const exampleUser = iam.User.fromUserName(coreDataLake, 'exampleGrantee', 'paul1' );
+
+
 // var exampleGrant: DataLakeEnrollment.TablePermissionGrant = {
 //     tables: ["association_data", "evidence_data","target_list","disease_list"],
 //     DatabasePermissions: [DataLakeEnrollment.DatabasePermission.Alter, DataLakeEnrollment.DatabasePermission.CreateTable, DataLakeEnrollment.DatabasePermission.Drop],

lib/constructs/data-lake-enrollment.ts

@@ -18,13 +18,16 @@ export class DataLakeEnrollment extends cdk.Construct {
     public CoarseAthenaAccessPolicy: iam.ManagedPolicy;
     private CoarseResourceAccessPolicy: iam.ManagedPolicy;
     private CoarseIamPolciesApplied: boolean;
+    private WorkflowCronScheduleExpression?: string;
+    
 
     constructor(scope: cdk.Construct, id: string, props: DataLakeEnrollment.DataLakeEnrollmentProps) {
         super(scope, id);
 
 
         this.DataSetName = props.DataSetName;
         this.CoarseIamPolciesApplied = false;
+        this.WorkflowCronScheduleExpression = props.WorkflowCronScheduleExpression;
 
     }
 
@@ -542,6 +545,7 @@ export namespace DataLakeEnrollment
     	GlueScriptPath: string;
     	GlueScriptArguments: any;
     	DataSetName: string;
+    	WorkflowCronScheduleExpression?: string;
     }
 
     export interface DatabasePermissionGrant {

lib/constructs/data-set-enrollment.ts

@@ -154,6 +154,7 @@ export interface DataSetEnrollmentProps extends cdk.StackProps {
 		GlueScriptArguments: any;
 		SourceAccessPolicy?: iam.Policy;
 		MaxDPUs: number;
+		WorkflowCronScheduleExpression?: string;
 }
 
 
@@ -258,7 +259,7 @@ export class DataSetEnrollment extends cdk.Construct {
 			}, 
 			name: `${props.dataSetName}_src_to_dl_etl`, 
 			timeout: 2880, 
-			glueVersion: "1.0", 
+			glueVersion: "2.0", 
 			maxCapacity: props.MaxDPUs,
 			command: {
 				scriptLocation: `s3://${glueScript.s3BucketName}/${glueScript.s3ObjectKey}`, 
@@ -292,12 +293,13 @@ export class DataSetEnrollment extends cdk.Construct {
 			workfowName: `${props.dataSetName}_DataLakeEnrollmentWorkflow`,
 			srcCrawler: sourceCrawler,
 			etlJob: etl_job,
-			datalakeCrawler: datalake_crawler
-			
+			datalakeCrawler: datalake_crawler,
+			WorkflowCronScheduleExpression: props.WorkflowCronScheduleExpression
 		});
 
-	}
+	} 
 
+
 
 }
 
@@ -307,32 +309,55 @@ export interface DataLakeEnrollmentWorkflowProps {
 	srcCrawler: glue.CfnCrawler,
 	etlJob: glue.CfnJob,
 	datalakeCrawler: glue.CfnCrawler
+	WorkflowCronScheduleExpression?: string;
 }
 
 export class DataLakeEnrollmentWorkflow extends cdk.Construct {
 
-	public readonly StartTrigger: glue.CfnTrigger;
+	public StartTrigger: glue.CfnTrigger;
     public readonly SrcCrawlerCompleteTrigger: glue.CfnTrigger;
     public readonly ETLCompleteTrigger: glue.CfnTrigger; 
     public readonly Workflow: glue.CfnWorkflow; 
+    private readonly sourceCrawler: glue.CfnCrawler;
 
 	constructor(scope: cdk.Construct, id: string, props: DataLakeEnrollmentWorkflowProps) {
 		super(scope, id);
 
 		this.Workflow = new glue.CfnWorkflow(this, "etlWorkflow", {
 			name: props.workfowName
 		});
+
+		this.sourceCrawler = props.srcCrawler;
 
-		this.StartTrigger = new glue.CfnTrigger(this,"startTrigger",{
-            actions: [
-                {
-                    crawlerName: props.srcCrawler.name
-                }
-            ], 
-            type: "ON_DEMAND", 
-            name: `startWorkflow-${this.Workflow.name}`, 
-            workflowName: this.Workflow.name
-        });
+		if(props.WorkflowCronScheduleExpression == null){
+			this.StartTrigger = new glue.CfnTrigger(this,"startTrigger",{
+	            actions: [
+	                {
+	                    crawlerName: props.srcCrawler.name
+	                }
+	            ], 
+	            type: "ON_DEMAND", 
+	            name: `startWorkflow-${this.Workflow.name}`, 
+	            workflowName: this.Workflow.name
+	        });
+			
+		}else{
+			
+			this.StartTrigger = new glue.CfnTrigger(this,"startTrigger",{
+	            actions: [
+	                {
+	                    crawlerName: this.sourceCrawler.name
+	                }
+	            ], 
+	            type: "SCHEDULED", 
+	            schedule: props.WorkflowCronScheduleExpression,
+	            name: `startWorkflow-${this.Workflow.name}`, 
+	            workflowName: this.Workflow.name
+	        });
+		}
+		
+		
+
 
 		this.SrcCrawlerCompleteTrigger = new glue.CfnTrigger(this,"srcCrawlerCompleteTrigger",{
 	        predicate: {

lib/constructs/s3-data-set-enrollment.ts

@@ -81,13 +81,21 @@ export class S3dataSetEnrollment extends DataLakeEnrollment{
             var tableFolderName = prefixFolders[prefixFolders.length-2]
             var tableFolderName = tableFolderName.toLowerCase().replace(/\./g,"_").replace(/-/g,"_");
 
-            s3DataLakePaths.push({
-                path: `s3://${props.dataLakeBucket.bucketName}/${dataSetName}/${tableFolderName}/`
-            });
+            if(props.sourceBucketDataPrefixes.length > 1){
+                s3DataLakePaths.push({
+                    path: `s3://${props.dataLakeBucket.bucketName}/${dataSetName}/${tableFolderName}/`
+                });                
+            }else{
+                s3DataLakePaths.push({
+                    path: `s3://${props.dataLakeBucket.bucketName}/${dataSetName}/`
+                });
+            }
+            
+
         }
+
 
-        
-		
+
 		this.DataEnrollment = new DataSetEnrollment(this, `${props.DataSetName}-s3Enrollment`, {
 		    dataLakeBucket: props.dataLakeBucket,
 			dataSetName: dataSetName,
@@ -100,7 +108,8 @@ export class S3dataSetEnrollment extends DataLakeEnrollment{
 			DataLakeTargets: {
 			    s3Targets: s3DataLakePaths
 			},
-			GlueScriptArguments: props.GlueScriptArguments
+			GlueScriptArguments: props.GlueScriptArguments,
+			WorkflowCronScheduleExpression: props.WorkflowCronScheduleExpression
 		});
 
 

scripts/glue.s3import.fullcopy.rds.py

@@ -45,11 +45,11 @@
 
 for table in tables:
 
-  datasource = glueContext.create_dynamic_frame.from_catalog(database = glue_database, table_name = table)   
-  dropnullfields = DropNullFields.apply(frame = datasource, transformation_ctx = "dropnullfields1")
+  datasource = glueContext.create_dynamic_frame.from_catalog(database = glue_database, table_name = table, transformation_ctx = "datasource")   
+  dropnullfields = DropNullFields.apply(frame = datasource, transformation_ctx = "dropnullfields")
 
   try:
-    datasink = glueContext.write_dynamic_frame.from_options(frame = dropnullfields, connection_type = "s3", connection_options = {"path": "s3://"+dataLakeBucket + dataLakePrefix + table}, format = target_format)
+    datasink = glueContext.write_dynamic_frame.from_options(frame = dropnullfields, connection_type = "s3", connection_options = {"path": "s3://"+dataLakeBucket + dataLakePrefix + table}, format = target_format, transformation_ctx = "datasink")
   except:
     print("Unable to write" + table)
 

scripts/glue.s3import.fullcopy.s3.py

@@ -45,11 +45,11 @@
 
 for table in tables:
 
-  datasource = glueContext.create_dynamic_frame.from_catalog(database = glue_database, table_name = table)   
-  dropnullfields = DropNullFields.apply(frame = datasource, transformation_ctx = "dropnullfields1")
+  datasource = glueContext.create_dynamic_frame.from_catalog(database = glue_database, table_name = table, transformation_ctx = "datasource")   
+  dropnullfields = DropNullFields.apply(frame = datasource, transformation_ctx = "dropnullfields")
 
   try:
-    datasink = glueContext.write_dynamic_frame.from_options(frame = dropnullfields, connection_type = "s3", connection_options = {"path": "s3://"+dataLakeBucket + dataLakePrefix + table}, format = target_format)
+    datasink = glueContext.write_dynamic_frame.from_options(frame = dropnullfields, connection_type = "s3", connection_options = {"path": "s3://"+dataLakeBucket + dataLakePrefix + table}, format = target_format, transformation_ctx = "datasink")
   except:
     print("Unable to write" + table)