Description: >-
  Amazon Redshift integration for Apache Spark
Mappings:
  SubnetConfig:
    VPC:
      CIDR: 10.192.0.0/16
    Public1:
      CIDR: 10.192.10.0/24
    Public2:
      CIDR: 10.192.11.0/24
    Private1:
      CIDR: 10.192.20.0/24
    Private2:
      CIDR: 10.192.21.0/24
  ClusterConfigurations:
    emr:
      masterInstanceType: m5.xlarge
      coreInstanceType: m5.xlarge
      masterInstanceCount: 1
      coreInstanceCount: 2
      emrReleaseVersion: emr-6.9.0
      notebookPath: notebooks
      notebookName: redshift-spark-emr.ipynb
    emrServerless:
      emrReleaseVersion: emr-6.9.0
      scriptPath: scripts/
      scriptName: redshift-spark-emr.py
    Glue:
      glueVersion: 4.0
      scriptPath: scripts
      scriptName: redshift-spark-glue.py
    redshift:
      userName: redshiftmasteruser
      dbName: dev
      nodeType: ra3.4xlarge
      # Make sure you use the right value for "nodeCount" when using betweeen "multi-node" and "single-node"
      clusterType: multi-node
      nodeCount: 2
Resources:
  VPC:
    Type: 'AWS::EC2::VPC'
    Properties:
      CidrBlock: !FindInMap 
        - SubnetConfig
        - VPC
        - CIDR
      InstanceTenancy: default
      EnableDnsSupport: 'true'
      EnableDnsHostnames: 'true'
      Tags:
        - Key: Name
          Value: Default-VPC
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  InternetGateway:
    Type: 'AWS::EC2::InternetGateway'
    Properties:
      Tags:
        - Key: Name
          Value: IG
  VPCGatewayAttachment:
    Type: 'AWS::EC2::VPCGatewayAttachment'
    Properties:
      VpcId: !Ref VPC
      InternetGatewayId: !Ref InternetGateway
  PublicSubnet1:
    Type: 'AWS::EC2::Subnet'
    Properties:
      CidrBlock: !FindInMap 
        - SubnetConfig
        - Public1
        - CIDR
      MapPublicIpOnLaunch: 'true'
      AvailabilityZone: !Select 
        - 0
        - !GetAZs 
          Ref: 'AWS::Region'
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: PublicSubnet1
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  PublicSubnet2:
    Type: 'AWS::EC2::Subnet'
    Properties:
      CidrBlock: !FindInMap 
        - SubnetConfig
        - Public2
        - CIDR
      MapPublicIpOnLaunch: 'true'
      AvailabilityZone: !Select 
        - 1
        - !GetAZs 
          Ref: 'AWS::Region'
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: PublicSubnet2
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  PrivateSubnet1:
    Type: 'AWS::EC2::Subnet'
    Properties:
      CidrBlock: !FindInMap 
        - SubnetConfig
        - Private1
        - CIDR
      AvailabilityZone: !Select 
        - 0
        - !GetAZs 
          Ref: 'AWS::Region'
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: PrivateSubnet1
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  PrivateSubnet2:
    Type: 'AWS::EC2::Subnet'
    Properties:
      CidrBlock: !FindInMap 
        - SubnetConfig
        - Private2
        - CIDR
      AvailabilityZone: !Select 
        - 1
        - !GetAZs 
          Ref: 'AWS::Region'
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: PrivateSubnet2
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  NATGateway1EIP:
    Type: 'AWS::EC2::EIP'
    DependsOn: VPCGatewayAttachment
    Properties:
      Domain: vpc
  NATGateway2EIP:
    Type: 'AWS::EC2::EIP'
    DependsOn: VPCGatewayAttachment
    Properties:
      Domain: vpc
  NATGateway1:
    Type: 'AWS::EC2::NatGateway'
    Properties:
      SubnetId: !Ref PublicSubnet1
      AllocationId: !GetAtt 
        - NATGateway1EIP
        - AllocationId
      Tags:
        - Key: Name
          Value: NAT-Gateway1
  NATGateway2:
    Type: 'AWS::EC2::NatGateway'
    Properties:
      SubnetId: !Ref PublicSubnet2
      AllocationId: !GetAtt 
        - NATGateway2EIP
        - AllocationId
      Tags:
        - Key: Name
          Value: NAT-Gateway2
  PublicRouteTable:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: PublicRouteTable
  DefaultPublicRoute:
    Type: 'AWS::EC2::Route'
    DependsOn: VPCGatewayAttachment
    Properties:
      DestinationCidrBlock: 0.0.0.0/0
      RouteTableId: !Ref PublicRouteTable
      GatewayId: !Ref InternetGateway
  PublicSubnet1RouteTableAssociation:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      RouteTableId: !Ref PublicRouteTable
      SubnetId: !Ref PublicSubnet1
  PublicSubnet2RouteTableAssociation:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      RouteTableId: !Ref PublicRouteTable
      SubnetId: !Ref PublicSubnet2
  PrivateRouteTable1:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: PrivateRouteTable1
  DefaultPrivateRoute1:
    Type: 'AWS::EC2::Route'
    Properties:
      DestinationCidrBlock: 0.0.0.0/0
      RouteTableId: !Ref PrivateRouteTable1
      NatGatewayId: !Ref NATGateway1
  PrivateSubnet1RouteTableAssociation:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      RouteTableId: !Ref PrivateRouteTable1
      SubnetId: !Ref PrivateSubnet1
  PrivateRouteTable2:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: PrivateRouteTable2
  DefaultPrivateRoute2:
    Type: 'AWS::EC2::Route'
    Properties:
      DestinationCidrBlock: 0.0.0.0/0
      RouteTableId: !Ref PrivateRouteTable2
      NatGatewayId: !Ref NATGateway2
  PrivateSubnet2RouteTableAssociation:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      RouteTableId: !Ref PrivateRouteTable2
      SubnetId: !Ref PrivateSubnet2
  S3Endpoint:
    Type: 'AWS::EC2::VPCEndpoint'
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Action: '*'
            Principal: '*'
            Resource: '*'
      RouteTableIds:
        - !Ref PrivateRouteTable1
        - !Ref PrivateRouteTable2
        - !Ref PublicRouteTable
      ServiceName: !Sub 'com.amazonaws.${AWS::Region}.s3'
      VpcId: !Ref VPC
  DataBucket:
    Type: AWS::S3::Bucket
    Properties:
      PublicAccessBlockConfiguration:
        BlockPublicAcls: True
        BlockPublicPolicy: True
        IgnorePublicAcls: True
        RestrictPublicBuckets: True
  EMRStudioDefaultSecGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupName: Default-SG
      GroupDescription: Security Group for Default-SG
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: '-1'
          CidrIp: !FindInMap 
            - SubnetConfig
            - VPC
            - CIDR
      Tags:
        - Key: Name
          Value: Default-SG
  VPCDefaultSecurityGroupIngress:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !GetAtt 
        - VPC
        - DefaultSecurityGroup
      IpProtocol: '-1'
      CidrIp: !FindInMap 
        - SubnetConfig
        - VPC
        - CIDR
  EMRSecurityGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupName: EMR-SG
      GroupDescription: Security Group for EMR-SecGroup
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: '-1'
          CidrIp: !FindInMap 
            - SubnetConfig
            - VPC
            - CIDR
      Tags:
        - Key: Name
          Value: EMR-SG
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  EMRSecurityGroupSelfIcmp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSecurityGroup
      IpProtocol: icmp
      FromPort: -1
      ToPort: -1
      SourceSecurityGroupId: !Ref EMRSecurityGroup
  EMRSecurityGroupSelfAllTcp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSecurityGroup
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRSecurityGroup
  EMRSecurityGroupSelfAllUdp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSecurityGroup
      IpProtocol: udp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRSecurityGroup
  EMRSecurityGroupIngressServiceAccess:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSecurityGroup
      IpProtocol: tcp
      FromPort: 8443
      ToPort: 8443
      SourceSecurityGroupId: !Ref EMRServiceAccessSecurityGroup
  EMRServiceAccessSecurityGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupName: EMR-ServiceAccess-SG
      GroupDescription: Security Group for EMRStudio-EMR-SecGroup
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 9443
          ToPort: 9443
          SourceSecurityGroupId: !Ref EMRSecurityGroup
      SecurityGroupEgress:
        - IpProtocol: tcp
          FromPort: 8443
          ToPort: 8443
          DestinationSecurityGroupId: !Ref EMRSecurityGroup
      Tags:
        - Key: Name
          Value: EMR-ServiceAccess-SG
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  WorkspaceSecurityGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupName: EMRStudio-Workspace-SG
      GroupDescription: Security group for EMR Studio Workspace
      VpcId: !Ref VPC
      SecurityGroupEgress:
        - IpProtocol: tcp
          FromPort: 443
          ToPort: 443
          CidrIp: 0.0.0.0/0
        - IpProtocol: tcp
          FromPort: 18888
          ToPort: 18888
          DestinationSecurityGroupId: !GetAtt 
            - EngineSecurityGroup
            - GroupId
      Tags:
        - Key: Name
          Value: EMRStudio-Workspace-SG
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  EngineSecurityGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupName: EMRStudio-Engine-SG
      GroupDescription: Security group for EMR Studio Engine
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 0
          ToPort: 65535
          SourceSecurityGroupId: !Ref EMRSecurityGroup
        - IpProtocol: tcp
          FromPort: 0
          ToPort: 65535
          SourceSecurityGroupId: !Ref EMRServiceAccessSecurityGroup
      Tags:
        - Key: Name
          Value: EMRStudio-Engine-SG
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  EngineSecurityGroupIngress:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EngineSecurityGroup
      IpProtocol: tcp
      FromPort: 18888
      ToPort: 18888
      SourceSecurityGroupId: !Ref WorkspaceSecurityGroup
  EMRStudioServiceRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - elasticmapreduce.amazonaws.com
            Action: 'sts:AssumeRole'
      Policies:
        - PolicyName: EMRStudio-Service-Policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Sid: AllowEMRReadOnlyActions
                Effect: Allow
                Action:
                  - 'elasticmapreduce:ListInstances'
                  - 'elasticmapreduce:DescribeCluster'
                  - 'elasticmapreduce:ListSteps'
                Resource: '*'
              - Sid: AllowEC2ENIActions
                Effect: Allow
                Action:
                  - 'ec2:CreateNetworkInterfacePermission'
                  - 'ec2:DeleteNetworkInterface'
                Resource:
                  - 'arn:aws:ec2:*:*:network-interface/*'
              - Sid: AllowEC2ENIAttributeAction
                Effect: Allow
                Action:
                  - 'ec2:ModifyNetworkInterfaceAttribute'
                Resource:
                  - 'arn:aws:ec2:*:*:instance/*'
                  - 'arn:aws:ec2:*:*:network-interface/*'
                  - 'arn:aws:ec2:*:*:security-group/*'
              - Sid: AllowEC2SecurityGroupActions
                Effect: Allow
                Action:
                  - 'ec2:AuthorizeSecurityGroupEgress'
                  - 'ec2:AuthorizeSecurityGroupIngress'
                  - 'ec2:RevokeSecurityGroupEgress'
                  - 'ec2:RevokeSecurityGroupIngress'
                  - 'ec2:DeleteNetworkInterfacePermission'
                Resource: '*'
              - Sid: AllowDefaultEC2SecurityGroupsCreation
                Effect: Allow
                Action:
                  - 'ec2:CreateSecurityGroup'
                Resource:
                  - 'arn:aws:ec2:*:*:security-group/*'
              - Sid: AllowDefaultEC2SecurityGroupsCreationInVPC
                Effect: Allow
                Action:
                  - 'ec2:CreateSecurityGroup'
                Resource:
                  - 'arn:aws:ec2:*:*:vpc/*'
              - Sid: AllowAddingEMRTagsDuringDefaultSecurityGroupCreation
                Effect: Allow
                Action:
                  - 'ec2:CreateTags'
                Resource: 'arn:aws:ec2:*:*:security-group/*'
                Condition:
                  StringEquals:
                    'ec2:CreateAction': CreateSecurityGroup
              - Sid: AllowEC2ENICreation
                Effect: Allow
                Action:
                  - 'ec2:CreateNetworkInterface'
                Resource:
                  - 'arn:aws:ec2:*:*:network-interface/*'
              - Sid: AllowEC2ENICreationInSubnetAndSecurityGroup
                Effect: Allow
                Action:
                  - 'ec2:CreateNetworkInterface'
                Resource:
                  - 'arn:aws:ec2:*:*:subnet/*'
                  - 'arn:aws:ec2:*:*:security-group/*'
              - Sid: AllowAddingTagsDuringEC2ENICreation
                Effect: Allow
                Action:
                  - 'ec2:CreateTags'
                Resource: 'arn:aws:ec2:*:*:network-interface/*'
                Condition:
                  StringEquals:
                    'ec2:CreateAction': CreateNetworkInterface
              - Sid: AllowEC2ReadOnlyActions
                Effect: Allow
                Action:
                  - 'ec2:DescribeSecurityGroups'
                  - 'ec2:DescribeNetworkInterfaces'
                  - 'ec2:DescribeTags'
                  - 'ec2:DescribeInstances'
                  - 'ec2:DescribeSubnets'
                  - 'ec2:DescribeVpcs'
                Resource: '*'
              - Effect: Allow
                Action:
                  - 's3:PutObject'
                  - 's3:GetObject'
                  - 's3:GetEncryptionConfiguration'
                  - 's3:ListBucket'
                  - 's3:DeleteObject'
                Resource: 'arn:aws:s3:::*'
  EMRStudio:
    Type: AWS::EMR::Studio
    Properties:
      AuthMode: IAM
      DefaultS3Location: !Join [ '', [ 's3://', !Ref DataBucket, '/', !FindInMap [ ClusterConfigurations, emr, notebookPath ]]]
      Name: EMR-Studio
      EngineSecurityGroupId: !Ref EngineSecurityGroup
      WorkspaceSecurityGroupId: !Ref WorkspaceSecurityGroup
      ServiceRole: !GetAtt
        - EMRStudioServiceRole
        - Arn
      VpcId: !Ref VPC
      SubnetIds:
        - !Ref PrivateSubnet1
        - !Ref PrivateSubnet2
      Tags:
        - Key: Name
          Value: EMRStudio
  EMRServiceIAMRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - 'elasticmapreduce.amazonaws.com'
            Action: 'sts:AssumeRole'
      Path: '/'
      Policies:
        - PolicyName: 'EMRServiceRole'
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: 'Allow'
                Action:
                  - ec2:AuthorizeSecurityGroupEgress
                  - ec2:AuthorizeSecurityGroupIngress
                  - ec2:CancelSpotInstanceRequests
                  - ec2:CreateNetworkInterface
                  - ec2:CreateSecurityGroup
                  - ec2:CreateTags
                  - ec2:DeleteNetworkInterface
                  - ec2:DeleteTags
                  - ec2:DescribeAvailabilityZones
                  - ec2:DescribeAccountAttributes
                  - ec2:DescribeDhcpOptions
                  - ec2:DescribeImages
                  - ec2:DescribeInstanceStatus
                  - ec2:DescribeInstances
                  - ec2:DescribeKeyPairs
                  - ec2:DescribeNetworkAcls
                  - ec2:DescribeNetworkInterfaces
                  - ec2:DescribePrefixLists
                  - ec2:DescribeRouteTables
                  - ec2:DescribeSecurityGroups
                  - ec2:DescribeSpotInstanceRequests
                  - ec2:DescribeSpotPriceHistory
                  - ec2:DescribeSubnets
                  - ec2:DescribeTags
                  - ec2:DescribeVpcAttribute
                  - ec2:DescribeVpcEndpoints
                  - ec2:DescribeVpcEndpointServices
                  - ec2:DescribeVpcs
                  - ec2:DetachNetworkInterface
                  - ec2:ModifyImageAttribute
                  - ec2:ModifyInstanceAttribute
                  - ec2:RequestSpotInstances
                  - ec2:RevokeSecurityGroupEgress
                  - ec2:RunInstances
                  - ec2:TerminateInstances
                  - ec2:DeleteVolume
                  - ec2:DescribeVolumeStatus
                  - ec2:DescribeVolumes
                  - ec2:DetachVolume
                  - iam:GetRole
                  - iam:GetRolePolicy
                  - iam:ListInstanceProfiles
                  - iam:ListRolePolicies
                  - s3:CreateBucket
                  - sdb:BatchPutAttributes
                  - sdb:Select
                  - cloudwatch:PutMetricAlarm
                  - cloudwatch:DescribeAlarms
                  - cloudwatch:DeleteAlarms
                  - application-autoscaling:RegisterScalableTarget
                  - application-autoscaling:DeregisterScalableTarget
                  - application-autoscaling:PutScalingPolicy
                  - application-autoscaling:DeleteScalingPolicy
                  - application-autoscaling:Describe*
                Resource: '*'
              - Effect: 'Allow'
                Action:
                  - s3:GetBucketLocation
                  - s3:GetBucketCORS
                  - s3:GetObjectVersionForReplication
                  - s3:GetObject
                  - s3:GetBucketTagging
                  - s3:GetObjectVersion
                  - s3:GetObjectTagging
                  - s3:ListMultipartUploadParts
                  - s3:ListBucketByTags
                  - s3:ListBucket
                  - s3:ListObjects
                  - s3:ListBucketMultipartUploads
                Resource:
                  - !Join [ '', [ !GetAtt DataBucket.Arn, '/*' ]]
              - Effect: 'Allow'
                Action:
                  - sqs:CreateQueue
                  - sqs:DeleteQueue
                  - sqs:DeleteMessage
                  - sqs:DeleteMessageBatch
                  - sqs:GetQueueAttributes
                  - sqs:GetQueueUrl
                  - sqs:PurgeQueue
                  - sqs:ReceiveMessage
                Resource:
                  - !Sub arn:aws:sqs:*:${AWS::AccountId}:AWS-ElasticMapReduce-*
              - Effect: 'Allow'
                Action:
                  - iam:CreateServiceLinkedRole
                Resource:
                  - arn:aws:iam::*:role/aws-service-role/spot.amazonaws.com/AWSServiceRoleForEC2Spot*
                Condition:
                  StringLike:
                    iam:AWSServiceName: spot.amazonaws.com
              - Effect: 'Allow'
                Action:
                  - iam:PassRole
                Resource:
                  - arn:aws:iam::*:role/EMR_AutoScaling_DefaultRole
                  - !GetAtt EMRJobFlowRole.Arn
  EMRJobFlowRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - 'ec2.amazonaws.com'
            Action:
              - 'sts:AssumeRole'
      Path: '/'
      Policies:
        - PolicyName: 'EMRInstanceRole'
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: 'Allow'
                Action:
                  - ec2:Describe*
                  - elasticmapreduce:Describe*
                  - elasticmapreduce:ListBootstrapActions
                  - elasticmapreduce:ListClusters
                  - elasticmapreduce:ListInstanceGroups
                  - elasticmapreduce:ListInstances
                  - elasticmapreduce:ListStep
                Resource: '*'
              - Effect: 'Allow'
                Action:
                  - s3:GetBucketLocation
                  - s3:GetBucketCORS
                  - s3:GetObjectVersionForReplication
                  - s3:GetObject
                  - s3:GetBucketTagging
                  - s3:GetObjectVersion
                  - s3:GetObjectTagging
                  - s3:ListMultipartUploadParts
                  - s3:ListBucketByTags
                  - s3:ListBucket
                  - s3:ListObjects
                  - s3:ListBucketMultipartUploads
                  - s3:PutObject
                  - s3:PutObjectTagging
                  - s3:HeadBucket
                  - s3:DeleteObject
                Resource:
                  - !GetAtt DataBucket.Arn
                  - !Join [ '', [ !GetAtt DataBucket.Arn, '/*' ]]
              - Effect: 'Allow'
                Action:
                  - sqs:CreateQueue
                  - sqs:DeleteQueue
                  - sqs:DeleteMessage
                  - sqs:DeleteMessageBatch
                  - sqs:GetQueueAttributes
                  - sqs:GetQueueUrl
                  - sqs:PurgeQueue
                  - sqs:ReceiveMessage
                Resource:
                  - !Sub arn:aws:sqs:*:${AWS::AccountId}:AWS-ElasticMapReduce-*
              - Sid: RedshiftGetTempCredentials
                Effect: Allow
                Action:
                  - 'redshift:GetClusterCredentials'
                Resource:
                  - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':dbuser:', !Ref RedshiftCluster, '/', !FindInMap [ ClusterConfigurations, redshift, userName ]]]
                  - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':dbname:', !Ref RedshiftCluster, '/', !FindInMap [ ClusterConfigurations, redshift, dbName ]]]
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore'
  EMRJobFlowInstanceProfile:
    Type: 'AWS::IAM::InstanceProfile'
    Properties:
      Roles:
        - !Ref EMRJobFlowRole
  EMRCluster:
    Type: 'AWS::EMR::Cluster'
    DependsOn:
      - DefaultPrivateRoute1
      - DefaultPrivateRoute2
    Properties:
      Name: !Sub emrCluster-${AWS::StackName}
      ReleaseLabel: !FindInMap [ ClusterConfigurations, emr, emrReleaseVersion ]
      JobFlowRole: !Ref EMRJobFlowInstanceProfile
      ServiceRole: !Ref EMRServiceIAMRole
      # AutoScalingRole: !Ref EMRAutoScalingRole
      LogUri: !Sub s3n://${DataBucket}/elasticmapreduce/logs
      VisibleToAllUsers: true
      Applications:
        - Name: Hadoop
        - Name: Spark
        - Name: Livy
        - Name: JupyterEnterpriseGateway
      Configurations:
        - Classification: spark-hive-site
          ConfigurationProperties:
            hive.metastore.client.factory.class: 'com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory'
        - Classification: livy-conf
          ConfigurationProperties:
            livy.file.local-dir-whitelist: '/usr/share/aws/redshift/'
      Instances:
        MasterInstanceGroup:
          Name: MasterInstance
          InstanceCount: !FindInMap [ ClusterConfigurations, emr, masterInstanceCount ]
          InstanceType: !FindInMap [ ClusterConfigurations, emr, masterInstanceType ]
          Market: ON_DEMAND
        CoreInstanceGroup:
          Name: CoreInstance
          InstanceCount: !FindInMap [ ClusterConfigurations, emr, coreInstanceCount ]
          InstanceType: !FindInMap [ ClusterConfigurations, emr, coreInstanceType ]
          Market: ON_DEMAND
        TerminationProtected: false
        Ec2SubnetId: !Ref PrivateSubnet1
        EmrManagedMasterSecurityGroup: !Ref EMRSecurityGroup
        EmrManagedSlaveSecurityGroup: !Ref EMRSecurityGroup
        ServiceAccessSecurityGroup: !Ref EMRServiceAccessSecurityGroup
  EMRServerlessSecurityGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupName: EMR-Serverless-SG
      GroupDescription: Security Group for EMR-Serverless-SecGroup
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: '-1'
          CidrIp: !FindInMap 
            - SubnetConfig
            - VPC
            - CIDR
      Tags:
        - Key: Name
          Value: EMR-Serverless-SG
        - Key: for-use-with-amazon-emr-managed-policies
          Value: 'true'
  EMRServerlessSecurityGroupSelfIcmp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRServerlessSecurityGroup
      IpProtocol: icmp
      FromPort: -1
      ToPort: -1
      SourceSecurityGroupId: !Ref EMRServerlessSecurityGroup
  EMRServerlessSecurityGroupSelfAllTcp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRServerlessSecurityGroup
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRServerlessSecurityGroup
  EMRServerlessSecurityGroupSelfAllUdp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRServerlessSecurityGroup
      IpProtocol: udp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRServerlessSecurityGroup
  EMRServerlessApplication:
    Type: AWS::EMRServerless::Application
    Properties:
      Name: emr-spark-redshift
      ReleaseLabel: !FindInMap [ ClusterConfigurations, emrServerless, emrReleaseVersion ]
      Type: Spark
      NetworkConfiguration:
        SecurityGroupIds:
          - !Ref EMRServerlessSecurityGroup
        SubnetIds:
          - !Ref PrivateSubnet1
          - !Ref PrivateSubnet2
      AutoStartConfiguration:
        Enabled: true
      AutoStopConfiguration:
        Enabled: true
        IdleTimeoutMinutes: 30
  EMRServerlessJobRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - emr-serverless.amazonaws.com
            Action:
              - "sts:AssumeRole"
      Description: "Service role for EMR Studio"
      Policies:
        - PolicyName: EMRServerlessS3andRedshiftAccess
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - s3:ListBucket
                Resource: "*"
              - Effect: Allow
                Action:
                  - s3:GetObject
                  - s3:PutObject
                  - s3:DeleteObject
                Resource:
                  - !Sub 'arn:aws:s3:::${DataBucket}/*'
                  - !Sub 'arn:aws:s3:::${DataBucket}'
              - Sid: RedshiftGetTempCredentials
                Effect: Allow
                Action:
                  - redshift:GetClusterCredentials
                Resource:
                  - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':dbuser:', !Ref RedshiftCluster, '/', !FindInMap [ ClusterConfigurations, redshift, userName ]]]
                  - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':dbname:', !Ref RedshiftCluster, '/', !FindInMap [ ClusterConfigurations, redshift, dbName ]]]
  RedShiftSG:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: 'Security group for Redshift cluster'
      VpcId: !Ref VPC
  RedshiftSGIngressSelfAllTcp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref RedShiftSG
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref RedShiftSG
  RedshiftSGIngressSelfAllUdp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref RedShiftSG
      IpProtocol: udp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref RedShiftSG
  RedshiftSGIngressEMR:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref RedShiftSG
      IpProtocol: tcp
      FromPort: 5439
      ToPort: 5439
      SourceSecurityGroupId: !Ref EMRSecurityGroup
  RedshiftSGIngressEMRServerless:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref RedShiftSG
      IpProtocol: tcp
      FromPort: 5439
      ToPort: 5439
      SourceSecurityGroupId: !Ref EMRServerlessSecurityGroup
  RedshiftSubnetGroup:
    Type: 'AWS::Redshift::ClusterSubnetGroup'
    Properties:
      Description: 'Redshift cluster subnet group'
      SubnetIds:
        - !Ref PrivateSubnet1
        - !Ref PrivateSubnet2
  RedshiftClusterParameterGroup:
    Type: 'AWS::Redshift::ClusterParameterGroup'
    Properties:
      Description: !Join [ " ", [ !Ref 'AWS::StackName', " - Redshift Cluster Parameter group" ]] 
      ParameterGroupFamily: redshift-1.0
  RedshiftCreds:
    Type: 'AWS::SecretsManager::Secret'
    Properties:
      Description: !Sub Redshift cluster master user credentials for ${AWS::StackName}
      GenerateSecretString:
        SecretStringTemplate: !Join [ '', [ '{"username": "', !FindInMap [ ClusterConfigurations, redshift, userName ], '"}' ]]
        GenerateStringKey: 'password'
        PasswordLength: 16
        ExcludePunctuation: true
  RedshiftIamRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - redshift.amazonaws.com
            Action: 'sts:AssumeRole'
      Policies:
        - PolicyName: Redshift-IAM-Policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Sid: AllowAccesstoDataBucket
                Effect: Allow
                Action:
                  - 's3:AbortMultipartUpload'
                  - 's3:DeleteObject'
                  - 's3:GetBucketVersioning'
                  - 's3:GetObject'
                  - 's3:GetObjectTagging'
                  - 's3:GetObjectVersion'
                  - 's3:ListBucket'
                  - 's3:ListBucketMultipartUploads'
                  - 's3:ListBucketVersions'
                  - 's3:ListMultipartUploadParts'
                  - 's3:PutBucketVersioning'
                  - 's3:PutObject'
                  - 's3:PutObjectTagging'
                Resource:
                  - !Sub 'arn:aws:s3:::${DataBucket}'
                  - !Sub 'arn:aws:s3:::${DataBucket}/*'
              - Sid: ListPermissions
                Effect: Allow
                Action:
                  - 's3:ListBucket'
                Resource:
                  - '*'
  RedshiftCluster:
    Type: 'AWS::Redshift::Cluster'
    DependsOn:
      - RedshiftCreds
    Properties:
      ClusterIdentifier: !Sub ${AWS::StackName}-Redshift-Cluster
      DBName: !FindInMap [ ClusterConfigurations, redshift, dbName ]
      MasterUsername: !Join [ '', [ '{{resolve:secretsmanager:', !Ref RedshiftCreds, ':SecretString:username}}' ]]
      MasterUserPassword: !Join [ '', [ '{{resolve:secretsmanager:', !Ref RedshiftCreds, ':SecretString:password}}' ]]
      NodeType: !FindInMap [ ClusterConfigurations, redshift, nodeType ]
      ClusterType: !FindInMap [ ClusterConfigurations, redshift, clusterType ]
      NumberOfNodes: !FindInMap [ ClusterConfigurations, redshift, nodeCount ]
      PubliclyAccessible: false
      VpcSecurityGroupIds:
        - !Ref RedShiftSG
      IamRoles:
        - !GetAtt RedshiftIamRole.Arn
      ClusterSubnetGroupName: !Ref RedshiftSubnetGroup
      ClusterParameterGroupName: !Ref RedshiftClusterParameterGroup
      Encrypted: true
  GlueRedshiftConnection:
    Type: AWS::Glue::Connection
    Properties:
      CatalogId: !Ref AWS::AccountId
      ConnectionInput:
          ConnectionType: JDBC
          Name: !Sub GlueRedshiftConnection-${AWS::StackName}
          PhysicalConnectionRequirements:
            SecurityGroupIdList:
              - !Ref RedShiftSG
            SubnetId: !Ref PrivateSubnet1
            AvailabilityZone: !Select 
              - 0
              - !GetAZs
                Ref: 'AWS::Region'
          ConnectionProperties: {
            "JDBC_CONNECTION_URL": !Join [ '', [ 'jdbc:redshift://', !GetAtt RedshiftCluster.Endpoint.Address, ':', !GetAtt RedshiftCluster.Endpoint.Port, '/', !FindInMap [ ClusterConfigurations, redshift, dbName ]]],
            "USERNAME": !Join [ '', [ '{{resolve:secretsmanager:', !Ref RedshiftCreds, ':SecretString:username}}' ]],
            "PASSWORD": !Join [ '', [ '{{resolve:secretsmanager:', !Ref RedshiftCreds, ':SecretString:password}}' ]]
          }
  GlueJobIamRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - glue.amazonaws.com
            Action: 'sts:AssumeRole'
      Policies:
        - PolicyName: Glue-Job-IAM-Policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Sid: AllowAccesstoDataBucket
                Effect: Allow
                Action:
                  - s3:AbortMultipartUpload
                  - s3:DeleteObject
                  - s3:GetBucketVersioning
                  - s3:GetObject
                  - s3:GetObjectTagging
                  - s3:GetObjectVersion
                  - s3:ListBucket
                  - s3:ListBucketMultipartUploads
                  - s3:ListBucketVersions
                  - s3:ListMultipartUploadParts
                  - s3:PutBucketVersioning
                  - s3:PutObject
                  - s3:PutObjectTagging
                Resource:
                  - !Sub 'arn:aws:s3:::${DataBucket}'
                  - !Sub 'arn:aws:s3:::${DataBucket}/*'
                  - 'arn:aws:s3:::aws-glue-*'
              - Sid: AllowCreateGlueS3Bucket
                Effect: Allow
                Action:
                  - s3:CreateBucket
                  - s3:PutBucketPublicAccessBlock
                Resource:
                  - 'arn:aws:s3:::aws-glue-*'
              - Sid: AllowGlueS3Bucket
                Effect: Allow
                Action:
                  - s3:GetObject
                  - s3:PutObject
                  - s3:DeleteObject
                Resource:
                  - 'arn:aws:s3:::aws-glue-*'
                  - 'arn:aws:s3:::aws-glue-*/*'
                  - 'arn:aws:s3:::*/*aws-glue-*/*'
              - Sid: CloudWatchAccess
                Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource:
                  - 'arn:aws:logs:*:*:/aws-glue/*'
              - Sid: Ec2TagPermissions
                Effect: Allow
                Action:
                  - ec2:CreateTags
                  - ec2:DeleteTags
                Resource:
                  - 'arn:aws:ec2:*:*:network-interface/*'
                  - 'arn:aws:ec2:*:*:security-group/*'
                  - 'arn:aws:ec2:*:*:instance/*'
                Condition:
                  'ForAllValues:StringEquals':
                    'aws:TagKeys':
                      - 'aws-glue-service-resource'
              - Sid: Ec2ReadandOtherPermissions
                Effect: Allow
                Action:
                  - ec2:DescribeVpcEndpoints
                  - ec2:DescribeRouteTables
                  - ec2:CreateNetworkInterface
                  - ec2:DeleteNetworkInterface
                  - ec2:DescribeNetworkInterfaces
                  - ec2:DescribeSecurityGroups
                  - ec2:DescribeSubnets
                  - ec2:DescribeVpcAttribute
                  - iam:ListRolePolicies
                  - iam:GetRole
                  - iam:GetRolePolicy
                  - cloudwatch:PutMetricData
                  - s3:GetBucketLocation
                  - s3:ListBucket
                  - s3:ListAllMyBuckets
                  - s3:GetBucketAcl
                Resource:
                  - '*'
              - Sid: GluePermissions
                Effect: Allow
                Action:
                  - glue:GetConnection
                Resource:
                  - !Join [ '', [ 'arn:aws:glue:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':catalog' ]]
                  - !Join [ '', [ 'arn:aws:glue:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':connection/', !Ref GlueRedshiftConnection ]]
              - Sid: ListPermissions
                Effect: Allow
                Action:
                  - s3:ListBucket
                Resource:
                  - '*'
              - Sid: RedshiftGetTempCredentials
                Effect: Allow
                Action:
                  - redshift:GetClusterCredentials
                Resource:
                  - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':dbuser:', !Ref RedshiftCluster, '/', !FindInMap [ ClusterConfigurations, redshift, userName ]]]
                  - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':dbname:', !Ref RedshiftCluster, '/', !FindInMap [ ClusterConfigurations, redshift, dbName ]]]
  GlueJob:
    Type: AWS::Glue::Job
    DependsOn:
      - UploadScripts
    Properties:
      Name: !Sub "Glue-Job-${AWS::StackName}"
      GlueVersion: !FindInMap [ ClusterConfigurations, Glue, glueVersion ]
      Command:
        Name: glueetl
        ScriptLocation: !Join [ '', [ 's3://', !Ref DataBucket, '/', !FindInMap [ ClusterConfigurations, Glue, scriptPath ], !FindInMap [ ClusterConfigurations, Glue, scriptName ]]]
      Role: !Ref GlueJobIamRole
      Connections:
        Connections:
          - !Ref GlueRedshiftConnection
      DefaultArguments:
                --TempDir: !Join [ '', [ 's3://', !Ref DataBucket, '/glue-temporary/' ]]
                --enable-continuous-cloudwatch-log: "true"
                --enable-glue-datacatalog: "true"
                --enable-job-insights: "true"
                --enable-metrics: "true"
                --enable-spark-ui: "true"
                --job-bookmark-option: "job-bookmark-disable"
                --job-language: "python"
                --spark-event-logs-path: !Join [ '', [ 's3://', !Ref DataBucket, '/glue-history/sparkHistoryLogs/' ]]
                --REDSHIFT_JDBC_URL: !Join [ '', [ 'jdbc:redshift:iam://', !GetAtt RedshiftCluster.Endpoint.Address, ':', !GetAtt RedshiftCluster.Endpoint.Port, '/', !FindInMap [ ClusterConfigurations, redshift, dbName ]]]
                --REDSHIFT_TEMPORARY_DIRECTORY: !Join [ '', [ 's3://', !Ref DataBucket, '/redshift-temp-dir/' ]]
                --AWS_IAM_ROLE_ARN: !GetAtt RedshiftIamRole.Arn
  DownloadZipFile:
    Type: Custom::DownloadZipFile
    Properties:
      ServiceToken: !GetAtt DownloadZipFileFunction.Arn
      downloadUrl: 'https://docs.aws.amazon.com/redshift/latest/gsg/samples/tickitdb.zip'
      downloadPath: '/tmp/download_file/tickitdb.zip'
      uploadS3Bucket: !Ref DataBucket
      uploadS3BucketPath: 'sample_data/'
      redshiftIamRoleArn: !GetAtt RedshiftIamRole.Arn
      regionName: !Ref AWS::Region
      redshiftUserName: !FindInMap [ ClusterConfigurations, redshift, userName ]
      redshiftClusterIdentifier: !Ref RedshiftCluster
      redshiftDatabase: !FindInMap [ ClusterConfigurations, redshift, dbName ]
  DownloadZipFileFunction:
    Type: "AWS::Lambda::Function"
    DependsOn:
      - DataBucket
    Properties:
      Description: "Download the zip file and upload to S3"
      Handler: index.lambda_handler
      Role: !GetAtt DownloadZipFileFunctionExecutionRole.Arn
      Timeout: 360
      Runtime: python3.9
      Code:
        ZipFile: |
          import json
          import boto3
          import urllib3
          import zipfile
          import os
          import time

          def lambda_handler(event, context):
              try:
                  url = event['ResourceProperties']['downloadUrl']
                  file_name = event['ResourceProperties']['downloadPath']
                  file_path = os.path.dirname(file_name)
                  bucket_name = event['ResourceProperties']['uploadS3Bucket']
                  upload_path = event['ResourceProperties']['uploadS3BucketPath']
                  redshift_role = event['ResourceProperties']['redshiftIamRoleArn']
                  region_name = event['ResourceProperties']['regionName']
                  redshift_id = event['ResourceProperties']['redshiftClusterIdentifier']
                  redshift_user = event['ResourceProperties']['redshiftUserName']
                  redshift_db = event['ResourceProperties']['redshiftDatabase']
                  print(event)
                  if event['RequestType'] == 'Create':
                      if not os.path.exists(file_path):
                          os.makedirs(file_path)

                      http = urllib3.PoolManager()
                      r = http.request('GET', url, preload_content=False)
                      with open(file_name, 'wb') as out_file:
                          while True:
                              data = r.read(65536) # Downloading in chunk size of 64kb
                              if not data:
                                  break
                              out_file.write(data)

                      r.release_conn()
                      
                      # Unzip the file - to be used only if we are dealing with zip files
                      unzip_path = '/tmp/extract_folder/'

                      if not os.path.exists(unzip_path):
                          os.makedirs(unzip_path)

                      with zipfile.ZipFile(file_name, 'r') as zip_ref:
                          zip_ref.extractall(unzip_path)

                      # Upload the extracted folder / downloaded file to S3
                      s3_client = boto3.client('s3')

                      for root, dirs, files in os.walk(unzip_path):
                          for file in files:
                              upload_file = upload_path+file
                              s3_client.upload_file(os.path.join(root,file), bucket_name, upload_file)

                      # Load the sample data into Redshift
                      redshift_client = boto3.client('redshift-data')

                      create_schema = "create schema if not exists tickit;"

                      create_users_query = "create table if not exists tickit.users(\
                                            userid integer not null distkey sortkey,\
                                            username char(8),\
                                            firstname varchar(30),\
                                            lastname varchar(30),\
                                            city varchar(30),\
                                            state char(2),\
                                            email varchar(100),\
                                            phone char(14),\
                                            likesports boolean,\
                                            liketheatre boolean,\
                                            likeconcerts boolean,\
                                            likejazz boolean,\
                                            likeclassical boolean,\
                                            likeopera boolean,\
                                            likerock boolean,\
                                            likevegas boolean,\
                                            likebroadway boolean,\
                                            likemusicals boolean);"

                      create_venue_query = "create table if not exists tickit.venue(\
                                            venueid smallint not null distkey sortkey,\
                                            venuename varchar(100),\
                                            venuecity varchar(30),\
                                            venuestate char(2),\
                                            venueseats integer);"

                      create_category_query = "create table if not exists tickit.category(\
                                              catid smallint not null distkey sortkey,\
                                              catgroup varchar(10),\
                                              catname varchar(10),\
                                              catdesc varchar(50));"

                      create_date_query = "create table if not exists tickit.date(\
                                          dateid smallint not null distkey sortkey,\
                                          caldate date not null,\
                                          day character(3) not null,\
                                          week smallint not null,\
                                          month character(5) not null,\
                                          qtr character(5) not null,\
                                          year smallint not null,\
                                          holiday boolean default('N'));"

                      create_event_query = "create table if not exists tickit.event(\
                                            eventid integer not null distkey,\
                                            venueid smallint not null,\
                                            catid smallint not null,\
                                            dateid smallint not null sortkey,\
                                            eventname varchar(200),\
                                            starttime timestamp);"

                      create_listing_query = "create table if not exists tickit.listing(\
                                              listid integer not null distkey,\
                                              sellerid integer not null,\
                                              eventid integer not null,\
                                              dateid smallint not null  sortkey,\
                                              numtickets smallint not null,\
                                              priceperticket decimal(8,2),\
                                              totalprice decimal(8,2),\
                                              listtime timestamp);"

                      create_sales_query = "create table if not exists tickit.sales(\
                                            salesid integer not null,\
                                            listid integer not null distkey,\
                                            sellerid integer not null,\
                                            buyerid integer not null,\
                                            eventid integer not null,\
                                            dateid smallint not null sortkey,\
                                            qtysold smallint not null,\
                                            pricepaid decimal(8,2),\
                                            commission decimal(8,2),\
                                            saletime timestamp);"

                      copy_users_query = "copy tickit.users from 's3://" +bucket_name+ "/" +upload_path+ "allusers_pipe.txt'\
                                          iam_role '" +redshift_role+ "'\
                                          delimiter '|' region '" +region_name+"';"

                      copy_venue_query = "copy tickit.venue from 's3://" +bucket_name+ "/" +upload_path+ "venue_pipe.txt'\
                                          iam_role '" +redshift_role+ "'\
                                          delimiter '|' region '" +region_name+"';"

                      copy_category_query = "copy tickit.category from 's3://" +bucket_name+ "/" +upload_path+ "category_pipe.txt'\
                                            iam_role '" +redshift_role+ "'\
                                            delimiter '|' region '" +region_name+"';"

                      copy_date_query = "copy tickit.date from 's3://" +bucket_name+ "/" +upload_path+ "date2008_pipe.txt'\
                                        iam_role '" +redshift_role+ "'\
                                        delimiter '|' region '" +region_name+"';"

                      copy_event_query = "copy tickit.event from 's3://" +bucket_name+ "/" +upload_path+ "allevents_pipe.txt'\
                                          iam_role '" +redshift_role+ "'\
                                          delimiter '|' timeformat 'YYYY-MM-DD HH:MI:SS' region '" +region_name+"';"

                      copy_listing_query = "copy tickit.listing from 's3://" +bucket_name+ "/" +upload_path+ "listings_pipe.txt'\
                                            iam_role '" +redshift_role+ "'\
                                            delimiter '|' region '" +region_name+"';"

                      copy_sales_query = "copy tickit.sales from 's3://" +bucket_name+ "/" +upload_path+ "sales_tab.txt'\
                                          iam_role '" +redshift_role+ "'\
                                          delimiter '\t' timeformat 'MM/DD/YYYY HH:MI:SS' region '" +region_name+"';"

                      query_array = [create_schema, create_users_query, create_venue_query, create_category_query, create_date_query, create_event_query, create_listing_query, create_sales_query, copy_users_query, copy_venue_query, copy_category_query, copy_date_query, copy_event_query, copy_listing_query, copy_sales_query]

                      for element in range(len(query_array)):
                          query = query_array[element]
                          print('Executing Query: ' +query)
                          
                          execute_sql_response = redshift_client.execute_statement(
                              ClusterIdentifier=redshift_id,
                              Database=redshift_db,
                              DbUser=redshift_user,
                              Sql=query
                          )

                          # Sleep for 5 seconds to get statement status
                          time.sleep(5)

                          statement_status = redshift_client.describe_statement(Id=execute_sql_response['ResponseMetadata']['RequestId'])
                          print('Statement status is: ' +str(statement_status))

                  else:
                    print('Do nothing!')

                  sendResponse(event, context, "SUCCESS")
              except Exception as e:
                  print(e)
                  sendResponse(event, context, "FAILED")

          def sendResponse(event, context, status):
              http = urllib3.PoolManager()
              response_body = {'Status': status,
                              'Reason': 'Log stream name: ' + context.log_stream_name,
                              'PhysicalResourceId': context.log_stream_name,
                              'StackId': event['StackId'],
                              'RequestId': event['RequestId'],
                              'LogicalResourceId': event['LogicalResourceId'],
                              'Data': json.loads("{}")}
              http.request('PUT', event['ResponseURL'], body=json.dumps(response_body))
  DownloadZipFileFunctionExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Statement:
        - Action:
          - sts:AssumeRole
          Effect: Allow
          Principal:
            Service:
            - lambda.amazonaws.com
        Version: '2012-10-17'
      Path: "/"
      Policies:
      - PolicyDocument:
          Statement:
          - Action:
            - logs:CreateLogGroup
            - logs:CreateLogStream
            - logs:PutLogEvents
            Effect: Allow
            Resource: arn:aws:logs:*:*:*
          Version: '2012-10-17'
        PolicyName: !Sub ${AWS::StackName}-${AWS::Region}-CleanupS3Function
      - PolicyDocument:
          Statement:
            - Action:
                - s3:PutObject
                - s3:DeleteObject
                - s3:List*
                - s3:Get*
                - s3:*Version*
                - redshift-data:ExecuteStatement
                - redshift:GetClusterCredentials
              Effect: Allow
              Resource:
                - !Sub arn:aws:s3:::${DataBucket}/*
                - !Sub arn:aws:s3:::${DataBucket}
                - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':cluster:', !Ref RedshiftCluster ]]
                - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':dbuser:', !Ref RedshiftCluster, '/', !FindInMap [ ClusterConfigurations, redshift, userName ]]]
                - !Join [ '', [ 'arn:aws:redshift:', !Ref AWS::Region, ':', !Ref AWS::AccountId, ':dbname:', !Ref RedshiftCluster, '/', !FindInMap [ ClusterConfigurations, redshift, dbName ]]]
            - Action:
                - redshift-data:DescribeStatement
              Effect: Allow
              Resource: '*'
          Version: '2012-10-17'
        PolicyName: !Sub ${AWS::StackName}-${AWS::Region}-DownloadZipFileFunction
  UploadScripts:
    Type: Custom::UploadScripts
    Properties:
      ServiceToken: !GetAtt UploadScriptsFunction.Arn
      uploadS3Bucket: !Ref DataBucket
      scriptPathEmr: !FindInMap [ ClusterConfigurations, emrServerless, scriptPath ]
      scriptNameEmr: !FindInMap [ ClusterConfigurations, emrServerless, scriptName ]
      scriptPathGlue: !FindInMap [ ClusterConfigurations, Glue, scriptPath ]
      scriptNameGlue: !FindInMap [ ClusterConfigurations, Glue, scriptName ]
      redshiftJdbcUrl: !Join [ '', [ 'jdbc:redshift:iam://', !GetAtt RedshiftCluster.Endpoint.Address, ':', !GetAtt RedshiftCluster.Endpoint.Port, '/', !FindInMap [ ClusterConfigurations, redshift, dbName ]]]
      redshiftTempDirectory: !Join [ '', [ 's3://', !Ref DataBucket, '/redshift-temp-dir/' ]]
      redshiftIamRoleArn: !GetAtt RedshiftIamRole.Arn
      emrNotebookPath: !FindInMap [ ClusterConfigurations, emr, notebookPath ]
      emrNotebookName: !FindInMap [ ClusterConfigurations, emr, notebookName ]
  UploadScriptsFunction:
    Type: "AWS::Lambda::Function"
    DependsOn:
      - DataBucket
    Properties:
      Description: "Upload the notebook and script for EMR, EMR Serverless, and Glue to S3"
      Handler: index.lambda_handler
      Role: !GetAtt UploadScriptsFunctionExecutionRole.Arn
      Timeout: 360
      Runtime: python3.9
      Code:
        ZipFile: !Sub |
          import json
          import boto3
          import urllib3

          def lambda_handler(event, context):
              try:
                  bucket_name = event['ResourceProperties']['uploadS3Bucket']
                  script_path_emr = event['ResourceProperties']['scriptPathEmr']
                  script_name_emr = event['ResourceProperties']['scriptNameEmr']
                  script_path_glue = event['ResourceProperties']['scriptPathGlue']
                  script_name_glue = event['ResourceProperties']['scriptNameGlue']
                  redshift_jdbc_url = event['ResourceProperties']['redshiftJdbcUrl']
                  redshift_temp_directory = event['ResourceProperties']['redshiftTempDirectory']
                  redshift_iam_role_arn = event['ResourceProperties']['redshiftIamRoleArn']
                  emr_notebook_path = event['ResourceProperties']['emrNotebookPath']
                  emr_notebook_name = event['ResourceProperties']['emrNotebookName']

                  print(event)
                  if event['RequestType'] == 'Create':
                      s3_client = boto3.client('s3')

                      # redshift-spark-emr.ipynb
                      file_emr_notebook = '{\n "cells": [\n  {\n   "cell_type": "markdown",\n   "id": "e243b60b-9647-41d9-b9d7-48b178682b06",\n   "metadata": {},\n   "source": [\n    "# Using spark-redshift connector with Amazon EMR on Amazon EC2 to execute SQL Statements on Amazon Redshift \\n",\n    "\\n",\n    "## Topics covered in this example\\n",\n    "\\n",\n    "* Configuring the spark-redshift Connector\\n",\n    "* Connecting to Amazon Redshift using spark-redshift connector and issue a query to read data\\n",\n    "* See how the joins, agregations, filters are pushed down to Amazon Redshift\\n",\n    "* Connect to Amazon Redshift using spark-redshift connector and write data\\n",\n    "\\n",\n    "\\n",\n    "## Prerequisites\\n",\n    "\\n",\n    "1. An Amazon EMR cluster with version higher than **emr-6.9.0**\\n",\n    "2. An Amazon Redshift cluster\\n",\n    "3. As a part of this blog the configuration changes below are performed for you already. This is just for your information.\\n",\n    "\\n",\n    "a. The spark-redshift connector and redshift JDBC drivers are locally available within EMR at the the location: `/usr/share/aws/redshift/`\\n",\n    "    ##### The following commands should be run from the EMR cluster attached to the notebook\\n",\n    "\\n",\n    "    a. Open the Livy configuration file\\n",\n    "    \\n",\n    "            vim /etc/livy/conf/livy.conf\\n",\n    "\\n",\n    "\\n",\n    "    b. Add the direvtory /usr/share/aws/redshift/ to whitelist\\n",\n    "    \\n",\n    "           livy.file.local-dir-whitelist /usr/share/aws/redshift/\\n",\n    "\\n",\n    "\\n",\n    "    c. Restart Livy\\n",\n    "    \\n",\n    "           systemctl restart livy-server.service\\n",\n    "\\n",\n    "b. Optionally you can upload these jar files mentioned below to an Amazon S3 bucket and use the path for these jar files to import in the Spark application.\\n",\n    "\\n",\n    "    - RedshiftJDBC.jar\\n",\n    "    - spark-redshift.jar\\n",\n    "    - spark-avro.jar\\n",\n    "    - minimal-json.jar\\n",\n    "\\n",\n    "4. An Amazon S3 bucket to hold the temporary data from Amazon Redshift\\n",\n    "\\n",\n    "## Introduction\\n",\n    "\\n",\n    "This notebook demonstrate how users can make use of spark-redshift connector with Amazon EMR on Amazon EC2 to read and write data from Amazon Redshift using EMR Studio Notebook using PySpark/Python\\n",\n    "\\n",\n    "\\n",\n    "## Define the Apache Spark configuration\\n"\n   ]\n  },\n  {\n   "cell_type": "code",\n   "execution_count": null,\n   "id": "a9daee6c-8fe0-4ff3-bf02-4295fe91041a",\n   "metadata": {\n    "tags": []\n   },\n   "outputs": [],\n   "source": [\n    "%%configure -f\\n",\n    "{\\n",\n    "    \\"conf\\": {\\n",\n    "        \\"spark.jars\\":\\"file:///usr/share/aws/redshift/jdbc/RedshiftJDBC.jar,file:///usr/share/aws/redshift/spark-redshift/lib/spark-redshift.jar,file:///usr/share/aws/redshift/spark-redshift/lib/spark-avro.jar,file:///usr/share/aws/redshift/spark-redshift/lib/minimal-json.jar\\",\\n",\n    "        \\"spark.pyspark.python\\": \\\"python3\\\",\\n",\n    "        \\"spark.pyspark.virtualenv.enabled\\": \\"true\\",\\n",\n    "        \\"spark.pyspark.virtualenv.type\\":\\"native\\",\\n",\n    "        \\"spark.pyspark.virtualenv.bin.path\\":\\"/usr/bin/virtualenv\\"\\n",\n    "    }\\n",\n    "}"\n   ]\n  },\n  {\n   "cell_type": "markdown",\n   "id": "8c5fd8f3-7534-42fa-9482-e2a20a3477f4",\n   "metadata": {},\n   "source": [\n    "## Import spark modules and initiate SparkSession"\n   ]\n  },\n  {\n   "cell_type": "code",\n   "execution_count": null,\n   "id": "3bf02797-ffd5-4337-a844-b34c2ecc6456",\n   "metadata": {},\n   "outputs": [],\n   "source": [\n    "from pyspark.sql import SparkSession\\n",\n    "from pyspark.sql.functions import col\\n",\n    "spark = SparkSession \\\\\\n",\n    "        .builder \\\\\\n",\n    "        .appName(\\"SparkRedshiftConnector\\") \\\\\\n",\n    "        .enableHiveSupport() \\\\\\n",\n    "        .getOrCreate()"\n   ]\n  },\n  {\n   "cell_type": "markdown",\n   "id": "8b0818d5-1854-4a5b-97a4-e8729b37fb9b",\n   "metadata": {},\n   "source": [\n    "## Define the variables to connect to Amazon Redshift cluster\\n",\n    "\\n",\n    "**You can find the values for these variables in the Amazon CloudFormation output**\\n",\n    "\\n",\n    "1. You can authenticate to Amazon Redshift using AWS IAM credentials or using username and password. In this example we are using **AWS IAM credentials**. Provide the JDBC URL for Amazon Redshift - value for **RedshiftClusterJdbcUrl**\\n",\n    "2. Provide the Amazon S3 location that acts as temporary directory - value for **RedshiftS3TempPath**\\n",\n    "3. Provide an AWS IAM role that Amazon Redshift can use to connect to Amazon S3 and Amazon Redshift - **RedshiftIamRoleArn**\\n",\n    "\\n",\n    "**Make sure you replace the values for `jdbc_iam_url`, `temp_dir`, and `aws_role` with the approriate values from the CloudFormation stack output**"\n   ]\n  },\n  {\n   "cell_type": "code",\n   "execution_count": null,\n   "id": "6acdb577-be7a-4c62-8020-1f735177e5bd",\n   "metadata": {\n    "tags": []\n   },\n   "outputs": [],\n   "source": [\n    "jdbc_iam_url = \'' +redshift_jdbc_url+ '\'\\n",\n    "temp_dir = \'' +redshift_temp_directory+ '\'\\n",\n    "aws_role = \'' +redshift_iam_role_arn+ '\'\\n",\n    "\\n",\n    "# Set query group for the query\\n",\n    "queryGroup = \'spark-redshift-emr\'\\n",\n    "jdbc_iam_url_withQueryGroup = jdbc_iam_url+\'?queryGroup=\'+queryGroup\\n",\n    "\\n",\n    "# Set User name for the query\\n",\n    "userName = \'redshiftmasteruser\'\\n",\n    "jdbc_iam_url_withUserName = jdbc_iam_url_withQueryGroup+\';user=\'+userName\\n",\n    "\\n",\n    "\\n",\n    "# Define the Redshift context\\n",\n    "redshiftOptions = {\\n",\n    "    \'url\': jdbc_iam_url_withUserName,\\n",\n    "    \'tempdir\': temp_dir,\\n",\n    "    \'aws_iam_role\' : aws_role\\n",\n    "}"\n   ]\n  },\n  {\n   "cell_type": "markdown",\n   "id": "4f478cf0-a3f4-4c6d-8efb-d5b377cc85a6",\n   "metadata": {},\n   "source": [\n    "## Create spark dataframes\\n",\n    "\\n",\n    "1. Create dataframes for the tables `sales` and `date` in `tickit` schema\\n",\n    "2. Perform inner join between `sales_df` and `date_df` using `dateid` column\\n",\n    "3. Apply a filter on `date` table for year **2008**\\n",\n    "4. Aggregate `qtysold` and group by `quarter`\\n",\n    "5. Sort the output by `quarter`\\n",\n    "6. Lastly rename the column `qtysold` to `total_quantity_sold`"\n   ]\n  },\n  {\n   "cell_type": "code",\n   "execution_count": null,\n   "id": "951c4d32-b662-462a-aa4b-de3369491404",\n   "metadata": {\n    "tags": []\n   },\n   "outputs": [],\n   "source": [\n    "# Create the sales DataFrame from Amazon Redshift table using io.github.spark_redshift_community.spark.redshift class\\n",\n    "sales_df = (\\n",\n    "    spark.read\\n",\n    "        .format(\'io.github.spark_redshift_community.spark.redshift\')\\n",\n    "        .options(**redshiftOptions)\\n",\n    "        .option(\'dbtable\', \'tickit.sales\')\\n",\n    "        .load()\\n",\n    ")\\n",\n    "\\n",\n    "# Create the date Data Frame from Amazon Redshift table\\n",\n    "date_df = (\\n",\n    "    spark.read\\n",\n    "        .format(\'io.github.spark_redshift_community.spark.redshift\')\\n",\n    "        .options(**redshiftOptions)\\n",\n    "        .option(\'dbtable\', \'tickit.date\')\\n",\n    "        .load()\\n",\n    ")\\n",\n    "\\n",\n    "# Assign a Data Frame to the above output which will be written back to Amazon Redshift\\n",\n    "output_df= sales_df.join(date_df, sales_df.dateid == date_df.dateid, \'inner\').where(\\n",\n    "    col(\'year\') == 2008).groupBy(\'qtr\').sum(\'qtysold\').select(\\n",\n    "        col(\'qtr\'), col(\'sum(qtysold)\')).sort([\'qtr\'], ascending=[1]).withColumnRenamed(\'sum(qtysold)\',\'total_quantity_sold\')\\n",\n    "\\n",\n    "# Display the output\\n",\n    "output_df.show()"\n   ]\n  },\n  {\n   "cell_type": "markdown",\n   "id": "65cb4dec-a44c-417f-8365-66fb60682831",\n   "metadata": {},\n   "source": [\n    "## Get the last query executed on Amazon Redshift with query_label = \'spark-redshift-emr\'\\n",\n    "\\n",\n    "We can see which operations are pushed down to Amazon Redshift by looking at the query issues by Spark on Amazon Redshift. Notice that the the spark-connector pushes down filter, aggregation, sort, join, order by to Amazon Redshift by issuing an appropriate query"\n   ]\n  },\n  {\n   "cell_type": "code",\n   "execution_count": null,\n   "id": "5af42a08-0796-4d53-b9b8-825fb290604a",\n   "metadata": {\n    "tags": []\n   },\n   "outputs": [],\n   "source": [\n    "## Lets drop the queryGroup for easy validation of push down queries\\n",\n    "# Set User name for the query\\n",\n    "jdbc_iam_url_withUserName = jdbc_iam_url+\'?user=\'+userName\\n",\n    "\\n",\n    "# Define the Redshift context\\n",\n    "redshiftOptions = {\\n",\n    "    \'url\': jdbc_iam_url_withUserName,\\n",\n    "    \'tempdir\': temp_dir,\\n",\n    "    \'aws_iam_role\' : aws_role\\n",\n    "}\\n",\n    "\\n",\n    "test_df = (\\n",\n    "    spark.read\\n",\n    "        .format(\\\"io.github.spark_redshift_community.spark.redshift\\\")\\n",\n    "        .options(**redshiftOptions)\\n",\n    "        .option(\\"query\\", \\\"SELECT query_text FROM SYS_QUERY_HISTORY WHERE query_label = \'spark-redshift-emr\' ORDER BY start_time DESC LIMIT 1\\\")\\n",\n    "        .load()\\n",\n    ")\\n",\n    "\\n",\n    "def GetValueFromDataframe(_df,columnName):\\n",\n    "    for row in _df.rdd.collect():       \\n",\n    "        return row[columnName].strip()\\n",\n    "    \\n",\n    "print( GetValueFromDataframe(test_df,\'query_text\'))"\n   ]\n  },\n  {\n   "cell_type": "markdown",\n   "id": "d1f870c7-7e2b-411d-bca6-6f72812d1d61",\n   "metadata": {},\n   "source": [\n    "## Write data to Amazon Redshift\\n",\n    "\\n",\n    "We will now write the results from the previous aggregation we performed back to Amazon Redshift. The results will be stored in `tickit.test_emr` table."\n   ]\n  },\n  {\n   "cell_type": "code",\n   "execution_count": null,\n   "id": "b18ca8e8-b6f2-419c-a79a-2c9dab0fcc87",\n   "metadata": {\n    "tags": []\n   },\n   "outputs": [],\n   "source": [\n    "# Define the Amazon Redshift context\\n",\n    "redshiftWriteOptions = {\\n",\n    "    \'url\': jdbc_iam_url_withUserName,\\n",\n    "    \'tempdir\': temp_dir,\\n",\n    "    \'aws_iam_role\' : aws_role\\n",\n    "}\\n",\n    "\\n",\n    "# Write the Data Frame back to Amazon Redshift\\n",\n    "output_df.write \\\\\\n",\n    "    .format(\'io.github.spark_redshift_community.spark.redshift\') \\\\\\n",\n    "    .mode(\'overwrite\') \\\\\\n",\n    "    .options(**redshiftWriteOptions) \\\\\\n",\n    "    .option(\'dbtable\', \'tickit.test_emr\') \\\\\\n",\n    "    .save()"\n   ]\n  }\n ],\n "metadata": {\n  "kernelspec": {\n   "display_name": "PySpark",\n   "language": "python",\n   "name": "pysparkkernel"\n  },\n  "language_info": {\n   "codemirror_mode": {\n    "name": "python",\n    "version": 3\n   },\n   "file_extension": ".py",\n   "mimetype": "text/x-python",\n   "name": "pyspark",\n   "pygments_lexer": "python3"\n  }\n },\n "nbformat": 4,\n "nbformat_minor": 5\n}'
                      file_notebook_byte = file_emr_notebook.encode()
                      response_file_emr_notebook = s3_client.put_object(
                                                        Bucket=bucket_name,
                                                        Key=emr_notebook_path+'/'+emr_notebook_name,
                                                        Body=file_notebook_byte)

                      # redshift-spark-emr.py
                      file_emr_string = '#\n# Copyright 2021 Amazon.com, Inc. or its affiliates. All Rights Reserved.\n#\n# Licensed under the Amazon Software License (the "License"). You may not use this file except in compliance\n# with the License. A copy of the License is located at\n#\n#      http://aws.amazon.com/asl/\n#\n# or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES\n# OR CONDITIONS OF ANY KIND, express or implied. See the License for the specific language governing permissions\n# and limitations under the License.\n#\n\nimport sys\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql.functions import col\n\nif __name__ == "__main__":\n\n    spark = SparkSession \\\n            .builder \\\n            .appName("SparkRedshiftConnector") \\\n            .enableHiveSupport() \\\n            .getOrCreate()\n    \n    jdbc_iam_url = \'' +redshift_jdbc_url+ '\'\n    temp_dir = \'' +redshift_temp_directory+ '\'\n    aws_role = \'' +redshift_iam_role_arn+ '\'\n    \n    # Set query group for the query\n    queryGroup = "emr-serverless-redshift"\n    jdbc_iam_url_withQueryGroup = jdbc_iam_url+\'?queryGroup=\'+queryGroup\n    \n    # Set User name for the query\n    userName = \'redshiftmasteruser\'\n    jdbc_iam_url_withUserName = jdbc_iam_url_withQueryGroup+\';user=\'+userName\n    \n    # Define the Redshift context\n    redshiftOptions = {\n        "url": jdbc_iam_url_withUserName,\n        "tempdir": temp_dir,\n        "aws_iam_role" : aws_role\n    };\n    \n    # Create the sales DataFrame from Amazon Redshift table using io.github.spark_redshift_community.spark.redshift class\n    sales_df = (\n        spark.read\n            .format("io.github.spark_redshift_community.spark.redshift")\n            .options(**redshiftOptions)\n            .option("dbtable", "tickit.sales")\n            .load()\n    )\n    \n    # Create the date Data Frame from Amazon Redshift table\n    date_df = (\n        spark.read\n            .format("io.github.spark_redshift_community.spark.redshift")\n            .options(**redshiftOptions)\n            .option("dbtable", "tickit.date")\n            .load()\n    )\n    \n    # Assign a Data Frame to the above output which will be written back to Amazon Redshift\n    output_df= sales_df.join(date_df, sales_df.dateid == date_df.dateid, \'inner\').where(\n        col("year") == 2008).groupBy("qtr").sum("qtysold").select(\n            col("qtr"), col("sum(qtysold)")).sort(["qtr"], ascending=[1]).withColumnRenamed("sum(qtysold)","total_quantity_sold")\n    \n    # Display the output\n    output_df.show()\n    \n    ## Lets drop the queryGroup for easy validation of push down queries\n    # Set User name for the query\n    userName = \'redshiftmasteruser\'\n    jdbc_iam_url_withUserName = jdbc_iam_url+\'?user=\'+userName\n    \n    # Define the Amazon Redshift context\n    redshiftWriteOptions = {\n        "url": jdbc_iam_url_withUserName,\n        "tempdir": temp_dir,\n        "aws_iam_role" : aws_role\n    }\n    \n    # Write the Data Frame back to Amazon Redshift\n    output_df.write \\\n        .format("io.github.spark_redshift_community.spark.redshift") \\\n        .mode("overwrite") \\\n        .options(**redshiftWriteOptions) \\\n        .option("dbtable", "tickit.test-emrserverless") \\\n        .save()\n    \n    spark.stop()'
                      file_emr_byte = file_emr_string.encode()
                      response_file_emr = s3_client.put_object(
                                                  Bucket=bucket_name,
                                                  Key=script_path_emr+script_name_emr,
                                                  Body=file_emr_byte)

                      # redshift-spark-glue.py
                      file_glue = b'import sys\nfrom awsglue.transforms import *\nfrom awsglue.utils import getResolvedOptions\nfrom pyspark.context import SparkContext\nfrom awsglue.context import GlueContext\nfrom awsglue.job import Job\nfrom pyspark.sql.functions import col\n\n## @params: [JOB_NAME]\nargs = getResolvedOptions(\n    sys.argv,\n    [\n        \'JOB_NAME\',\n        \'REDSHIFT_JDBC_URL\',\n        \'REDSHIFT_TEMPORARY_DIRECTORY\',\n        \'AWS_IAM_ROLE_ARN\'\n    ]\n)\n\nsc = SparkContext()\nglueContext = GlueContext(sc)\nspark = glueContext.spark_session\njob = Job(glueContext)\njob.init(args[\'JOB_NAME\'], args)\n\njdbc_iam_url = args[\'REDSHIFT_JDBC_URL\']\ntemp_dir = args[\'REDSHIFT_TEMPORARY_DIRECTORY\']\naws_role = args[\'AWS_IAM_ROLE_ARN\']\n\n# Set query group for the query\nqueryGroup = \'glue-redshift\'\njdbc_iam_url_withQueryGroup = jdbc_iam_url+\'?queryGroup=\'+queryGroup\n\n# Set User name for the query\nuserName = \'redshiftmasteruser\'\njdbc_iam_url_withUserName = jdbc_iam_url_withQueryGroup+\';user=\'+userName\n\n# Define the Redshift context\nredshiftOptions = {\n    \'url\': jdbc_iam_url_withUserName,\n    \'tempdir\': temp_dir,\n    \'aws_iam_role\': aws_role\n}\n\n# Create the sales DataFrame from Redshift table\nsales_df = (\n    spark.read\n        .format(\'io.github.spark_redshift_community.spark.redshift\')\n        .options(**redshiftOptions)\n        .option(\'dbtable\', \'tickit.sales\')\n        .load()\n)\n\n# Create the date DataFrame from Redshift table\ndate_df = (\n    spark.read\n        .format(\'io.github.spark_redshift_community.spark.redshift\')\n        .options(**redshiftOptions)\n        .option(\'dbtable\', \'tickit.date\')\n        .load()\n)\n\n# Assign a DataFrame to the above output which will be written back to Redshift\noutput_df= sales_df.join(date_df, sales_df.dateid == date_df.dateid, \'inner\').where(\n    col(\'year\') == 2008).groupBy(\'qtr\').sum(\'qtysold\').select(\n        col(\'qtr\'), col(\'sum(qtysold)\')).sort([\'qtr\'], ascending=[1]).withColumnRenamed(\'sum(qtysold)\',\'total_quantity_sold\')\n\n# Issue query to redshift and show results\noutput_df.show()\n\n## Lets drop the queryGroup for easy validation of push down queries\n\n# Set User name for the query\njdbc_iam_url_withUserName = jdbc_iam_url+\'?user=\'+userName\n\n# Define the Redshift context\nredshiftWriteOptions = {\n    \'url\': jdbc_iam_url_withUserName,\n    \'tempdir\': temp_dir,\n    \'aws_iam_role\': aws_role\n}\n\n# Write the Data Frame back to Redshift\noutput_df.write \\\n    .format(\'io.github.spark_redshift_community.spark.redshift\') \\\n    .mode(\'overwrite\') \\\n    .options(**redshiftWriteOptions) \\\n    .option(\'dbtable\', \'tickit.test-glue\') \\\n    .save()\n\njob.commit()'
                      response_file_glue = s3_client.put_object(
                                                  Bucket=bucket_name,
                                                  Key=script_path_glue+script_name_glue,
                                                  Body=file_glue)
                  else:
                    print('Do nothing!')

                  sendResponse(event, context, "SUCCESS")
              except Exception as e:
                  print(e)
                  sendResponse(event, context, "FAILED")

          def sendResponse(event, context, status):
              http = urllib3.PoolManager()
              response_body = {'Status': status,
                              'Reason': 'Log stream name: ' + context.log_stream_name,
                              'PhysicalResourceId': context.log_stream_name,
                              'StackId': event['StackId'],
                              'RequestId': event['RequestId'],
                              'LogicalResourceId': event['LogicalResourceId'],
                              'Data': json.loads("{}")}
              http.request('PUT', event['ResponseURL'], body=json.dumps(response_body))
  UploadScriptsFunctionExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Statement:
        - Action:
          - sts:AssumeRole
          Effect: Allow
          Principal:
            Service:
            - lambda.amazonaws.com
        Version: '2012-10-17'
      Path: "/"
      Policies:
      - PolicyName: !Sub ${AWS::StackName}-${AWS::Region}-UploadScriptsFunction
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
            - Action:
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
              Effect: Allow
              Resource: arn:aws:logs:*:*:*
            - Action:
                - s3:PutObject
                - s3:DeleteObject
                - s3:List*
                - s3:Get*
                - s3:*Version*
              Effect: Allow
              Resource:
                - !Sub arn:aws:s3:::${DataBucket}/*
                - !Sub arn:aws:s3:::${DataBucket}
  CleanupS3Resource:
    Type: Custom::CleanupS3Resource
    DependsOn:
      - EMRStudio
    Properties:
      ServiceToken: !GetAtt CleanupS3Function.Arn
      inputBucketName: !Ref DataBucket

  CleanupS3Function:
    Type: "AWS::Lambda::Function"
    DependsOn:
      - DataBucket
      - EMRStudio
    Properties:
      Description: "Work with S3 Buckets!"
      Handler: index.lambda_handler
      Role: !GetAtt CleanupS3FunctionExecutionRole.Arn
      Timeout: 360
      Runtime: python3.9
      Code:
        ZipFile: |
          import json
          import boto3
          import urllib3

          def empty_bucket(bucket_name):
              print("Attempting to empty the bucket {0}".format(bucket_name))
              s3_client = boto3.client('s3')
              s3 = boto3.resource('s3')

              try:
                  bucket = s3.Bucket(bucket_name).load()
              except ClientError:
                  print("Bucket {0} does not exist".format(bucket_name))
                  return
              # Confirm if versioning is enabled
              version_status = s3_client.get_bucket_versioning(Bucket=bucket_name)
              status = version_status.get('Status','')
              if status == 'Enabled':
                  version_status = s3_client.put_bucket_versioning(Bucket=bucket_name,
                                                            VersioningConfiguration={'Status': 'Suspended'})
              version_paginator = s3_client.get_paginator('list_object_versions')
              version_iterator = version_paginator.paginate(
                  Bucket=bucket_name
              )

              for page in version_iterator:
                  print(page)
                  if 'DeleteMarkers' in page:
                      delete_markers = page['DeleteMarkers']
                      if delete_markers is not None:
                          for delete_marker in delete_markers:
                              key = delete_marker['Key']
                              versionId = delete_marker['VersionId']
                              s3_client.delete_object(Bucket=bucket_name, Key=key, VersionId=versionId)
                  if 'Versions' in page and page['Versions'] is not None:
                      versions = page['Versions']
                      for version in versions:
                          print(version)
                          key = version['Key']
                          versionId = version['VersionId']
                          s3_client.delete_object(Bucket=bucket_name, Key=key, VersionId=versionId)
              object_paginator = s3_client.get_paginator('list_objects_v2')
              object_iterator = object_paginator.paginate(
                  Bucket=bucket_name
              )
              for page in object_iterator:
                  if 'Contents' in page:
                      for content in page['Contents']:
                          key = content['Key']
                          s3_client.delete_object(Bucket=bucket_name, Key=content['Key'])
              print("Successfully emptied the bucket {0}".format(bucket_name))

          def lambda_handler(event, context):
              try:
                  bucket = event['ResourceProperties']['inputBucketName']
                  if event['RequestType'] == 'Delete':
                      empty_bucket(bucket)
                  sendResponse(event, context, "SUCCESS")
              except Exception as e:
                  print(e)
                  sendResponse(event, context, "FAILED")

          def sendResponse(event, context, status):
              http = urllib3.PoolManager()
              response_body = {'Status': status,
                              'Reason': 'Log stream name: ' + context.log_stream_name,
                              'PhysicalResourceId': context.log_stream_name,
                              'StackId': event['StackId'],
                              'RequestId': event['RequestId'],
                              'LogicalResourceId': event['LogicalResourceId'],
                              'Data': json.loads("{}")}
              http.request('PUT', event['ResponseURL'], body=json.dumps(response_body))

  CleanupS3FunctionExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Statement:
        - Action:
          - sts:AssumeRole
          Effect: Allow
          Principal:
            Service:
            - lambda.amazonaws.com
        Version: '2012-10-17'
      Path: "/"
      Policies:
      - PolicyDocument:
          Statement:
          - Action:
            - logs:CreateLogGroup
            - logs:CreateLogStream
            - logs:PutLogEvents
            Effect: Allow
            Resource: arn:aws:logs:*:*:*
          Version: '2012-10-17'
        PolicyName: !Sub ${AWS::StackName}-${AWS::Region}-CleanupS3Function
      - PolicyDocument:
          Statement:
          - Action:
            - s3:PutObject
            - s3:DeleteObject
            - s3:List*
            - s3:Get*
            - s3:*Version*
            Effect: Allow
            Resource:
            - !Sub arn:aws:s3:::${DataBucket}/*
            - !Sub arn:aws:s3:::${DataBucket}
          Version: '2012-10-17'
        PolicyName: !Sub ${AWS::StackName}-${AWS::Region}-CleanupS3Function-S3

Outputs:
  EMRStudioURL:
    Description: Access URL for EMR Studio
    Value: !GetAtt EMRStudio.Url
  RedshiftClusterJdbcUrl:
    Description: JDBC URL for EMR Redshift Cluster
    Value: !Join [ '', [ 'jdbc:redshift:iam://', !GetAtt RedshiftCluster.Endpoint.Address, ':', !GetAtt RedshiftCluster.Endpoint.Port, '/', !FindInMap [ ClusterConfigurations, redshift, dbName ]]]
  RedshiftUserName:
    Description: Username for Redshift cluster
    Value: !FindInMap [ ClusterConfigurations, redshift, userName ]
  RedshiftIamRoleArn:
    Description: AWS IAM role ARN for Redshift
    Value: !GetAtt RedshiftIamRole.Arn
  RedshiftS3TempPath:
    Description: Temporary path used by Redshift to store data
    Value: !Join [ '', [ 's3://', !Ref DataBucket, '/redshift-temp-dir/' ]]
  GlueJob:
    Description: AWS Glue job
    Value: !Join [ '', [ 'https://console.aws.amazon.com/gluestudio/home?region=', !Ref AWS::Region, '#/editor/job/', !Ref GlueJob ]]
  EMRServerlessExecutionScript:
    Description: Execution script to be executed with EMR Serverless
    Value: !Join [ '', [ 's3://', !Ref DataBucket, '/', !FindInMap [ ClusterConfigurations, emrServerless, scriptPath ], !FindInMap [ ClusterConfigurations, emrServerless, scriptName ]]]
  EMRServerlessJobExecutionRole:
    Description: AWS IAM role ARN for EMR Serverless job
    Value: !GetAtt EMRServerlessJobRole.Arn
  EMRStudioNotebook:
    Description: EMR Studio Notebook download URL
    Value: !Join [ '', [ 'https://console.aws.amazon.com/s3/object/', !Ref DataBucket, '?prefix=', !FindInMap [ ClusterConfigurations, emr, notebookPath ], '/', !FindInMap [ ClusterConfigurations, emr, notebookName ]]]