AWS Cloud

This is a recording of breakout sessions from AWS Heroes at re:Invent 2022. Posted with permission. Both serverless and Kubernetes have benefits for your operational production environments, but how do you choose? In this video session, we will have a “battle” between the serverless and the Kubernetes approach examining use cases and insights from each speaker's experience. After an overview of each architecture and the AWS services that are a part of it like databases, queues, and more, we will compare: Maintenance and compliance Scaling Developer experience Cost Monitoring and logging Ecosystem For each category, we will show the advantages and disadvantages of each architecture side by side with the audience voting on who wins each round.

This is a recording of the breakout session led by AWS Hero Margaret Valtierra at AWS re:Invent 2022, Las Vegas. Posted with permission. Curious how, for mere dollars a month and minimal upkeep, you can centrally track and manage Outposts capacity across multiple AWS accounts? In this session, we’ll show a unique solution implemented at Morningstar by the Cloud Services team to do just that. We'll walk through how we arrived at the architecture of the solution that uses lambdas, DynamoDB, CloudWatch, S3, and a custom API to track capacity and block users from overspending their quota.

This is a recording of a breakout session from AWS Heroes at re:Invent 2022, presented by AWS Hero Zainab Maleki. Posted with permission. In software engineering, we've learned that building robust and stable applications has a direct correlation with overall organization performance. The data community is striving to incorporate the core concepts of engineering rigor found in software communities but still has further to go. This talk covers ways to leverage software engineering practices for data engineering and demonstrates how measuring key performance metrics could help build more robust and reliable data pipelines. This is achieved through practices like Infrastructure as Code for deployments, automated testing, application observability, and end-to-end application lifecycle ownership.

This is a recording of breakout sessions from AWS Heroes at re:Invent 2022. Posted with permission. A new category of developer tools is emerging that challenges the primitive-centric approach to building modern cloud applications. The complexity of configuring services, implementing best practices, and securing workloads have led to major problems with developer productivity, cost, performance, and time-to-market. In the current "Cloud Native" landscape, most organizations are left with unfillable skill gaps and failed cloud initiatives.

AWS Controllers for Kubernetes (also known as ACK) is built around the Kubernetes extension concepts of Custom Resource and Custom Resource Definitions. You can use ACK to define and use AWS services directly from Kubernetes. This helps you take advantage of managed AWS services for your Kubernetes applications without needing to define resources outside of the cluster. Say you need to use an AWS S3 Bucket in your application that’s deployed to Kubernetes. Instead of using AWS console, AWS CLI, AWS CloudFormation etc., you can define the AWS S3 Bucket in a YAML manifest file and deploy it using familiar tools such as kubectl. The end goal is to allow users (Software Engineers, DevOps engineers, operators etc.) to use the same interface (Kubernetes API in this case) to describe and manage AWS services along with native Kubernetes resources such as Deployment, Service etc. Here is a diagram from the ACK documentation, that provides a high-level overview: Working of ACK What’s Covered in This Blog Post? ACK supports many AWS services, including Amazon DynamoDB. One of the topics that this blog post covers is how to use ACK on Amazon EKS for managing DynamoDB. But, just creating a DynamoDB table isn't going to be all that interesting! In addition to it, you will also work with and deploy a client application — this is a trimmed-down version of the URL shortener app covered in a previous blog post. While the first half of the blog will involve manual steps to help you understand the mechanics and get started, in the second half, we will switch to cdk8s and achieve the same goals using nothing but Go code. Cdk8s? What, Why? Because Infrastructure Is Code cdk8s (Cloud Development Kit for Kubernetes) is an open-source framework (part of CNCF) that allows you to define your Kubernetes applications using regular programming languages (instead of yaml). I have written a few blog posts around cdk8s and Go, that you may find useful. We will continue on the same path i.e. push yaml to the background and use the Go programming language to define the core infrastructure (that happens to be DynamoDB in this example, but could be so much more) as well as the application components (Kubernetes Deployment, Service etc.). This is made possible due to the following cdk8s features: cdk8s support for Kubernetes Custom Resource definitions that lets us magically import CRD as APIs. a cdk8s-plus library that helps reduce/eliminate a lot of boilerplate code while working with Kubernetes resources in our Go code (or any other language for that matter) Before you dive in, please ensure you complete the prerequisites in order to work through the tutorial. The entire code for the infrastructure and the application is available on GitHub Pre-requisites To follow along step-by-step, in addition to an AWS account, you will need to have AWS CLI, cdk8s CLI, kubectl, helm and the Go programming language installed. There are a variety of ways in which you can create an Amazon EKS cluster. I prefer using eksctl CLI because of the convenience it offers! Ok, let’s get started. The first thing we need to do is… Set up the DynamoDB Controller Most of the below steps are adapted from the ACK documentation - Install an ACK Controller Install It Using Helm: Shell export SERVICE=dynamodb Change/update this as required as per the latest release document Shell export RELEASE_VERSION=0.1.7 Shell export ACK_SYSTEM_NAMESPACE=ack-system # you can change the region if required export AWS_REGION=us-east-1 aws ecr-public get-login-password --region us-east-1 | helm registry login --username AWS --password-stdin public.ecr.aws helm install --create-namespace -n $ACK_SYSTEM_NAMESPACE ack-$SERVICE-controller \ oci://public.ecr.aws/aws-controllers-k8s/$SERVICE-chart --version=$RELEASE_VERSION --set=aws.region=$AWS_REGION To confirm, run: Shell kubectl get crd Shell # output (multiple CRDs) tables.dynamodb.services.k8s.aws fieldexports.services.k8s.aws globaltables.dynamodb.services.k8s.aws # etc.... Since the DynamoDB controller has to interact with AWS Services (make API calls), and we need to configure IAM Roles for Service Accounts (also known as IRSA). Refer to Configure IAM Permissions for details IRSA Configuration First, create an OIDC identity provider for your cluster. Shell export EKS_CLUSTER_NAME=<name of your EKS cluster> export AWS_REGION=<cluster region> eksctl utils associate-iam-oidc-provider --cluster $EKS_CLUSTER_NAME --region $AWS_REGION --approve The goal is to create an IAM role and attach appropriate permissions via policies. We can then create a Kubernetes Service Account and attach the IAM role to it. Thus, the controller Pod will be able to make AWS API calls. Note that we are using providing all DynamoDB permissions to our control via the arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess policy. Thanks to eksctl, this can be done with a single line! Shell export SERVICE=dynamodb export ACK_K8S_SERVICE_ACCOUNT_NAME=ack-$SERVICE-controller Shell # recommend using the same name export ACK_SYSTEM_NAMESPACE=ack-system export EKS_CLUSTER_NAME=<enter EKS cluster name> export POLICY_ARN=arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess Shell # IAM role has a format - do not change it. you can't use any arbitrary name export IAM_ROLE_NAME=ack-$SERVICE-controller-role Shell eksctl create iamserviceaccount \ --name $ACK_K8S_SERVICE_ACCOUNT_NAME \ --namespace $ACK_SYSTEM_NAMESPACE \ --cluster $EKS_CLUSTER_NAME \ --role-name $IAM_ROLE_NAME \ --attach-policy-arn $POLICY_ARN \ --approve \ --override-existing-serviceaccounts The policy is per recommended policy To confirm, you can check whether the IAM role was created and also introspect the Kubernetes service account Shell aws iam get-role --role-name=$IAM_ROLE_NAME --query Role.Arn --output text Shell kubectl describe serviceaccount/$ACK_K8S_SERVICE_ACCOUNT_NAME -n $ACK_SYSTEM_NAMESPACE You will see similar output: Shell Name: ack-dynamodb-controller Namespace: ack-system Labels: app.kubernetes.io/instance=ack-dynamodb-controller app.kubernetes.io/managed-by=eksctl app.kubernetes.io/name=dynamodb-chart app.kubernetes.io/version=v0.1.3 helm.sh/chart=dynamodb-chart-v0.1.3 k8s-app=dynamodb-chart Annotations: eks.amazonaws.com/role-arn: arn:aws:iam::<your AWS account ID>:role/ack-dynamodb-controller-role meta.helm.sh/release-name: ack-dynamodb-controller meta.helm.sh/release-namespace: ack-system Image pull secrets: <none> Mountable secrets: ack-dynamodb-controller-token-bbzxv Tokens: ack-dynamodb-controller-token-bbzxv Events: <none> For IRSA to take effect, you need to restart the ACK Deployment: Shell # Note the deployment name for ACK service controller from following command kubectl get deployments -n $ACK_SYSTEM_NAMESPACE Shell kubectl -n $ACK_SYSTEM_NAMESPACE rollout restart deployment ack-dynamodb-controller-dynamodb-chart Confirm that the Deployment has restarted (currently Running) and the IRSA is properly configured: Shell kubectl get pods -n $ACK_SYSTEM_NAMESPACE Shell kubectl describe pod -n $ACK_SYSTEM_NAMESPACE ack-dynamodb-controller-dynamodb-chart-7dc99456c6-6shrm | grep "^\s*AWS_" # The output should contain following two lines: Shell AWS_ROLE_ARN=arn:aws:iam::<AWS_ACCOUNT_ID>:role/<IAM_ROLE_NAME> AWS_WEB_IDENTITY_TOKEN_FILE=/var/run/secrets/eks.amazonaws.com/serviceaccount/token Now that we’re done with the configuration … We Can Move On to the Fun Parts! Start by Creating the DynamoDB Table Here is what the manifest looks like: YAML apiVersion: dynamodb.services.k8s.aws/v1alpha1 kind: Table metadata: name: dynamodb-urls-ack spec: tableName: urls attributeDefinitions: - attributeName: shortcode attributeType: S billingMode: PAY_PER_REQUEST keySchema: - attributeName: email keyType: HASH Clone the project, change to the correct directory and apply the manifest: Shell git clone https://github.com/abhirockzz/dynamodb-ack-cdk8s cd dynamodb-ack-cdk8s Shell # create table kubectl apply -f manual/dynamodb-ack.yaml You can check the DynamoDB table in the AWS console, or use the AWS CLI (aws dynamodb list-tables) to confirm.Our table is ready, now we can deploy our URL shortener application. But, before that, we need to create a Docker image and push it to a private repository in Amazon Elastic Container Registry (ECR). Create a Private Repository in Amazon ECR Login to ECR: Shell aws ecr get-login-password --region <enter region> | docker login --username AWS --password-stdin <enter aws_account_id>.dkr.ecr.<enter region>.amazonaws.com Create Repository: Shell aws ecr create-repository \ --repository-name dynamodb-app \ --region <enter AWS region> Build the Image and Push It to ECR Shell # if you're on Mac M1 #export DOCKER_DEFAULT_PLATFORM=linux/amd64 docker build -t dynamodb-app . Shell docker tag dynamodb-app:latest <enter aws_account_id>.dkr.ecr.<enter region>.amazonaws.com/dynamodb-app:latest Shell docker push <enter aws_account_id>.dkr.ecr.<enter region>.amazonaws.com/dynamodb-app:latest Just like the controller, our application also needs IRSA to be able to execute GetItem and PutItem API calls on DynamoDB. Let’s Create Another IRSA for the Application Shell # you can change the policy name. make sure yo use the same name in subsequent commands aws iam create-policy --policy-name dynamodb-irsa-policy --policy-document file://manual/policy.json Shell eksctl create iamserviceaccount --name eks-dynamodb-app-sa --namespace default --cluster <enter EKS cluster name> --attach-policy-arn arn:aws:iam::<enter AWS account ID>:policy/dynamodb-irsa-policy --approve Shell kubectl describe serviceaccount/eks-dynamodb-app-sa Output: Shell Name: eks-dynamodb-app-sa Namespace: default Labels: app.kubernetes.io/managed-by=eksctl Annotations: eks.amazonaws.com/role-arn: arn:aws:iam::<AWS account ID>:role/eksctl-eks-cluster-addon-iamserviceaccount-d-Role1-2KTGZO1GJRN Image pull secrets: <none> Mountable secrets: eks-dynamodb-app-sa-token-5fcvf Tokens: eks-dynamodb-app-sa-token-5fcvf Events: <none> Finally, we can deploy our application! In the manual/app.yaml file, make sure you replace the following attributes as per your environment: Service account name — the above example used eks-dynamodb-app-sa Docker image Container environment variables for AWS region (e.g. us-east-1) and table name (this will be urls since that's the name we used) Shell kubectl apply -f app.yaml Shell # output deployment.apps/dynamodb-app configured service/dynamodb-app-service created This will create a Deployment as well as Service to access the application. Since the Service type is LoadBalancer, an appropriate AWS Load Balancer will be provisioned to allow for external access. Check Pod and Service: Shell kubectl get pods kubectl get service/dynamodb-app-service Shell # to get the load balancer IP APP_URL=$(kubectl get service/dynamodb-app-service -o jsonpath="{.status.loadBalancer.ingress[0].hostname}") echo $APP_URL Shell # output example a0042d5b5b0ad40abba9c6c42e6342a2-879424587.us-east-1.elb.amazonaws.com You have deployed the application and know the endpoint over which it’s publicly accessible. You can try out the URL shortener service Shell curl -i -X POST -d 'https://abhirockzz.github.io/' $APP_URL:9090/ Shell # output HTTP/1.1 200 OK Date: Thu, 21 Jul 2022 11:03:40 GMT Content-Length: 25 Content-Type: text/plain; charset=utf-8 JSON {"ShortCode":"ae1e31a6"} If you get a Could not resolve host error while accessing the LB URL, wait for a minute or so and re-try You should receive a JSON response with a short code. Enter the following in your browser http://<enter APP_URL>:9090/<shortcode> e.g. http://a0042d5b5b0ad40abba9c6c42e6342a2-879424587.us-east-1.elb.amazonaws.com:9090/ae1e31a6 - you will be redirected to the original URL. You can also use curl: Shell # example curl -i http://a0042d5b5b0ad40abba9c6c42e6342a2-879424587.us-east-1.elb.amazonaws.com:9090/ae1e31a6 # output HTTP/1.1 302 Found Location: https://abhirockzz.github.io/ Date: Thu, 21 Jul 2022 11:07:58 GMT Content-Length: 0 Enough of YAML I guess! As I promised earlier, the second half will demonstrate how to achieve the same using cdk8s and Go. Kubernetes Infrastructure as Go Code With Cdk8s Assuming you’ve already cloned the project (as per the above instructions), change to a different directory: Shell cd dynamodb-cdk8s This is a pre-created cdk8s project that you can use. The entire logic is present in main.go file. We will first try it out and confirm that it works the same way. After that, we will dive into the nitty-gritty of the code. Delete the previously created DynamoDB table and along with the Deployment (and Service) as well: Shell # you can also delete the table directly from AWS console aws dynamodb delete-table --table-name urls # this will delete Deployment and Service (as well as AWS Load Balancer) kubectl delete -f manual/app.yaml Use cdk8s synth to generate the manifest for the DynamoDB table and the application. We can then apply it using kubectl See the difference? Earlier, we defined the DynamoDB table, Deployment (and Service) manifests manually. cdk8s does not remove YAML altogether, but it provides a way for us to leverage regular programming languages (Go in this case) to define the components of our solution. Shell export TABLE_NAME=urls export SERVICE_ACCOUNT=eks-dynamodb-app-sa export DOCKER_IMAGE=<enter ECR repo that you created earlier> cdk8s synth ls -lrt dist/ #output 0000-dynamodb.k8s.yaml 0001-deployment.k8s.yaml You will see two different manifests being generated by cdk8s since we defined two separate cdk8s.Charts in the code - more on this soon. We can deploy them one by one: Shell kubectl apply -f dist/0000-dynamodb.k8s.yaml #output table.dynamodb.services.k8s.aws/dynamodb-dynamodb-ack-cdk8s-table-c88d874d created configmap/export-dynamodb-urls-info created fieldexport.services.k8s.aws/export-dynamodb-tablename created fieldexport.services.k8s.aws/export-dynamodb-region created As always, you can check the DynamoDB table either in the console or AWS CLI - aws dynamodb describe-table --table-name urls. Looking at the output, the DynamoDB table part seems familiar... But what’s fieldexport.services.k8s.aws?? … And why do we need a ConfigMap? I will give you the gist here. In the previous iteration, we hard-coded the table name and region in manual/app.yaml. While this works for this sample app, it is not scalable and might not even work for a few resources in case the metadata (like name etc.) is randomly generated. That's why there is this concept of a FieldExport that can "export any spec or status field from an ACK resource into a Kubernetes ConfigMap or Secret" You can read up on the details in the ACK documentation along with some examples. What you will see here is how to define a FieldExport and ConfigMap along with the Deployment which needs to be configured to accept its environment variables from the ConfigMap - all this in code, using Go (more on this during code walk-through). Check the FieldExport and ConfigMap: Shell kubectl get fieldexport #output NAME AGE export-dynamodb-region 19s export-dynamodb-tablename 19s kubectl get configmap/export-dynamodb-urls-info -o yaml We started out with a blank ConfigMap (as per cdk8s logic), but ACK magically populated it with the attributes from the Table custom resource. YAML apiVersion: v1 data: default.export-dynamodb-region: us-east-1 default.export-dynamodb-tablename: urls immutable: false kind: ConfigMap #....omitted We can now use the second manifest — no surprises here. Just like in the previous iteration, all it contains is the application Deployment and the Service. Check Pod and Service: Shell kubectl apply -f dist/0001-deployment.k8s.yaml #output deployment.apps/dynamodb-app created service/dynamodb-app-service configured kubectl get pods kubectl get svc The entire setup is ready, just like it was earlier and you can test it the same way. I will not repeat the steps here. Instead, I will move to something more interesting. Cdk8s Code Walk-Through The logic is divided into two Charts. I will only focus on key sections of the code and the rest will be omitted for brevity. DynamoDB and Configuration We start by defining the DynamoDB table (name it urls) as well as the ConfigMap (note that it does not have any data at this point): Go func NewDynamoDBChart(scope constructs.Construct, id string, props *MyChartProps) cdk8s.Chart { //... table := ddbcrd.NewTable(chart, jsii.String("dynamodb-ack-cdk8s-table"), &ddbcrd.TableProps{ Spec: &ddbcrd.TableSpec{ AttributeDefinitions: &[]*ddbcrd.TableSpecAttributeDefinitions{ {AttributeName: jsii.String(primaryKeyName), AttributeType: jsii.String("S")}, BillingMode: jsii.String(billingMode), TableName: jsii.String(tableName), KeySchema: &[]*ddbcrd.TableSpecKeySchema{ {AttributeName: jsii.String(primaryKeyName), KeyType: jsii.String(hashKeyType)}}) //... cfgMap = cdk8splus22.NewConfigMap(chart, jsii.String("config-map"), &cdk8splus22.ConfigMapProps{ Metadata: &cdk8s.ApiObjectMetadata{ Name: jsii.String(configMapName)}) Then we move on to the FieldExports - one each for the AWS region and the table name. As soon as these are created, the ConfigMap is populated with the required data as per from and to configuration in the FieldExport. Go //... fieldExportForTable = servicesk8saws.NewFieldExport(chart, jsii.String("fexp-table"), &servicesk8saws.FieldExportProps{ Metadata: &cdk8s.ApiObjectMetadata{Name: jsii.String(fieldExportNameForTable)}, Spec: &servicesk8saws.FieldExportSpec{ From: &servicesk8saws.FieldExportSpecFrom{Path: jsii.String(".spec.tableName"), Resource: &servicesk8saws.FieldExportSpecFromResource{ Group: jsii.String("dynamodb.services.k8s.aws"), Kind: jsii.String("Table"), Name: table.Name()}, To: &servicesk8saws.FieldExportSpecTo{ Name: cfgMap.Name(), Kind: servicesk8saws.FieldExportSpecToKind_CONFIGMAP}}) Go fieldExportForRegion = servicesk8saws.NewFieldExport(chart, jsii.String("fexp-region"), &servicesk8saws.FieldExportProps{ Metadata: &cdk8s.ApiObjectMetadata{Name: jsii.String(fieldExportNameForRegion)}, Spec: &servicesk8saws.FieldExportSpec{ From: &servicesk8saws.FieldExportSpecFrom{ Path: jsii.String(".status.ackResourceMetadata.region"), Resource: &servicesk8saws.FieldExportSpecFromResource{ Group: jsii.String("dynamodb.services.k8s.aws"), Kind: jsii.String("Table"), Name: table.Name()}, To: &servicesk8saws.FieldExportSpecTo{ Name: cfgMap.Name(), Kind: servicesk8saws.FieldExportSpecToKind_CONFIGMAP}}) //... The Application Chart The core of our application is Deployment itself: Go func NewDeploymentChart(scope constructs.Construct, id string, props *MyChartProps) cdk8s.Chart { //... dep := cdk8splus22.NewDeployment(chart, jsii.String("dynamodb-app-deployment"), &cdk8splus22.DeploymentProps{ Metadata: &cdk8s.ApiObjectMetadata{ Name: jsii.String("dynamodb-app")}, ServiceAccount: cdk8splus22.ServiceAccount_FromServiceAccountName( chart, jsii.String("aws-irsa"), jsii.String(serviceAccountName))}) The next important part is the container and its configuration. We specify the ECR image repository along with the environment variables — they reference the ConfigMap we defined in the previous chart (everything is connected!): Go //... container := dep.AddContainer( &cdk8splus22.ContainerProps{ Name: jsii.String("dynamodb-app-container"), Image: jsii.String(image), Port: jsii.Number(appPort)}) Go container.Env().AddVariable(jsii.String("TABLE_NAME"), cdk8splus22.EnvValue_FromConfigMap( cfgMap, jsii.String("default."+*fieldExportForTable.Name()), &cdk8splus22.EnvValueFromConfigMapOptions{Optional: jsii.Bool(false)})) Go container.Env().AddVariable(jsii.String("AWS_REGION"), cdk8splus22.EnvValue_FromConfigMap( cfgMap, jsii.String("default."+*fieldExportForRegion.Name()), &cdk8splus22.EnvValueFromConfigMapOptions{Optional: jsii.Bool(false)})) Finally, we define the Service (type LoadBalancer) which enables external application access and ties it all together in the main function: Go //... dep.ExposeViaService( &cdk8splus22.DeploymentExposeViaServiceOptions{ Name: jsii.String("dynamodb-app-service"), ServiceType: cdk8splus22.ServiceType_LOAD_BALANCER, Ports: &[]*cdk8splus22.ServicePort{ {Protocol: cdk8splus22.Protocol_TCP, Port: jsii.Number(lbPort), TargetPort: jsii.Number(appPort)}}) //... Go func main() { app := cdk8s.NewApp(nil) dynamodDB := NewDynamoDBChart(app, "dynamodb", nil) deployment := NewDeploymentChart(app, "deployment", nil) deployment.AddDependency(dynamodDB) app.Synth() } That’s all I have for you in this blog! Don’t Forget To Delete Resources... Shell # delete DynamoDB table, Deployment, Service etc. kubectl delete -f dist/ # to uninstall the ACK controller export SERVICE=dynamodb helm uninstall -n $ACK_SYSTEM_NAMESPACE ack-$SERVICE-controller # delete the EKS cluster. if created via eksctl: eksctl delete cluster --name <enter name of eks cluster> Wrap Up... AWS Controllers for Kubernetes help bridge the gap between traditional Kubernetes resources and AWS services by allowing you to manage both from a single control plane. In this blog, you saw how to do this in the context of DynamoDB and a URL shortener application (deployed to Kubernetes). I encourage you to try out other AWS services that ACK supports - here is a complete list. The approach presented here will work well if just want to use cdk8s. However, depending on your requirements, there is another way this can do by bringing AWS CDK into the picture. I want to pause right here since this is something I might cover in a future blog post. Until then, Happy Building!

A previous post covered how to deploy a Go Lambda function and trigger it in response to events sent to a topic in an MSK Serverless cluster. This blog will take it a notch further. The solution consists of an MSK Serverless cluster, a producer application on AWS App Runner, and a consumer application in Kubernetes (EKS) persisting data to DynamoDB. The core components (MSK cluster, EKS, and DynamoDB) and the producer application will be provisioned using Infrastructure-as-code with AWS CDK. Since the consumer application on EKS will interact with both MSK and DynamoDB, you will also need to configure appropriate IAM roles. All the components in this solution have been written in Go. The MSK producer and consumer app use the franz-go library (it also supports MSK IAM authentication). The CDK stacks have been written using CDK Go library. Prerequisites You will need the following: An AWS account Install AWS CDK, AWS CLI, Docker, eksctl and curl. Use CDK to Provision MSK, EKS, and DynamoDB AWS CDK is a framework for defining cloud infrastructure in code and provisioning it through AWS CloudFormation. The AWS CDK lets you build reliable, scalable, cost-effective applications in the cloud with the considerable expressive power of a programming language. All the code and config are present in this GitHub repo. Clone the GitHub repo and change it to the right directory: git clone https://github.com/abhirockzz/msk-cdk-apprunner-eks-dynamodb cd msk-cdk-apprunner-eks-dynamodb/cdk Deploy the first CDK stack: cdk deploy MSKDynamoDBEKSInfraStack Wait for all the components to get provisioned, including the MSK Serverless cluster, EKS cluster and DynamoDB. You can check its progress in the AWS CloudFormation console. You can take a look at the CDK stack code here. Deploy MSK Producer Application to App Runner Using CDK Deploy the second CDK stack. Note that in addition to deploying the producer application to App Runner, it also builds and uploads the consumer application Docker image to an ECR repository. Make sure to enter the MSK Serverless broker endpoint URL. export MSK_BROKER=<enter endpoint> export MSK_TOPIC=test-topic cdk deploy AppRunnerServiceStack Wait for the producer application to get deployed to App Runner. You can check its progress in the AWS CloudFormation console. You can take a look at the CDK stack code and the producer application. Once complete, make a note of the App Runner application public endpoint as well as the ECR repository for the consumer application. You should see these in the stack output as such: Outputs: AppRunnerServiceStack.AppURL = <app URL> AppRunnerServiceStack.ConsumerAppDockerImage = <ecr docker image> .... Now, you can verify if the application is functioning properly. Get the publicly accessible URL for the App Runner application and invoke it using curl. This will create the MSK topic and send data specified in the HTTP POST body. export APP_RUNNER_URL=<enter app runner URL> curl -i -X POST -d '{"email":"user1@foo.com","name":"user1"}' $APP_RUNNER_URL Now you can deploy the consumer application to the EKS cluster. Before that, execute the steps to configure appropriate permissions for the application to interact with MSK and DynamoDB. Configure IRSA for Consumer Application Applications in a pod's containers can use an AWS SDK or the AWS CLI to make API requests to AWS services using AWS Identity and Access Management (IAM) permissions. Applications must sign their AWS API requests with AWS credentials. IAM roles for service accounts provide the ability to manage credentials for your applications, similar to the way that Amazon EC2 instance profiles provide credentials to Amazon EC2 instances. Instead of creating and distributing your AWS credentials to the containers or using the Amazon EC2 instance's role, you associate an IAM role with a Kubernetes service account and configure your pods to use the service account. Exit the cdk directory and change to the root of the project: cd .. Create an IAM OIDC Identity Provider for Your Cluster With eksctl export EKS_CLUSTER_NAME=<EKS cluster name> oidc_id=$(aws eks describe-cluster --name $EKS_CLUSTER_NAME --query "cluster.identity.oidc.issuer" --output text | cut -d '/' -f 5) aws iam list-open-id-connect-providers | grep $oidc_id eksctl utils associate-iam-oidc-provider --cluster $EKS_CLUSTER_NAME --approve Define IAM Roles for the Application Configure IAM Roles for Service Accounts (also known as IRSA). Refer to the documentation for details. Start by creating a Kubernetes Service Account: kubectl apply -f - <<EOF apiVersion: v1 kind: ServiceAccount metadata: name: eks-app-sa EOF To verify: kubectl get serviceaccount/eks-app-sa -o yaml Set your AWS Account ID and OIDC Identity provider as environment variables: ACCOUNT_ID=$(aws sts get-caller-identity --query "Account" --output text) export AWS_REGION=<enter region e.g. us-east-1> OIDC_PROVIDER=$(aws eks describe-cluster --name $EKS_CLUSTER_NAME --query "cluster.identity.oidc.issuer" --output text | sed -e "s/^https:\/\///") Create a JSON file with Trusted Entities for the role: read -r -d '' TRUST_RELATIONSHIP <<EOF { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Federated": "arn:aws:iam::${ACCOUNT_ID}:oidc-provider/${OIDC_PROVIDER}" }, "Action": "sts:AssumeRoleWithWebIdentity", "Condition": { "StringEquals": { "${OIDC_PROVIDER}:aud": "sts.amazonaws.com", "${OIDC_PROVIDER}:sub": "system:serviceaccount:default:eks-app-sa" } } } ] } EOF echo "${TRUST_RELATIONSHIP}" > trust.json To verify: cat trust.json Now, create the IAM role: export ROLE_NAME=msk-consumer-app-irsa aws iam create-role --role-name $ROLE_NAME --assume-role-policy-document file://trust.json --description "IRSA for MSK consumer app on EKS" You will need to create and attach the policy to the role, since we only want the consumer application to consume data from the MSK cluster and put data to DynamoDB table. This needs to be fine-grained. In the policy.json file, replace values for MSK cluster and DynamoDB. Create and attach the policy to the role you just created: aws iam create-policy --policy-name msk-consumer-app-policy --policy-document file://policy.json aws iam attach-role-policy --role-name $ROLE_NAME --policy-arn=arn:aws:iam::$ACCOUNT_ID:policy/msk-consumer-app-policy Finally, associate the IAM role with the Kubernetes Service Account that you created earlier: kubectl annotate serviceaccount -n default eks-app-sa eks.amazonaws.com/role-arn=arn:aws:iam::$ACCOUNT_ID:role/$ROLE_NAME #confirm kubectl get serviceaccount/eks-app-sa -o yaml Deploy MSK Consumer Application to EKS You can refer to the consumer application code here. Make sure to update the consumer application manifest (app-iam.yaml) with the MSK cluster endpoint and ECR image (obtained from the stack output). kubectl apply -f msk-consumer/app-iam.yaml # verify Pods kubectl get pods -l=app=msk-iam-consumer-app Verify End-To-End Solution Continue to send records using the App Runner producer application: export APP_RUNNER_URL=<enter app runner URL> curl -i -X POST -d '{"email":"user2@foo.com","name":"user2"}' $APP_RUNNER_URL curl -i -X POST -d '{"email":"user3@foo.com","name":"user3"}' $APP_RUNNER_URL curl -i -X POST -d '{"email":"user4@foo.com","name":"user4"}' $APP_RUNNER_URL Check consumer app logs on EKS to verify: kubectl logs -f $(kubectl get pods -l=app=msk-iam-consumer-app -o jsonpath='{.items[0].metadata.name}') Scale-Out Consumer App The MSK topic created by the producer application has three topic partitions, so we can have a maximum of three consumer instances. Scale-out to three replicas: kubectl scale deployment/msk-iam-consumer-app --replicas=3 Verify the number of Pods and check logs for each of them. Notice how the data consumption is balanced across the three instances. kubectl get pods -l=app=msk-iam-consumer-app Conclusion You were able to deploy the end-to-end application using CDK. This comprised of a producer on App Runner sending data to MSK Serverless cluster and a consumer on EKS persisting data to DynamoDB. All the components were written using the Go programming language!

In this post, you will learn how to deploy a Go Lambda function and trigger it in response to events sent to a topic in an MSK Serverless cluster. The following topics have been covered: How to use the franz-go Go Kafka client to connect to MSK Serverless using IAM authentication Write a Go Lambda function to process data in MSK topic. Create the infrastructure: VPC, subnets, MSK cluster, Cloud9 etc. Configure Lambda and Cloud9 to access MSK using IAM roles and fine-grained permissions. MSK Serverless is a cluster type for Amazon MSK that makes it possible for you to run Apache Kafka without having to manage and scale cluster capacity. It automatically provisions and scales capacity while managing the partitions in your topic, so you can stream data without thinking about right-sizing or scaling clusters. Consider using a serverless cluster if your applications need on-demand streaming capacity that scales up and down automatically.- MSK Serverless Developer Guide Prerequisites You will need an AWS account to install AWS CLI, as well as a recent version of Go (1.18 or above). Clone this GitHub repository and change it to the right directory: git clone https://github.com/abhirockzz/lambda-msk-serverless-trigger-golang cd lambda-msk-serverless-trigger-golang Infrastructure Setup AWS CloudFormation is a service that helps you model and set up your AWS resources so that you can spend less time managing those resources and more time focusing on your applications that run in AWS. You create a template that describes all the AWS resources that you want (like Amazon EC2 instances or Amazon RDS DB instances), and CloudFormation takes care of provisioning and configuring those resources for you. You don't need to individually create and configure AWS resources and figure out what's dependent on what; CloudFormation handles that.- AWS CloudFormation User Guide Create VPC and Other Resources Use a CloudFormation template for this. aws cloudformation create-stack --stack-name msk-vpc-stack --template-body file://template.yaml Wait for the stack creation to complete before proceeding to other steps. Create MSK Serverless Cluster Use AWS Console to create the cluster. Configure the VPC and private subnets created in the previous step. Create an AWS Cloud9 Instance Make sure it is in the same VPC as the MSK Serverless cluster and choose the public subnet that you created earlier. Configure MSK Cluster Security Group After the Cloud9 instance is created, edit the MSK cluster security group to allow access from the Cloud9 instance. Configure Cloud9 To Send Data to MSK Serverless Cluster The code that we run from Cloud9 is going to produce data to the MSK Serverless cluster. So we need to ensure that it has the right privileges. For this, we need to create an IAM role and attach the required permissions policy. aws iam create-role --role-name Cloud9MSKRole --assume-role-policy-document file://ec2-trust-policy.json Before creating the policy, update the msk-producer-policy.json file to reflect the required details including MSK cluster ARN etc. aws iam put-role-policy --role-name Cloud9MSKRole --policy-name MSKProducerPolicy --policy-document file://msk-producer-policy.json Attach the IAM role to the Cloud9 EC2 instance: Send Data to MSK Serverless Using Producer Application Log into the Cloud9 instance and run the producer application (it is a Docker image) from a terminal. export MSK_BROKER=<enter the MSK Serverless endpoint> export MSK_TOPIC=test-topic docker run -p 8080:8080 -e MSK_BROKER=$MSK_BROKER -e MSK_TOPIC=$MSK_TOPIC public.ecr.aws/l0r2y6t0/msk-producer-app The application exposes a REST API endpoint using which you can send data to MSK. curl -i -X POST -d 'test event 1' http://localhost:8080 This will create the specified topic (since it was missing, to begin with) and also send the data to MSK. Now that the cluster and producer applications are ready, we can move on to the consumer. Instead of creating a traditional consumer, we will deploy a Lambda function that will be automatically invoked in response to data being sent to the topic in MSK. Configure and Deploy the Lambda Function Create Lambda Execution IAM Role and Attach the Policy A Lambda function's execution role is an AWS Identity and Access Management (IAM) role that grants the function permission to access AWS services and resources. When you invoke your function, Lambda automatically provides your function with temporary credentials by assuming this role. You don't have to call sts:AssumeRole in your function code. aws iam create-role --role-name LambdaMSKRole --assume-role-policy-document file://lambda-trust-policy.json aws iam attach-role-policy --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaMSKExecutionRole --role-name LambdaMSKRole Before creating the policy, update the msk-consumer-policy.json file to reflect the required details including MSK cluster ARN etc. aws iam put-role-policy --role-name LambdaMSKRole --policy-name MSKConsumerPolicy --policy-document file://msk-consumer-policy.json Build and Deploy the Go Function and Create a Zip File Build and zip the function code: GOOS=linux go build -o app zip func.zip app Deploy to Lambda: export LAMBDA_ROLE_ARN=<enter the ARN of the LambdaMSKRole created above e.g. arn:aws:iam::<your AWS account ID>:role/LambdaMSKRole> aws lambda create-function \ --function-name msk-consumer-function \ --runtime go1.x \ --zip-file fileb://func.zip \ --handler app \ --role $LAMBDA_ROLE_ARN Lambda VPC Configuration Make sure you choose the same VPC and private subnets as the MSK cluster. Also, select the same security group ID as MSK (for convenience). If you select a different one, make sure to update the MSK security group to add an inbound rule (for port 9098), just like you did for the Cloud9 instance in an earlier step. Configure the MSK Trigger for the Function When Amazon MSK is used as an event source, Lambda internally polls for new messages from the event source and then synchronously invokes the target Lambda function. Lambda reads the messages in batches and provides these to your function as an event payload. The maximum batch size is configurable (the default is 100 messages). Lambda reads the messages sequentially for each partition. After Lambda processes each batch, it commits the offsets of the messages in that batch. If your function returns an error for any of the messages in a batch, Lambda retries the whole batch of messages until processing succeeds or the messages expire. Lambda sends the batch of messages in the event parameter when it invokes your function. The event payload contains an array of messages. Each array item contains details of the Amazon MSK topic and partition identifier, together with a timestamp and a base64-encoded message. Make sure to choose the right MSK Serverless cluster and enter the correct topic name. Verify the Integration Go back to the Cloud9 terminal and send more data using the producer application. I used a handy JSON utility called jo (sudo yum install jo). APP_URL=http://localhost:8080 for i in {1..5}; do jo email=user${i}@foo.com name=user${i} | curl -i -X POST -d @- $APP_URL; done In the Lambda function logs, you should see the messages that you sent. Conclusion You were able to set up, configure and deploy a Go Lambda function and trigger it in response to events sent to a topic in an MSK Serverless cluster!

This article was authored by AWS Sr. Specialist SA Alexander Schueren and published with permission. We all like to build new systems, but there are too many business-critical systems that need to be improved. Thus, constantly evolving system architecture remains a major challenge for engineering teams. Decomposing the monolith is not a new topic. Strategies and techniques like domain-driven design and strangler patterns shaped the industry practice of modernization. NoSQL databases became popular for modernization projects. Better performance, flexible schemas, and cost-effectiveness are key reasons for adoption. They scale better and are more resilient than traditional SQL databases. Using a managed solution and reducing operational overhead is a big plus. But moving data is different: it’s messy and there are many unknowns. How do you design the schema, keep the data consistent, handle failures, or roll back? In this article, we will discuss two strategies that can help transition from SQL to NoSQL more smoothly: change data capture and dual-writes. Continuous Data Migration With agile software development, we now ship small batches of features every week instead of having deployment events twice a year, followed by fragile hotfixes and rollbacks. However, with data migrations, there is a tendency to migrate all the data at once. Well, most of the data migrations are homogenous (SQL to SQL), so the data structure remains compatible. Thus, many commercial tools can convert the schema and replicate data. But migrating from SQL to NoSQL is different. It requires an in-depth analysis of the use case and the access pattern to design a new data model. Once we have it, the challenge is to migrate data continuously and catch and recover from failures. What if we can migrate a single customer record, or ten customers from a specific region, or a specific product category? To avoid downtime, we can migrate the data continuously by applying the migration mechanism to a small subset of data. Over time we gain confidence, refine the mechanism, and expand to a larger dataset. This will ensure stability and we can also capitalize on the better performance or lower cost much earlier. Change Data Capture Change data capture (CDC) is a well-established and widely used method. Most relational database management systems (RDBMS) have an internal storage mechanism to collect data changes, often called transaction logs. Whenever you write, update, or delete data, the system captures this information. This is useful if you want to roll back to a previous state, move back in time or replicate data. We can hook into the transaction log and forward the data changes to another system. When moving data from SQL to AWS database services, such as Amazon RDS, AWS Database Migration Service (AWS DMS) is a popular choice. In combination with the schema conversion tool, you can move from Microsoft SQL or Oracle server to an open-source database, such as PostgreSQL or MySQL. But with DMS, we can also move from SQL to NoSQL databases, such as Amazon DynamoDB, S3, Neptune, Kinesis, Kafka, OpenSearch, Redis, and many others. Here is how it works: Define the source and the target endpoints with the right set of permission for read and write operations. Create a task definition specifying the CDC migration process. Add a table mapping with the rule type object-mapping to specify the partition key and attributes for your DynamoDB table. Here is an example of a mapping rule in AWS DMS: { "rules": [ { "rule-type": "object-mapping", "rule-id": "1", "rule-name": "TransformToDDB", "object-locator": { "schema-name": "source-schema", "table-name": "customer" }, "rule-action": "map-record-to-record", "target-table-name": "customer", "mapping-parameters": [ { "partition-key-name": "CustomerName", "attribute-type": "scalar", "attribute-sub-type": "string", "value": "${FIRST_NAME},${LAST_NAME}" }, { "target-attribute-name": "ContactDetails", "attribute-type": "document", "attribute-sub-type": "dynamodb-map", "value": "..." } ] } ] } This mapping rule will copy the data from the customer table and combine FIRST_NAME and LAST_NAME to a composite hash key, and add ContactDetails column with a DynamoDB map structure. For more information, you can see other object-mapping examples in the documentation. One of the major advantages of using CDC is that it allows for atomic data changes. All the changes made to a database, such as inserts, updates, and deletes, are captured as a single transaction. This ensures that the data replication is consistent, and with a transaction rollback, CDC will propagate these changes to the new system as well. Another advantage of CDC is that it does not require any application code changes. There might be situations when the engineering team can’t change the legacy code easily; for example, with a slow release process or lack of tests to ensure stability. Many database engines support CDC, including MySQL, Oracle, SQL Server, and more. This means you don’t have to write a custom solution to read the transaction logs. Finally, with AWS DMS, you can scale your replication instances to handle more data volume, again without additional code changes. AWS DMS and CDC are useful for database replication and migration but have some drawbacks. The major concern is the higher complexity and costs to set up and manage a replication system. You will spend some time fine-tuning the DMS configuration parameters to get the best performance. It also requires a good understanding of the underlying databases, and it’s challenging to troubleshoot errors or performance issues, especially for those who are not familiar with the subtle details of the database engine, replication mechanism, and transaction logs. Dual Writes Dual writes is another popular approach to migrate data continuously. The idea is to write the data to both systems in parallel in your application code. Once the data is fully replicated, we switch over to the new system entirely. This ensures that data is available in the new system before the cutover, and it also keeps the door open to fall back to the old system. With dual writes, we operate on the application level, as opposed to the database level with CDC; thus, we use more compute resources and need a robust delivery process to change and release code. Here is how it works: Applications continue to write data to the existing SQL-based system as they would. A separate process often called a “dual-writer” gets a copy of the data that has been written to the SQL-based system and writes it to the DynamoDB after the transaction. The dual-writer ensures we write the data to both systems in the same format and with the same constraints, such as unique key constraints. Once the dual-write process is complete, we switch over to read from and write to the DynamoDB system. We can control the data migration and apply dual writes to some data by using feature flags. For example, we can toggle the data replication or apply only to a specific subset. This can be a geographical region, customer size, product type, or a single customer. Because dual writes are instrumented on the application level we don’t run queries against the database directly. We work on the object level in our code. This allows us to have additional transformation, validation, or enrichment of the data. But there are also downsides, code complexity, consistency, and failure handling. Using feature flags helps to control the flow, but we still need to write code, add tests, deploy changes, and have a feature flag store. If you are already using feature flags, this might be negligible; otherwise, it's a good chance to introduce feature flags to your system. Data consistency and failure handling are the primary beasts to tame. Because we copy data after the database transaction, there might be cases of rollbacks, and with dual write, you can miss this case. To counter that, you’d need to collect operational and business metrics to keep track of read and write operations to each system, which will increase confidence over time. Conclusion Modernization is unavoidable and improving existing systems will become more common in the future. Over the years, we have learned how to decouple monolithic systems and with many NoSQL database solutions, we can improve products with better performance and lower costs. CDC and dual-writes are solid mechanisms for migrating data models from SQL to NoSQL. While CDC is more database and infrastructure-heavy, with dual-writes we operate on a code level with more control over data segmentation, but with higher code complexity. Thus, it is crucial to understand the use case and the requirements when deciding which mechanism to implement. Moving data continuously between systems should not be difficult, so we need to invest more and learn how to adjust our data model more easily and securely. Chances are high that this is not the last re-architecting initiative you are doing, and building these capabilities will be useful in the future. Do You Like Video Tutorials? If you are more of a visual person who likes to follow the steps to build something with a nice recorded tutorial, then I would encourage you to subscribe to the Build On AWS YouTube channel. This is the place you should go to dive deep into code and architecture, with people from the AWS community sharing their expertise in a visual and entertaining manner.

With the increase in popularity of online shopping, building an analytics platform for e-commerce is important for any organization, as it provides insights into the business, trends, and customer behavior. But more importantly, it can uncover hidden insights that can trigger revenue-generating business decisions and actions. In this blog, we will learn how to build a complete analytics platform in batch and real-time mode. The real-time analytics pipeline also shows how to detect distributed denial of service (DDoS) and bot attacks, which is a common requirement for such use cases. Introduction E-commerce analytics is the process of collecting data from all of the sources that affect a certain online business. Data analysts or business analysts can then utilize this information to deduce changes in customer behavior and online shopping patterns. E-commerce analytics spans the whole customer journey, starting from discovery through acquisition, conversion, and eventually retention and support. In this two-part blog series, we will build an e-commerce analytical platform that can help to analyze the data in real-time as well as in batch. We will use an e-commerce dataset from Kaggle to simulate the logs of user purchases, product views, cart history, and the user’s journey on the online platform to create two analytical pipelines: Batch processing Online/real-time processing You may like to refer to this session presented at AWS re:Invent 2022 for a video walk-through. Batch Processing The batch processing will involve data ingestion, lakehouse architecture, processing, and visualization using Amazon Kinesis, AWS Glue, Amazon S3, and Amazon QuickSight to draw insights regarding the following: Unique visitors per day During a certain time, the users add products to their carts but don’t buy them Top categories per hour or weekday (i.e., to promote discounts based on trends) To know which brands need more marketing Online/Real-Time Processing The real-time processing would involve detecting DDoS and bot attacks using AWS Lambda, Amazon DynamoDB, Amazon CloudWatch, and AWS SNS.This is the first part of the blog series, where we will focus only on the online/real-time processing data pipeline. In the second part of the blog series, we will dive into batch processing. Dataset For this blog, we are going to use the e-commerce behavior data from a multi-category store. This file contains the behavior data for 7 months (from October 2019 to April 2020) from a large multi-category online store, where each row in the file represents an event. All events are related to products and users. Each event is like a many-to-many relationship between products and users. Architecture Real-Time Processing We are going to build an end-to-end data engineering pipeline where we will start with this e-commerce behavior data from a multi-category store dataset as an input, which we will use to simulate a real-time e-commerce workload. This input raw stream of data will go into an Amazon Kinesis Data Stream (stream1), which will stream the data to Amazon Kinesis Data Analytics for analysis, where we will use an Apache Flink application to detect any DDoS attack, and the filtered data will be sent to another Amazon Kinesis Data Stream (stream2). We are going to use SQL to build the Apache Flink application using Amazon Kinesis Data Analytics and, hence, we would need a metadata store, for which we are going to use AWS Glue Data Catalog. And then this stream2 will trigger an AWS Lambda function which will send an Amazon SNS notification to the stakeholders and shall store the fraudulent transaction details in a DynamoDB table. The architecture would look like this: Batch Processing If we look into the architecture diagram above, we will see that we are not storing the raw incoming data anywhere. As the data enters through Kinesis Data Stream (stream1) we are passing it to Kinesis Data Analytics to analyze. And later on, we might discover some bug in our Apache Flink application, and at that point, we will fix the bug and resume processing the data, but we cannot process the old data (which was processed by our buggy Apache Flink application). This is because we have not stored the raw data anywhere, which can allow us to re-process it later. That's why it's recommended to always have a copy of the raw data stored in some storage (e.g., on Amazon S3) so that we can revisit the data if needed for reprocessing and/or batch processing. This is exactly what we are going to do. We will use the same incoming data stream from Amazon Kinesis Data Stream (stream1) and pass it on to Kinesis Firehose which can write the data on S3. Then we will use Glue to catalog that data and perform an ETL job using Glue ETL to process/clean that data so that we can further use the data for running some analytical queries using Athena.At last, we would leverage QuickSight to build a dashboard for visualization. Step-By-Step Walkthrough Let's build this application step-by-step. I'm going to use an AWS Cloud9 instance for this project, but it is not mandatory. If you wish to spin up an AWS Cloud9 instance, you may like to follow the steps mentioned here and proceed further. Download the Dataset and Clone the GitHub Repo Clone the project and change it to the right directory: Shell # Clone the project repository git clone https://github.com/debnsuma/build-a-managed-analytics-platform-for-e-commerce-business.git cd build-a-managed-analytics-platform-for-e-commerce-business/ # Create a folder to store the dataset mkdir dataset Download the dataset from here and move the downloaded file (2019-Nov.csv.zip) under the dataset folder: Now, let's unzip the file and create a sample version of the dataset by just taking the first 1000 records from the file. Shell cd dataset unzip 2019-Nov.csv.zip cat 2019-Nov.csv | head -n 1000 > 202019-Nov-sample.csv Create an Amazon S3 Bucket Now we can create an S3 bucket and upload this dataset: Name of the Bucket: e-commerce-raw-us-east-1-dev (replace <BUCKET_NAME> with your own bucket name) Shell # Copy all the files in the S3 bucket aws s3 cp 2019-Nov.csv.zip s3://<BUCKET_NAME>/ecomm_user_activity/p_year=2019/p_month=11/ aws s3 cp 202019-Nov-sample.csv s3://<BUCKET_NAME>/ecomm_user_activity_sample/202019-Nov-sample.csv aws s3 cp 2019-Nov.csv s3://<BUCKET_NAME>/ecomm_user_activity_unconcompressed/p_year=2019/p_month=11/ Create the Kinesis Data Stream Now, let's create the first Kinesis data stream (stream1 in our architecture diagram) which we will be using as the incoming stream. Open the AWS Console and then: Go to Amazon Kinesis. Click on Create data stream. Let's create another Kinesis data stream which we are going to use later on (stream2 in the architecture diagram). This time use the data stream name as e-commerce-raw-user-activity-stream-2. Start the E-Commerce Traffic We can now start the e-commerce traffic, as our Kinesis data stream is ready. This simulator which we are going to use is a simple python script which will read the data from a CSV file (202019-Nov-sample.csv, the dataset which we downloaded earlier) line by line and send it to the Kinesis data stream (stream1). But before you run the simulator, just edit the stream-data-app-simulation.py script with your <BUCKET_NAME> where you have the dataset. Shell # S3 buckect details (UPDATE THIS) BUCKET_NAME = "e-commerce-raw-us-east-1-dev" Once it's updated, we can run the simulator. Shell # Go back to the project root directory cd .. # Run simulator pip install boto3 python code/ecomm-simulation-app/stream-data-app-simulation.py HttpStatusCode: 200 , electronics.smartphone HttpStatusCode: 200 , appliances.sewing_machine HttpStatusCode: 200 , HttpStatusCode: 200 , appliances.kitchen.washer HttpStatusCode: 200 , electronics.smartphone HttpStatusCode: 200 , computers.notebook HttpStatusCode: 200 , computers.notebook HttpStatusCode: 200 , HttpStatusCode: 200 , HttpStatusCode: 200 , electronics.smartphone Integration with Kinesis Data Analytics and Apache Flink Now, we will create an Amazon Kinesis Data Analytics Streaming Application which will analyze this incoming stream for any DDoS or bot attack. Open the AWS Console and then: Go to Amazon Kinesis. Select Analytics applications. Click on Studio notebooks. Click on Create Studio notebook. Use ecomm-streaming-app-v1 as the Studio notebook name. Under the Permissions section, click on Create to create an AWS Glue database, name the database as my-db-ecomm. Use the same database, my-db-ecomm from the dropdown. Click on Create Studio notebook. Now, select the ecomm-streaming-app-v1 Studio notebook and click on Open in Apache Zeppelin: Once the Zeppelin Dashboard comes up, click on Import note and import this notebook: Open the sql-flink-ecomm-notebook-1 notebook. Flink interpreters supported by Apache Zeppelin notebook are Python, IPython, stream SQL, or batch SQL, and we are going to use SQL to write our code. There are many different ways to create a Flink Application but one of the easiest ways is to use Zeppelin notebook. Let's look at this notebook and briefly discuss what are we doing here: First, we are creating a table for the incoming source of data (which is the e-commerce-raw-user-activity-stream-1 incoming stream). Next, we are creating another table for the filtered data (which is for the e-commerce-raw-user-activity-stream-2 outgoing stream). And finally, we are putting the logic to simulate the DDoS attack. We are essentially looking into the last 10 seconds of the data and grouping them by user_id. If we notice more than 5 records within that 10 seconds, Flink will take that user_id and the no. of records within those 10 seconds and will push that data to the e-commerce-raw-user-activity-stream-2 outgoing stream. 5 products in the last 10 seconds, by user_id 1 SQL %flink.ssql /*Option 'IF NOT EXISTS' can be used, to protect the existing Schema */ DROP TABLE IF EXISTS ecomm_user_activity_stream_1; CREATE TABLE ecomm_user_activity_stream_1 ( `event_time` VARCHAR(30), `event_type` VARCHAR(30), `product_id` BIGINT, `category_id` BIGINT, `category_code` VARCHAR(30), `brand` VARCHAR(30), `price` DOUBLE, `user_id` BIGINT, `user_session` VARCHAR(30), `txn_timestamp` TIMESTAMP(3), WATERMARK FOR txn_timestamp as txn_timestamp - INTERVAL '10' SECOND ) PARTITIONED BY (category_id) WITH ( 'connector' = 'kinesis', 'stream' = 'e-commerce-raw-user-activity-stream-1', 'aws.region' = 'us-east-1', 'scan.stream.initpos' = 'LATEST', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' ); /*Option 'IF NOT EXISTS' can be used, to protect the existing Schema */ DROP TABLE IF EXISTS ecomm_user_activity_stream_2; CREATE TABLE ecomm_user_activity_stream_2 ( `user_id` BIGINT, `num_actions_per_watermark` BIGINT ) WITH ( 'connector' = 'kinesis', 'stream' = 'e-commerce-raw-user-activity-stream-2', 'aws.region' = 'us-east-1', 'format' = 'json', 'json.timestamp-format.standard' = 'ISO-8601' ); /* Inserting aggregation into Stream 2*/ insert into ecomm_user_activity_stream_2select user_id, count(1) as num_actions_per_watermarkfrom ecomm_user_activity_stream_1group by tumble(txn_timestamp, INTERVAL '10' SECOND), user_idhaving count(1) > 5; Create the Apache Flink Application Now that we have our notebook imported, we can create the Flink Application from the notebook directly. To do that: Click on Actions for ecomm-streaming-app-v1 in the top right corner. Click on Build sql-flink-ecomm-notebook-1 > Build and export. It will compile all the codes, will create a ZIP file, and would store the file on S3. Now we can deploy that application by simply clicking on Actions for ecomm-streaming-app-v1 on the top right corner. Click on Deploy sql-flink-ecomm-notebook-1 as Kinesis Analytics application > Deploy using AWS Console. Scroll down and click on Save changes. This is the power of Kinesis Data Analytics: just from a simple Zeppelin Notebook, we can create a real-world application without any hindrance. Finally, we can start the application by clicking on Run. It might take a couple of minutes to start the application, so let's wait until we see the Status as Running. Alarming DDoS Attack If we revisit our architecture, we will see that we are almost done with the real-time/online processing. The only thing which is pending is to create a Lambda function which will be triggered whenever there is an entry of a record inside the e-commerce-raw-user-activity-stream-2 stream. The Lambda function would perform the following: Write that record into a DynamoDB table. Send an SNS notification. Update the CloudWatch metrics. Let's first build the code for the Lambda function. The code is available under code/serverless-app folder. Shell # Install the aws_kinesis_agg package cd code/serverless-app/ pip install aws_kinesis_agg -t . # Build the lambda package and download the zip file. zip -r ../lambda-package.zip . # Upload the zip to S3 cd .. aws s3 cp lambda-package.zip s3://e-commerce-raw-us-east-1-dev/src/lambda/ Now, let's create the Lambda function. Open the AWS Lambda console. Click on Create function button. Enter the Function name as ecomm-detect-high-event-volume. Enter the Runtime as Python 3.7. Click on Create function. Once the Lambda function is created we need to upload the code which we stored in S3. Provide the location of the Lambda code and click on Save: We need to provide adequate privileges to our Lambda function so that it can talk to Kinesis Data Streams, DynamoDB, CloudWatch, and SNS. To modify the IAM Role: Go to the Configuration tab > Permission tab on the left. Click on the Role Name. Since this is just for the demo, we are adding Full Access, but it's NOT recommended for the production environment. We should always follow the least privilege principle to grant access to any user/resource. Let's create the SNS Topic: Open the Amazon SNS console. Click on Create Topic. Select the Type as Standard. Provide the Name as ecomm-user-high-severity-incidents. Click on Create Topic. Let's create a DynamoDB table: Open the Amazon DynamoDB console. Click on Create table. Create the table with the following details. Now, we can add the environment variables that are needed for the Lambda Function. These environment variables are used in the lambda function code. The following are the environment variables: Show Time So, now we are all done with the implementation and it's time to start generating the traffic using the python script which we created earlier, and see everything in action!! Shell cd build-a-managed-analytics-platform-for-e-commerce-business python code/ecomm-simulation-app/stream-data-app-simulation.py HttpStatusCode: 200 , electronics.smartphone HttpStatusCode: 200 , appliances.sewing_machine HttpStatusCode: 200 , HttpStatusCode: 200 , appliances.kitchen.washer HttpStatusCode: 200 , electronics.smartphone HttpStatusCode: 200 , computers.notebook HttpStatusCode: 200 , computers.notebook HttpStatusCode: 200 , HttpStatusCode: 200 , HttpStatusCode: 200 , electronics.smartphone HttpStatusCode: 200 , furniture.living_room.sofa We can also monitor this traffic using the Apache Flink Dashboard: Open the Amazon Kinesis Application dashboard. Select the Application, ecomm-streaming-app-v1-sql-flink-ecomm-notebook-1-2HFDAA9HY. Click on Open Apache Flink dashboard. Once you are on the Open Apache Flink dashboard. Click on Running Jobs > Job Name which is running. Finally, we can also see all the details of the users, which are classified as a DDoS attack by the Flink Application in the DynamoDB table. You can let the simulator run for the next 5-10 mins and can explore and monitor all the components we have built in this whole data pipeline. Summary In this blog post, we built an e-commerce analytical platform that can help analyze the data in real-time. We used a python script to simulate the real traffic using the dataset and used Amazon Kinesis as the incoming stream of data. That data is being analyzed by Amazon Kinesis Data Analytics using Apache Flink using SQL, which involves detecting distributed denial-of-service (DDoS) and bot attacks using AWS Lambda, DynamoDB, CloudWatch, and AWS SNS. In the second part of this blog series, we will dive deep and build the batch processing pipeline and build a dashboard using Amazon QuickSight, which will help us to get more insights about users. It will help us to know details like, who visits the e-commerce website more frequently, which are the top and bottom selling products, which are the top brands, and so on.

Introduction In this quick how-to guide, I will show you how you can use a Python AWS CDK application to automate the deployment and configuration of your Apache Airflow environments using Managed Workflows for Apache Airflow (MWAA) on AWS. What will you need: an AWS account with the right level of privileges a development environment with the AWS CDK configured and running (at the time of writing, you should be using AWS CDK v2) access to an AWS region where Managed Workflows for Apache Airflow is supported all code used in this how-to guide is provided in this GitHub repository Some things to watch out for: If you are deploying this in an environment that already has VPCs, you may generate an error if you exceed the number of VPCs within your AWS Account (by default, this is set to 5, but this is a soft limit which you can request an increase for). Make sure that the Amazon S3 bucket you define for your MWAA environment does not exist before running the CDK app Getting Started Make sure we are running the correct version of the AWS CDKv2 tool (at least v2.2) and then check out the git repo. Shell cdk --version > 2.28.1 (build d035432) git clone https://github.com/094459/blogpost-cdk-mwaa.git After checking out the repository you will have the following files on your local developer environment. Plain Text ├── app.py ├── cdk.json ├── dags │ ├── sample-cdk-dag-od.py │ └── sample-cdk-dag.py ├── mwaa_cdk │ ├── mwaa_cdk_backend.py │ └── mwaa_cdk_env.py └── requirements.txt The first thing we need to do is update our Python dependencies which are documented in the requirements.txt file. Note! If you are currently using in the process of moving between AWS CDKv1 and v2, then you should check out this blog post to help you prepare for this as the steps that follow may fail. Shell pip install -r requirements.txt Exploring the CDK Stack Our AWS CDK application consists of a number of files. The entry point to our application is the app.py file, where we define the structure and resources we are going to build. We then have two CDK stacks that deploy and configure AWS resources. Finally, we have resources that we deploy to our target Apache Airflow environment. If we take a look at the app.py file, we can see explore our CDK application in more detail. We are creating two stacks, one called mwaa_cdk_backend and the other called mwaa_cdk_env. The mwaa_cdk_backend will be used to set up the VPC network that the MWAA environment is going to use. The mwaa_cdk_env is the stack that will configure your MWAA environment. In order to do both though, first, we set up some configuration parameters so that we can maximise the re-use of this CDK application Python import aws_cdk as cdk from mwaa_cdk.mwaa_cdk_backend import MwaaCdkStackBackend from mwaa_cdk.mwaa_cdk_env import MwaaCdkStackEnv env_EU=cdk.Environment(region="{your-aws-region}", account="{your-aws-ac}") mwaa_props = {'dagss3location': '{your-unqiue-s3-bucket}','mwaa_env' : '{name-of-your-mwaa-env}'} app = cdk.App() mwaa_hybrid_backend = MwaaCdkStackBackend( scope=app, id="mwaa-hybrid-backend", env=env_EU, mwaa_props=mwaa_props ) mwaa_hybrid_env = MwaaCdkStackEnv( scope=app, id="mwaa-hybrid-environment", vpc=mwaa_hybrid_backend.vpc, env=env_EU, mwaa_props=mwaa_props ) app.synth() We define configuration parameters in the env_EU and mwaa_props lines. This will allow you to re-use this stack to create multiple different environments. You can also add/change the variables in mwaa_props if you wanted to make other configuration options changeable via a configuration property (for example, logging verbosity or perhaps the version of Apache Airflow) After changing the values in the app.py file and saving, we are ready to deploy. mwaa_cdk_backend There is nothing particularly interesting about this other than it creates the underlying network infrastructure that MWAA needs. There is nothing you need to do, but if you do want to experiment, then what I would say is that a) ensure you read and follow the networking guidance on the MWAA documentation site, as they provide you with details on what needs to be set up, b) if you are trying to lock down the networking, try just deploying the backend stack, and then manually creating an MWAA environment to see if it works/fails. Python from aws_cdk import ( aws_iam as iam, aws_ec2 as ec2, Stack, CfnOutput ) from constructs import Construct class MwaaCdkStackBackend(Stack): def __init__(self, scope: Construct, id: str, mwaa_props, **kwargs) -> None: super().__init__(scope, id, **kwargs) # Create VPC network self.vpc = ec2.Vpc( self, id="MWAA-Hybrid-ApacheAirflow-VPC", cidr="10.192.0.0/16", max_azs=2, nat_gateways=1, subnet_configuration=[ ec2.SubnetConfiguration( name="public", cidr_mask=24, reserved=False, subnet_type=ec2.SubnetType.PUBLIC), ec2.SubnetConfiguration( name="private", cidr_mask=24, reserved=False, subnet_type=ec2.SubnetType.PRIVATE_WITH_NAT) ], enable_dns_hostnames=True, enable_dns_support=True ) CfnOutput( self, id="VPCId", value=self.vpc.vpc_id, description="VPC ID", export_name=f"{self.region}:{self.account}:{self.stack_name}:vpc-id" ) We can see that once this stack has deployed, it will output the VPC details via the console as well as via the AWS CloudFormation Output tab. mwaa_cdk_env The MWAA environment stack is a little more interesting and I will break it down. The first part of the stack configures the Amazon S3 buckets that MWAA will use. Python from aws_cdk import ( aws_iam as iam, aws_ec2 as ec2, aws_s3 as s3, aws_s3_deployment as s3deploy, aws_mwaa as mwaa, aws_kms as kms, Stack, CfnOutput, Tags ) from constructs import Construct class MwaaCdkStackEnv(Stack): def __init__(self, scope: Construct, id: str, vpc, mwaa_props, **kwargs) -> None: super().__init__(scope, id, **kwargs) key_suffix = 'Key' # Create MWAA S3 Bucket and upload local dags s3_tags = { 'env': f"{mwaa_props['mwaa_env']}", 'service': 'MWAA Apache AirFlow' } dags_bucket = s3.Bucket( self, "mwaa-dags", bucket_name=f"{mwaa_props['dagss3location'].lower()}", versioned=True, block_public_access=s3.BlockPublicAccess.BLOCK_ALL ) for tag in s3_tags: Tags.of(dags_bucket).add(tag, s3_tags[tag]) s3deploy.BucketDeployment(self, "DeployDAG", sources=[s3deploy.Source.asset("./dags")], destination_bucket=dags_bucket, destination_key_prefix="dags", prune=False, retain_on_delete=False ) dags_bucket_arn = dags_bucket.bucket_arn What this also does, however, is it takes all the files it finds in the local dags folder (in this particular example, and what is in the GitHub repo, this will be two DAGs, sample-cdk-dag-od.py and sample-cdk-dag.py) and uploads those as part of the deployment process. You can tweak this to your own requirements if you want, and even comment it out/remove it as needed if you do not need to do this. Next up we have the code that creates the MWAA execution policy and the associated role that will be used by the MWAA worker nodes. This is taken from the MWAA documentation, but you can adjust it as needed for your own environment. You might need to do this if you are integrating with other AWS services — this has been set up with default none access, so anything you need to do will need to be added. Python mwaa_policy_document = iam.PolicyDocument( statements=[ iam.PolicyStatement( actions=["airflow:PublishMetrics"], effect=iam.Effect.ALLOW, resources=[f"arn:aws:airflow:{self.region}:{self.account}:environment/{mwaa_props['mwaa_env']}"], ), iam.PolicyStatement( actions=[ "s3:ListAllMyBuckets" ], effect=iam.Effect.DENY, resources=[ f"{dags_bucket_arn}/*", f"{dags_bucket_arn}" ], ), iam.PolicyStatement( actions=[ "s3:*" ], effect=iam.Effect.ALLOW, resources=[ f"{dags_bucket_arn}/*", f"{dags_bucket_arn}" ], ), iam.PolicyStatement( actions=[ "logs:CreateLogStream", "logs:CreateLogGroup", "logs:PutLogEvents", "logs:GetLogEvents", "logs:GetLogRecord", "logs:GetLogGroupFields", "logs:GetQueryResults", "logs:DescribeLogGroups" ], effect=iam.Effect.ALLOW, resources=[f"arn:aws:logs:{self.region}:{self.account}:log-group:airflow-{mwaa_props['mwaa_env']}-*"], ), iam.PolicyStatement( actions=[ "logs:DescribeLogGroups" ], effect=iam.Effect.ALLOW, resources=["*"], ), iam.PolicyStatement( actions=[ "sqs:ChangeMessageVisibility", "sqs:DeleteMessage", "sqs:GetQueueAttributes", "sqs:GetQueueUrl", "sqs:ReceiveMessage", "sqs:SendMessage" ], effect=iam.Effect.ALLOW, resources=[f"arn:aws:sqs:{self.region}:*:airflow-celery-*"], ), iam.PolicyStatement( actions=[ "ecs:RunTask", "ecs:DescribeTasks", "ecs:RegisterTaskDefinition", "ecs:DescribeTaskDefinition", "ecs:ListTasks" ], effect=iam.Effect.ALLOW, resources=[ "*" ], ), iam.PolicyStatement( actions=[ "iam:PassRole" ], effect=iam.Effect.ALLOW, resources=[ "*" ], conditions= { "StringLike": { "iam:PassedToService": "ecs-tasks.amazonaws.com" } }, ), iam.PolicyStatement( actions=[ "kms:Decrypt", "kms:DescribeKey", "kms:GenerateDataKey*", "kms:Encrypt", "kms:PutKeyPolicy" ], effect=iam.Effect.ALLOW, resources=["*"], conditions={ "StringEquals": { "kms:ViaService": [ f"sqs.{self.region}.amazonaws.com", f"s3.{self.region}.amazonaws.com", ] } }, ), ] ) mwaa_service_role = iam.Role( self, "mwaa-service-role", assumed_by=iam.CompositePrincipal( iam.ServicePrincipal("airflow.amazonaws.com"), iam.ServicePrincipal("airflow-env.amazonaws.com"), iam.ServicePrincipal("ecs-tasks.amazonaws.com"), ), inline_policies={"CDKmwaaPolicyDocument": mwaa_policy_document}, path="/service-role/" ) The next part configures the security group and subnets needed by MWAA. Python security_group = ec2.SecurityGroup( self, id = "mwaa-sg", vpc = vpc, security_group_name = "mwaa-sg" ) security_group_id = security_group.security_group_id security_group.connections.allow_internally(ec2.Port.all_traffic(),"MWAA") subnets = [subnet.subnet_id for subnet in vpc.private_subnets] network_configuration = mwaa.CfnEnvironment.NetworkConfigurationProperty( security_group_ids=[security_group_id], subnet_ids=subnets, ) The final part is the most interesting from the MWAA perspective, which is setting up and then configuring the environment. I have commented some of the environment settings out, so feel free to adjust for your own needs. The first thing we do is create a configuration for the MWAA logging. In this particular configuration, I have enabled everything with INFO level logging so feel free to enable/disable or change the logging level as you need. Python logging_configuration = mwaa.CfnEnvironment.LoggingConfigurationProperty( dag_processing_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty( enabled=True, log_level="INFO" ), task_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty( enabled=True, log_level="INFO" ), worker_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty( enabled=True, log_level="INFO" ), scheduler_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty( enabled=True, log_level="INFO" ), webserver_logs=mwaa.CfnEnvironment.ModuleLoggingConfigurationProperty( enabled=True, log_level="INFO" ) ) Next up we define some MWAA Apache Airflow configuration parameters. If you use custom properties, then this is where you will add them. Also, if you want to use TAGs for your MWAA environment, you can adjust accordingly. Python options = { 'core.load_default_connections': False, 'core.load_examples': False, 'webserver.dag_default_view': 'tree', 'webserver.dag_orientation': 'TB' } tags = { 'env': f"{mwaa_props['mwaa_env']}", 'service': 'MWAA Apache AirFlow' } Next, we need to create some additional IAM policies and permissions as well as an AWS KMS encryption key to keep everything encrypted. This part is optional if you decide to not configure KMS encryption when configuring your MWAA environment, but I have included the info here. Python kms_mwaa_policy_document = iam.PolicyDocument( statements=[ iam.PolicyStatement( actions=[ "kms:Create*", "kms:Describe*", "kms:Enable*", "kms:List*", "kms:Put*", "kms:Decrypt*", "kms:Update*", "kms:Revoke*", "kms:Disable*", "kms:Get*", "kms:Delete*", "kms:ScheduleKeyDeletion", "kms:GenerateDataKey*", "kms:CancelKeyDeletion" ], principals=[ iam.AccountRootPrincipal(), # Optional: # iam.ArnPrincipal(f"arn:aws:sts::{self.account}:assumed-role/AWSReservedSSO_rest_of_SSO_account"), ], resources=["*"]), iam.PolicyStatement( actions=[ "kms:Decrypt*", "kms:Describe*", "kms:GenerateDataKey*", "kms:Encrypt*", "kms:ReEncrypt*", "kms:PutKeyPolicy" ], effect=iam.Effect.ALLOW, resources=["*"], principals=[iam.ServicePrincipal("logs.amazonaws.com", region=f"{self.region}")], conditions={"ArnLike": {"kms:EncryptionContext:aws:logs:arn": f"arn:aws:logs:{self.region}:{self.account}:*"}, ), ] ) key = kms.Key( self, f"{mwaa_props['mwaa_env']}{key_suffix}", enable_key_rotation=True, policy=kms_mwaa_policy_document ) key.add_alias(f"alias/{mwaa_props['mwaa_env']}{key_suffix}") Now we come to actually creating the environment, using the stuff we have created or set up above. The following represents all the typical configuration options for the core Apache Airflow options within MWAA. You can change them to suit your own environment or parameterise them as mentioned above. Python managed_airflow = mwaa.CfnEnvironment( scope=self, id='airflow-test-environment', name=f"{mwaa_props['mwaa_env']}", airflow_configuration_options={'core.default_timezone': 'utc'}, airflow_version='2.0.2', dag_s3_path="dags", environment_class='mw1.small', execution_role_arn=mwaa_service_role.role_arn, kms_key=key.key_arn, logging_configuration=logging_configuration, max_workers=5, network_configuration=network_configuration, #plugins_s3_object_version=None, #plugins_s3_path=None, #requirements_s3_object_version=None, #requirements_s3_path=None, source_bucket_arn=dags_bucket_arn, webserver_access_mode='PUBLIC_ONLY', #weekly_maintenance_window_start=None ) managed_airflow.add_override('Properties.AirflowConfigurationOptions', options) managed_airflow.add_override('Properties.Tags', tags) CfnOutput( self, id="MWAASecurityGroup", value=security_group_id, description="Security Group name used by MWAA" ) This stack also outputs the MWAA security group, but you could export other information as well. Deploying Your CDK Application Now that we have reviewed the app, and modified it so that it contains your details (your AWS account/unique S3 bucket/etc), you can now run the app and deploy the CDK stacks. To do this we use the "cdk deploy" command. First of all, from the directory, make sure everything is working ok. To do this we can use the "cdk ls" command. It should return the following (which are the ids assigned in the stacks that this CDK application uses) if it is working ok. Shell cdk ls >MWAA-Backend >MWAA-Environment We can now deploy them, either altogether or one at a time. This CDK application needs the MWAA-Backend app deployed first as it contains the VPC networking that will be used in the MWAA-Environment stack, so we can deploy that by: Shell cdk deploy MWAA-Backend And if it is working ok, it should look similar to the following: Plain Text ✨ Synthesis time: 7.09s mwaa-hybrid-backend: deploying... [0%] start: Publishing 2695cb7a9f601cf94a4151c65c9069787d9ec312084346f2f4359e3f55ff2310:704533066374-eu-central-1 [100%] success: Published 2695cb7a9f601cf94a4151c65c9069787d9ec312084346f2f4359e3f55ff2310:704533066374-eu-central-1 mwaa-hybrid-backend: creating CloudFormation changeset... ✅ mwaa-hybrid-backend ✨ Deployment time: 172.13s Outputs: mwaa-hybrid-backend.ExportsOutputRefMWAAHybridApacheAirflowVPC677B092EF6F2F587 = vpc-0bbdeee3652ef21ff mwaa-hybrid-backend.ExportsOutputRefMWAAHybridApacheAirflowVPCprivateSubnet1Subnet2A6995DF7F8D3134 = subnet-01e48db64381efc7f mwaa-hybrid-backend.ExportsOutputRefMWAAHybridApacheAirflowVPCprivateSubnet2SubnetA28659530C36370A = subnet-0321530b8154f9bd2 mwaa-hybrid-backend.VPCId = vpc-0bbdeee3652ef21ff Stack ARN: arn:aws:cloudformation:eu-central-1:704533066374:stack/mwaa-hybrid-backend/b05897d0-f087-11ec-b5f3-02db3f47a5ca ✨ Total time: 179.22s You can then track/view what has been deployed by checking the CloudFormation stack via the AWS Console. We can now deploy the MWAA environment, which we can do simply by typing: Shell cdk deploy MWAA-Environment This time, it will pop up details about some of the security-related information, in this case the IAM policies and security groups that I mentioned earlier. Answer "Y" to deploy these changes. This will kick off the deployment which you can track by going to the CloudFormation console. This will take approx 20-25 minutes, so a good time to grab a cup of tea and read some of my other blog posts perhaps :-) If it has been successful, you will see the following output (again, your details will change but it should look similar to this): Plain Text Including dependency stacks: mwaa-hybrid-backend [Warning at /mwaa-hybrid-environment/mwaa-sg] Ignoring Egress rule since 'allowAllOutbound' is set to true; To add customize rules, set allowAllOutbound=false on the SecurityGroup ✨ Synthesis time: 12.37s mwaa-hybrid-backend mwaa-hybrid-backend: deploying... [0%] start: Publishing 2695cb7a9f601cf94a4151c65c9069787d9ec312084346f2f4359e3f55ff2310:704533066374-eu-central-1 [100%] success: Published 2695cb7a9f601cf94a4151c65c9069787d9ec312084346f2f4359e3f55ff2310:704533066374-eu-central-1 ✅ mwaa-hybrid-backend (no changes) ✨ Deployment time: 1.97s Outputs: mwaa-hybrid-backend.ExportsOutputRefMWAAHybridApacheAirflowVPC677B092EF6F2F587 = vpc-0bbdeee3652ef21ff mwaa-hybrid-backend.ExportsOutputRefMWAAHybridApacheAirflowVPCprivateSubnet1Subnet2A6995DF7F8D3134 = subnet-01e48db64381efc7f mwaa-hybrid-backend.ExportsOutputRefMWAAHybridApacheAirflowVPCprivateSubnet2SubnetA28659530C36370A = subnet-0321530b8154f9bd2 mwaa-hybrid-backend.VPCId = vpc-0bbdeee3652ef21ff Stack ARN: arn:aws:cloudformation:eu-central-1:704533066374:stack/mwaa-hybrid-backend/b05897d0-f087-11ec-b5f3-02db3f47a5ca ✨ Total time: 14.35s mwaa-hybrid-environment This deployment will make potentially sensitive changes according to your current security approval level (--require-approval broadening). Please confirm you intend to make the following modifications: IAM Statement Changes ┌───┬──────────────────────────────┬────────┬──────────────────────────────┬──────────────────────────────┬─────────────────────────────────┐ │ │ Resource │ Effect │ Action │ Principal │ Condition │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ ${Custom::CDKBucketDeploymen │ Allow │ sts:AssumeRole │ Service:lambda.amazonaws.com │ │ │ │ t8693BB64968944B69AAFB0CC9EB │ │ │ │ │ │ │ 8756C/ServiceRole.Arn} │ │ │ │ │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ ${mwaa-dags.Arn} │ Deny │ s3:ListAllMyBuckets │ AWS:${mwaa-service-role} │ │ │ │ ${mwaa-dags.Arn}/* │ │ │ │ │ │ + │ ${mwaa-dags.Arn} │ Allow │ s3:* │ AWS:${mwaa-service-role} │ │ │ │ ${mwaa-dags.Arn}/* │ │ │ │ │ │ + │ ${mwaa-dags.Arn} │ Allow │ s3:Abort* │ AWS:${Custom::CDKBucketDeplo │ │ │ │ ${mwaa-dags.Arn}/* │ │ s3:DeleteObject* │ yment8693BB64968944B69AAFB0C │ │ │ │ │ │ s3:GetBucket* │ C9EB8756C/ServiceRole} │ │ │ │ │ │ s3:GetObject* │ │ │ │ │ │ │ s3:List* │ │ │ │ │ │ │ s3:PutObject │ │ │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ ${mwaa-hybrid-demoKey.Arn} │ Allow │ kms:CancelKeyDeletion │ AWS:arn:${AWS::Partition}:ia │ │ │ │ │ │ kms:Create* │ m::704533066374:root │ │ │ │ │ │ kms:Decrypt* │ │ │ │ │ │ │ kms:Delete* │ │ │ │ │ │ │ kms:Describe* │ │ │ │ │ │ │ kms:Disable* │ │ │ │ │ │ │ kms:Enable* │ │ │ │ │ │ │ kms:GenerateDataKey* │ │ │ │ │ │ │ kms:Get* │ │ │ │ │ │ │ kms:List* │ │ │ │ │ │ │ kms:Put* │ │ │ │ │ │ │ kms:Revoke* │ │ │ │ │ │ │ kms:ScheduleKeyDeletion │ │ │ │ │ │ │ kms:Update* │ │ │ │ + │ ${mwaa-hybrid-demoKey.Arn} │ Allow │ kms:Decrypt* │ Service:logs.eu-central-1.am │ "ArnLike": { │ │ │ │ │ kms:Describe* │ azonaws.com │ "kms:EncryptionContext:aws:lo │ │ │ │ │ kms:Encrypt* │ │ gs:arn": "arn:aws:logs:eu-centr │ │ │ │ │ kms:GenerateDataKey* │ │ al-1:704533066374:*" │ │ │ │ │ kms:PutKeyPolicy │ │ } │ │ │ │ │ kms:ReEncrypt* │ │ │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ ${mwaa-service-role.Arn} │ Allow │ sts:AssumeRole │ Service:airflow-env.amazonaw │ │ │ │ │ │ │ s.com │ │ │ │ │ │ │ Service:airflow.amazonaws.co │ │ │ │ │ │ │ m │ │ │ │ │ │ │ Service:ecs-tasks.amazonaws. │ │ │ │ │ │ │ com │ │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ * │ Allow │ logs:DescribeLogGroups │ AWS:${mwaa-service-role} │ │ │ + │ * │ Allow │ ecs:DescribeTaskDefinition │ AWS:${mwaa-service-role} │ │ │ │ │ │ ecs:DescribeTasks │ │ │ │ │ │ │ ecs:ListTasks │ │ │ │ │ │ │ ecs:RegisterTaskDefinition │ │ │ │ │ │ │ ecs:RunTask │ │ │ │ + │ * │ Allow │ iam:PassRole │ AWS:${mwaa-service-role} │ "StringLike": { │ │ │ │ │ │ │ "iam:PassedToService": "ecs-t │ │ │ │ │ │ │ asks.amazonaws.com" │ │ │ │ │ │ │ } │ │ + │ * │ Allow │ kms:Decrypt │ AWS:${mwaa-service-role} │ "StringEquals": { │ │ │ │ │ kms:DescribeKey │ │ "kms:ViaService": [ │ │ │ │ │ kms:Encrypt │ │ "sqs.eu-central-1.amazonaws │ │ │ │ │ kms:GenerateDataKey* │ │ .com", │ │ │ │ │ kms:PutKeyPolicy │ │ "s3.eu-central-1.amazonaws. │ │ │ │ │ │ │ com" │ │ │ │ │ │ │ ] │ │ │ │ │ │ │ } │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ arn:${AWS::Partition}:s3:::c │ Allow │ s3:GetBucket* │ AWS:${Custom::CDKBucketDeplo │ │ │ │ dk-hnb659fds-assets-70453306 │ │ s3:GetObject* │ yment8693BB64968944B69AAFB0C │ │ │ │ 6374-eu-central-1 │ │ s3:List* │ C9EB8756C/ServiceRole} │ │ │ │ arn:${AWS::Partition}:s3:::c │ │ │ │ │ │ │ dk-hnb659fds-assets-70453306 │ │ │ │ │ │ │ 6374-eu-central-1/* │ │ │ │ │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ arn:aws:airflow:eu-central-1 │ Allow │ airflow:PublishMetrics │ AWS:${mwaa-service-role} │ │ │ │ :704533066374:environment/mw │ │ │ │ │ │ │ aa-hybrid-demo │ │ │ │ │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ arn:aws:logs:eu-central-1:70 │ Allow │ logs:CreateLogGroup │ AWS:${mwaa-service-role} │ │ │ │ 4533066374:log-group:airflow │ │ logs:CreateLogStream │ │ │ │ │ -mwaa-hybrid-demo-* │ │ logs:DescribeLogGroups │ │ │ │ │ │ │ logs:GetLogEvents │ │ │ │ │ │ │ logs:GetLogGroupFields │ │ │ │ │ │ │ logs:GetLogRecord │ │ │ │ │ │ │ logs:GetQueryResults │ │ │ │ │ │ │ logs:PutLogEvents │ │ │ ├───┼──────────────────────────────┼────────┼──────────────────────────────┼──────────────────────────────┼─────────────────────────────────┤ │ + │ arn:aws:sqs:eu-central-1:*:a │ Allow │ sqs:ChangeMessageVisibility │ AWS:${mwaa-service-role} │ │ │ │ irflow-celery-* │ │ sqs:DeleteMessage │ │ │ │ │ │ │ sqs:GetQueueAttributes │ │ │ │ │ │ │ sqs:GetQueueUrl │ │ │ │ │ │ │ sqs:ReceiveMessage │ │ │ │ │ │ │ sqs:SendMessage │ │ │ └───┴──────────────────────────────┴────────┴──────────────────────────────┴──────────────────────────────┴─────────────────────────────────┘ IAM Policy Changes ┌───┬───────────────────────────────────────────────────────────────────┬───────────────────────────────────────────────────────────────────┐ │ │ Resource │ Managed Policy ARN │ ├───┼───────────────────────────────────────────────────────────────────┼───────────────────────────────────────────────────────────────────┤ │ + │ ${Custom::CDKBucketDeployment8693BB64968944B69AAFB0CC9EB8756C/Ser │ arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasic │ │ │ viceRole} │ ExecutionRole │ └───┴───────────────────────────────────────────────────────────────────┴───────────────────────────────────────────────────────────────────┘ Security Group Changes ┌───┬────────────────────┬─────┬────────────┬────────────────────┐ │ │ Group │ Dir │ Protocol │ Peer │ ├───┼────────────────────┼─────┼────────────┼────────────────────┤ │ + │ ${mwaa-sg.GroupId} │ In │ Everything │ ${mwaa-sg.GroupId} │ │ + │ ${mwaa-sg.GroupId} │ Out │ Everything │ Everyone (IPv4) │ └───┴────────────────────┴─────┴────────────┴────────────────────┘ (NOTE: There may be security-related changes not in this list. See https://github.com/aws/aws-cdk/issues/1299) Do you wish to deploy these changes (y/n)? y mwaa-hybrid-environment: deploying... [0%] start: Publishing e9882ab123687399f934da0d45effe675ecc8ce13b40cb946f3e1d6141fe8d68:704533066374-eu-central-1 [0%] start: Publishing 983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4c:704533066374-eu-central-1 [0%] start: Publishing 91ab667f7c88c3b87cf958b7ef4158ef85fb9ba8bd198e5e0e901bb7f904d560:704533066374-eu-central-1 [0%] start: Publishing f2a926ee3d8ca4bd02b0cf073eb2bbb682e94c021925bf971a9730045ef4fb02:704533066374-eu-central-1 [25%] success: Published 983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4c:704533066374-eu-central-1 [50%] success: Published 91ab667f7c88c3b87cf958b7ef4158ef85fb9ba8bd198e5e0e901bb7f904d560:704533066374-eu-central-1 [75%] success: Published f2a926ee3d8ca4bd02b0cf073eb2bbb682e94c021925bf971a9730045ef4fb02:704533066374-eu-central-1 [100%] success: Published e9882ab123687399f934da0d45effe675ecc8ce13b40cb946f3e1d6141fe8d68:704533066374-eu-central-1 mwaa-hybrid-environment: creating CloudFormation changeset... ✅ mwaa-hybrid-environment ✨ Deployment time: 1412.35s Outputs: mwaa-hybrid-environment.MWAASecurityGroup = sg-0ea83e01caded2bb3 Stack ARN: arn:aws:cloudformation:eu-central-1:704533066374:stack/mwaa-hybrid-environment/450337a0-f088-11ec-a169-06ba63bfdfb2 ✨ Total time: 1424.72s Testing the Environment If we take a look at the Amazon S3 bucket we can see we have our MWAA bucket and dags folder created, as well as our local DAGs uploaded. If we go to the MWAA console, we can see our environment We can now grab the URL for this environment, either by getting it from the console or by using the AWS CLI. Just substitute the name of the MWAA environment and AWS region, and it should then give you the URL you can use in your browser (although you will have to append /home to it) Note I am using jq, if you do not have this in your environment, you can run the command without this but just need to find the entry in the output where it says "WebserverUrl" Shell aws mwaa get-environment --name {name of the environment created} --region={region} | jq -r '.Environment | .WebserverUrl' And as we can see, we have the two sample DAGS that were in the local folder, and are now available for us in the MWAA environment. Removing/Cleaning up our MWAA environment In order to remove everything we have deployed, all we need to do is: Shell cdk destroy MWAA-Environment It will take 20-30 minutes to clean up the MWAA environment. One thing that it will not do, however, is remove the Amazon S3 bucket we set up, so you will need to manually delete that via the console (or use the AWS CLI — that would be my approach). Once you have removed that S3 bucket, now clean up the backend stack Shell cdk destroy MWAA-Backend This should be much quicker to clean up. Once finished, you should be done. What's Next? That's all folks, I hope this has been helpful. Please let me know if you find this how-to guide useful, and if you run into any issues, please log an issue in GitHub and I will take a look.