An ETL pipeline with Amazon Redshift and AWS Glue example (#498)

examples/aws-redshift-glue-etl/.gitignore

@@ -0,0 +1,10 @@
+### Scala an JVM
+*.class
+*.log
+.bsp
+.scala-build
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*
+
+kubeconfig.json

examples/aws-redshift-glue-etl/Main.scala

@@ -0,0 +1,224 @@
+import besom.*
+import besom.api.aws
+import besom.json.*
+
+@main def main = Pulumi.run {
+
+  val clusterIdentifier = "my-redshift-cluster"
+  val clusterDBName     = "dev"
+  val clusterDBUsername = "admin"
+  val clusterDBPassword = "Password!123"
+  val glueDBName        = "my-glue-db"
+
+  // Create an S3 bucket to store some raw data.
+  val eventsBucket = aws.s3.Bucket(
+    name = "events",
+    aws.s3.BucketArgs(forceDestroy = true)
+  )
+
+  // Create a VPC.
+  val vpc = aws.ec2.Vpc(
+    name = "vpc",
+    aws.ec2.VpcArgs(
+      cidrBlock = "10.0.0.0/16",
+      enableDnsHostnames = true
+    )
+  )
+
+  // Create a private subnet within the VPC.
+  val subnet = aws.ec2.Subnet(
+    name = "subnet",
+    aws.ec2.SubnetArgs(
+      vpcId = vpc.id,
+      cidrBlock = "10.0.1.0/24"
+    )
+  )
+
+  // Declare a Redshift subnet group with the subnet ID.
+  val subnetGroup = aws.redshift.SubnetGroup(
+    name = "subnet-group",
+    aws.redshift.SubnetGroupArgs(
+      subnetIds = List(subnet.id)
+    )
+  )
+
+  // Create an IAM role granting Redshift read-only access to S3.
+  val redshiftRole = aws.iam.Role(
+    name = "redshift-role",
+    aws.iam.RoleArgs(
+      assumeRolePolicy = json"""{
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Action": "sts:AssumeRole",
+                "Effect": "Allow",
+                "Principal": {
+                    "Service": "redshift.amazonaws.com"
+                }
+            }
+        ]
+    }""".map(_.prettyPrint),
+      managedPolicyArns = List(
+        aws.iam.enums.ManagedPolicy.AmazonS3ReadOnlyAccess.value
+      )
+    )
+  )
+
+  // Create a VPC endpoint so the cluster can read from S3 over the private network.
+  val vpcEndpoint = aws.ec2.VpcEndpoint(
+    name = "s3-vpc-endpoint",
+    aws.ec2.VpcEndpointArgs(
+      vpcId = vpc.id,
+      serviceName = p"com.amazonaws.${aws.getRegion(aws.GetRegionArgs()).name}.s3",
+      routeTableIds = List(vpc.mainRouteTableId)
+    )
+  )
+
+  // Create a single-node Redshift cluster in the VPC.
+  val cluster = aws.redshift.Cluster(
+    name = "cluster",
+    aws.redshift.ClusterArgs(
+      clusterIdentifier = clusterIdentifier,
+      databaseName = clusterDBName,
+      masterUsername = clusterDBUsername,
+      masterPassword = clusterDBPassword,
+      nodeType = "ra3.xlplus",
+      clusterSubnetGroupName = subnetGroup.name,
+      clusterType = "single-node",
+      publiclyAccessible = false,
+      skipFinalSnapshot = true,
+      vpcSecurityGroupIds = List(vpc.defaultSecurityGroupId),
+      iamRoles = List(redshiftRole.arn)
+    )
+  )
+
+  // Define an AWS cron expression of "every 15 minutes".
+  // https://docs.aws.amazon.com/lambda/latest/dg/services-cloudwatchevents-expressions.html
+  val every15minutes = "cron(0/15 * * * ? *)"
+
+  // Create a Glue catalog database.
+  val glueCatalogDB = aws.glue.CatalogDatabase(
+    name = "glue-catalog-db",
+    aws.glue.CatalogDatabaseArgs(
+      name = glueDBName
+    )
+  )
+
+// Define an IAM role granting AWS Glue access to S3 and other Glue-required services.
+  val glueRole = aws.iam.Role(
+    name = "glue-role",
+    aws.iam.RoleArgs(
+      assumeRolePolicy = json"""{
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Action": "sts:AssumeRole",
+                "Effect": "Allow",
+                "Principal": {
+                    "Service": "glue.amazonaws.com"
+                }
+            }
+        ]
+    }""".map(_.prettyPrint),
+      managedPolicyArns = List(
+        aws.iam.enums.ManagedPolicy.AmazonS3FullAccess.value,
+        aws.iam.enums.ManagedPolicy.AWSGlueServiceRole.value
+      )
+    )
+  )
+
+  // Create a Glue crawler to process the contents of the data bucket on a schedule.
+  // https://docs.aws.amazon.com/glue/latest/dg/monitor-data-warehouse-schedule.html
+  val glueCrawler = aws.glue.Crawler(
+    name = "glue-crawler",
+    aws.glue.CrawlerArgs(
+      databaseName = glueCatalogDB.name,
+      role = glueRole.arn,
+      schedule = every15minutes,
+      s3Targets = List(
+        aws.glue.inputs.CrawlerS3TargetArgs(
+          path = p"s3://${eventsBucket.bucket}"
+        )
+      )
+    )
+  )
+
+  // Create a Glue connection to the Redshift cluster.
+  val glueRedshiftConnection = aws.glue.Connection(
+    name = "glue-redshift-connection",
+    aws.glue.ConnectionArgs(
+      connectionType = "JDBC",
+      connectionProperties = Map(
+        "JDBC_CONNECTION_URL" -> p"jdbc:redshift://${cluster.endpoint}/${clusterDBName}",
+        "USERNAME" -> clusterDBUsername,
+        "PASSWORD" -> clusterDBPassword
+      ),
+      physicalConnectionRequirements = aws.glue.inputs.ConnectionPhysicalConnectionRequirementsArgs(
+        securityGroupIdLists = cluster.vpcSecurityGroupIds,
+        availabilityZone = subnet.availabilityZone,
+        subnetId = subnet.id
+      )
+    )
+  )
+
+  // Create an S3 bucket for Glue scripts and temporary storage.
+  val glueJobBucket = aws.s3.Bucket(
+    name = "glue-job-bucket",
+    aws.s3.BucketArgs(
+      forceDestroy = true
+    )
+  )
+
+  // Upload a Glue job script.
+  val glueJobScript = aws.s3.BucketObject(
+    name = "glue-job.py",
+    aws.s3.BucketObjectArgs(
+      bucket = glueJobBucket.id,
+      source = Asset.FileAsset("./glue-job.py")
+    )
+  )
+
+  // Create a Glue job that runs our Python ETL script.
+  val glueJob = aws.glue.Job(
+    name = "glue-job",
+    aws.glue.JobArgs(
+      roleArn = glueRole.arn,
+      glueVersion = "3.0",
+      numberOfWorkers = 10,
+      workerType = "G.1X",
+      defaultArguments = Map(
+        // Enabling job bookmarks helps you avoid loading duplicate data.
+        // https://docs.aws.amazon.com/glue/latest/dg/monitor-continuations.html
+        "--job-bookmark-option" -> "job-bookmark-enable",
+        "--ConnectionName" -> glueRedshiftConnection.name,
+        "--GlueDBName" -> glueDBName,
+        "--GlueDBTableName" -> eventsBucket.bucket.map(_.replace("-", "_")),
+        "--RedshiftDBName" -> clusterDBName,
+        "--RedshiftDBTableName" -> "events",
+        "--RedshiftRoleARN" -> redshiftRole.arn,
+        "--TempDir" -> p"s3://${glueJobBucket.bucket}/glue-job-temp"
+      ),
+      connections = List(glueRedshiftConnection.name),
+      command = aws.glue.inputs.JobCommandArgs(
+        scriptLocation = p"s3://${glueJobBucket.bucket}/glue-job.py",
+        pythonVersion = "3"
+      )
+    )
+  )
+
+  // Create a Glue trigger to run the job every 15 minutes.
+  val glueJobTrigger = aws.glue.Trigger(
+    name = "trigger",
+    aws.glue.TriggerArgs(
+      schedule = every15minutes,
+      `type` = "SCHEDULED",
+      actions = List(
+        aws.glue.inputs.TriggerActionArgs(jobName = glueJob.name)
+      )
+    )
+  )
+
+  Stack(vpcEndpoint, glueCrawler, glueJobScript, glueJobTrigger).exports(
+    dataBucketName = eventsBucket.bucket
+  )
+}

examples/aws-redshift-glue-etl/Pulumi.yaml

@@ -0,0 +1,3 @@
+name: aws-redshift-glue-etl
+description: An ETL pipeline with Amazon Redshift and AWS Glue
+runtime: scala

examples/aws-redshift-glue-etl/README.md

@@ -0,0 +1,45 @@
+# ETL pipeline with Amazon Redshift and AWS Glue
+
+This example creates an ETL pipeline using Amazon Redshift and AWS Glue. The pipeline extracts data from an S3 bucket
+with a Glue crawler, transforms it with a Python script wrapped in a Glue job, and loads it into a Redshift database
+deployed in a VPC.
+
+## Prerequisites
+
+[Follow the instructions](https://www.pulumi.com/docs/clouds/aws/get-started/begin/)
+to get started with Pulumi & AWS.
+
+## Deploying
+
+1. Create a new stack, which is an isolated deployment target for this example:
+
+   ```bash
+   pulumi stack init dev
+   ```
+
+2. Set the AWS region:
+
+   ```bash
+   pulumi config set aws:region us-west-2
+   ```
+
+3. Stand up the cluster:
+
+   ```bash
+   pulumi up
+   ```
+4. In a few moments, the Redshift cluster and Glue components will be up and running and the S3 bucket name emitted as a
+   Pulumi stack output:
+
+5. Upload the included sample data file to S3 to verify the automation works as expected:
+
+    ```bash
+    aws s3 cp events-1.txt s3://$(pulumi stack output dataBucketName)
+    ```
+
+6. When you're ready, destroy your stack and remove it:
+
+    ```bash
+    pulumi destroy --yes
+    pulumi stack rm --yes
+    ```

examples/aws-redshift-glue-etl/events-1.txt

@@ -0,0 +1,3 @@
+{"id": 1, "name": "An interesting event"}
+{"id": 2, "name": "Another interesting event"}
+{"id": 3, "name": "An event of monumental importance"}

examples/aws-redshift-glue-etl/glue-job.py

@@ -0,0 +1,61 @@
+import sys
+from awsglue.utils import getResolvedOptions
+from awsglue.transforms import ApplyMapping
+from awsglue.context import GlueContext
+from awsglue.job import Job
+from pyspark.context import SparkContext
+
+# Collect the arguments passed in by the glue.Job run.
+args = getResolvedOptions(
+    sys.argv,
+    [
+        "JOB_NAME",
+        "TempDir",
+        "ConnectionName",
+        "GlueDBName",
+        "GlueDBTableName",
+        "RedshiftRoleARN",
+        "RedshiftDBName",
+        "RedshiftDBTableName",
+    ],
+)
+
+glueContext = GlueContext(SparkContext.getOrCreate())
+
+job = Job(glueContext)
+job.init(args["JOB_NAME"], args)
+
+# Extract all unprocessed data from the Glue catalog.
+source0 = glueContext.create_dynamic_frame.from_catalog(
+    database=args["GlueDBName"],
+    table_name=args["GlueDBTableName"],
+    additional_options={
+        "jobBookmarkKeys": ["id"],
+        "jobBookmarkKeysSortOrder": "asc",
+    },
+    transformation_ctx="source0",
+)
+
+# Transform the data (mostly just to show how to do so).
+transformed0 = ApplyMapping.apply(
+    frame=source0,
+    mappings=[
+        ("id", "int", "event_id", "int"),
+        ("name", "string", "event_name", "string"),
+    ],
+)
+
+# Load the data into the Redshift database.
+glueContext.write_dynamic_frame.from_jdbc_conf(
+    frame=transformed0,
+    catalog_connection=args["ConnectionName"],
+    connection_options={
+        "database": args["RedshiftDBName"],
+        "dbtable": args["RedshiftDBTableName"],
+        "aws_iam_role": args["RedshiftRoleARN"],
+    },
+    redshift_tmp_dir=args["TempDir"],
+)
+
+# Call commit() to reset the job bookmark for the next run.
+job.commit()

examples/aws-redshift-glue-etl/project.scala

@@ -0,0 +1,6 @@
+//> using scala "3.3.1"
+//> using options -Werror -Wunused:all -Wvalue-discard -Wnonunit-statement
+//> using dep "org.virtuslab::besom-core:0.4.0-SNAPSHOT"
+//> using dep "org.virtuslab::besom-aws:6.32.0-core.0.4-SNAPSHOT"
+
+//> using repository sonatype:snapshots