template.yml

---
AWSTemplateFormatVersion: '2010-09-09'
Description: 'VPC: public and private subnets in two availability zones'

Parameters:
  RedshiftDBName:
    Type: String
    Description: Redshift database name
    Default: ratings 
  SecretsManagerSecret:
    Type: String
    Description: Name of the Secrets Manager secret for the cluster
    Default: reviewssecret
  ArtifactsBucket:
    Type: String
    Description: Bucket with source data and PySpark scripts.
    Default: ala-artifacts


Resources:

  # VPC environment
  VPC:
    Type: 'AWS::EC2::VPC'
    Properties:
      CidrBlock: '10.71.0.0/16'
      EnableDnsSupport: true
      EnableDnsHostnames: true
      InstanceTenancy: default
      Tags:
      - Key: Name
        Value: '10.71.0.0/16'
  InternetGateway:
    Type: 'AWS::EC2::InternetGateway'
    Properties:
      Tags:
      - Key: Name
        Value: '10.71.0.0/16'
  VPCGatewayAttachment:
    Type: 'AWS::EC2::VPCGatewayAttachment'
    Properties:
      VpcId: !Ref VPC
      InternetGatewayId: !Ref InternetGateway
  SubnetAPublic:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [0, !GetAZs '']
      CidrBlock: '10.71.0.0/20'
      MapPublicIpOnLaunch: true
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'A public'
      - Key: Reach
        Value: public
  SubnetAPrivate:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [0, !GetAZs '']
      CidrBlock: '10.71.16.0/20'
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'A private'
      - Key: Reach
        Value: private
  SubnetBPublic:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [1, !GetAZs '']
      CidrBlock: '10.71.32.0/20'
      MapPublicIpOnLaunch: true
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'B public'
      - Key: Reach
        Value: public
  SubnetBPrivate:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [1, !GetAZs '']
      CidrBlock: '10.71.48.0/20'
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'B private'
      - Key: Reach
        Value: private
  RouteTablePublic: # should be RouteTableAPublic, but logical id was not changed for backward compatibility
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'A Public'
  RouteTablePrivate: # should be RouteTableAPrivate, but logical id was not changed for backward compatibility
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'A Private'
  RouteTableBPublic:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'B Public'
  RouteTableBPrivate:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'B Private'
  RouteTableAssociationAPublic:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref SubnetAPublic
      RouteTableId: !Ref RouteTablePublic
  RouteTableAssociationAPrivate:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref SubnetAPrivate
      RouteTableId: !Ref RouteTablePrivate
  RouteTableAssociationBPublic:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref SubnetBPublic
      RouteTableId: !Ref RouteTableBPublic
  RouteTableAssociationBPrivate:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref SubnetBPrivate
      RouteTableId: !Ref RouteTableBPrivate
  RouteTablePublicInternetRoute: # should be RouteTablePublicAInternetRoute, but logical id was not changed for backward compatibility
    Type: 'AWS::EC2::Route'
    DependsOn: VPCGatewayAttachment
    Properties:
      RouteTableId: !Ref RouteTablePublic
      DestinationCidrBlock: '0.0.0.0/0'
      GatewayId: !Ref InternetGateway
  RouteTablePublicBInternetRoute:
    Type: 'AWS::EC2::Route'
    DependsOn: VPCGatewayAttachment
    Properties:
      RouteTableId: !Ref RouteTableBPublic
      DestinationCidrBlock: '0.0.0.0/0'
      GatewayId: !Ref InternetGateway
  NetworkAclPublic:
    Type: 'AWS::EC2::NetworkAcl'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: Public
  NetworkAclPrivate:
    Type: 'AWS::EC2::NetworkAcl'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: Private
  SubnetNetworkAclAssociationAPublic:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref SubnetAPublic
      NetworkAclId: !Ref NetworkAclPublic
  SubnetNetworkAclAssociationAPrivate:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref SubnetAPrivate
      NetworkAclId: !Ref NetworkAclPrivate
  SubnetNetworkAclAssociationBPublic:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref SubnetBPublic
      NetworkAclId: !Ref NetworkAclPublic
  SubnetNetworkAclAssociationBPrivate:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref SubnetBPrivate
      NetworkAclId: !Ref NetworkAclPrivate
  NetworkAclEntryInPublicAllowAll:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPublic
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: false
      CidrBlock: '0.0.0.0/0'
  NetworkAclEntryOutPublicAllowAll:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPublic
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: true
      CidrBlock: '0.0.0.0/0'
  NetworkAclEntryInPrivateAllowVPC:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPrivate
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: false
      CidrBlock: '0.0.0.0/0'
  NetworkAclEntryOutPrivateAllowVPC:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPrivate
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: true
      CidrBlock: '0.0.0.0/0'
  NAT:
    DependsOn: VPCGatewayAttachment
    Type: AWS::EC2::NatGateway
    Properties:
      AllocationId:
        Fn::GetAtt:
        - EIP
        - AllocationId
      SubnetId:
        Ref: SubnetAPublic
      Tags:
        - Key: foo
          Value: bar
  EIP:
    Type: AWS::EC2::EIP
    Properties:
      Domain: !Ref VPC
  RouteANAT:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId:
        Ref: RouteTablePrivate
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId:
        Ref: NAT
  RouteBNAT:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId:
        Ref: RouteTableBPrivate
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId:
        Ref: NAT


# Redshift Cluster

  ScriptBucket:
    Type: AWS::S3::Bucket
    DeletionPolicy: Delete
    
  DataBucket:
    Type: AWS::S3::Bucket
    DeletionPolicy: Delete  

  RedshiftSG:
    Properties:
      GroupDescription: Redshift security group
      SecurityGroupEgress:
      - CidrIp: 0.0.0.0/0
        IpProtocol: '-1'
      VpcId:
        Ref: VPC
    Type: AWS::EC2::SecurityGroup
    
  InboundRule:
    Properties:
      FromPort: 0
      GroupId:
        Fn::GetAtt:
        - RedshiftSG
        - GroupId
      IpProtocol: '-1'
      SourceSecurityGroupId:
        Fn::GetAtt:
        - RedshiftSG
        - GroupId
      ToPort: 65535
    Type: AWS::EC2::SecurityGroupIngress

  RedshiftSubnetGroup:
    Properties:
      Description: Subnet Group for redshift
      SubnetIds: [!Ref SubnetAPrivate, !Ref SubnetBPrivate]
    Type: AWS::Redshift::ClusterSubnetGroup
    
  RedshiftRole:
    Properties:
      AssumeRolePolicyDocument:
        Statement:
        - Action: sts:AssumeRole
          Effect: Allow
          Principal:
            Service: redshift.amazonaws.com
          Sid: ''
        Version: 2012-10-17
    Type: AWS::IAM::Role
    
    
  RedshiftSecret:
    Type: "AWS::SecretsManager::Secret"
    Properties:
      Name: 'reviewssecret'
      Description: "This is a Secrets Manager secret for Redshift"
      GenerateSecretString:
        SecretStringTemplate: '{"username": "master"}'
        GenerateStringKey: "password"
        PasswordLength: 16
        ExcludeCharacters: '"@/\'
    

  RedshiftCluster:
    DependsOn: RedshiftSecret
    Properties:
      ClusterSubnetGroupName:
        Ref: RedshiftSubnetGroup
      ClusterType: single-node
      DBName: ratings
      PubliclyAccessible: false
      IamRoles:
      - Fn::GetAtt:
        - RedshiftRole
        - Arn
      MasterUserPassword: !Join ['', ['{{resolve:secretsmanager:', !Ref RedshiftSecret, ':SecretString:password}}' ]]
      MasterUsername: !Join ['', ['{{resolve:secretsmanager:', !Ref RedshiftSecret, ':SecretString:username}}' ]]
      NodeType: dc2.large
      VpcSecurityGroupIds:
      - Ref: RedshiftSG
    Type: AWS::Redshift::Cluster
  
  RedshiftPolicy:
    Properties:
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Action:
          - s3:GetBucketLocation
          - s3:GetObject
          - s3:ListBucket
          Effect: Allow
          Resource:
          - arn:aws:s3:::amazon-reviews-pds
          - arn:aws:s3:::amazon-reviews-pds/*
        - Action:
          - glue:CreateDatabase
          - glue:DeleteDatabase
          - glue:GetDatabase
          - glue:GetDatabases
          - glue:UpdateDatabase
          - glue:CreateTable
          - glue:DeleteTable
          - glue:BatchDeleteTable
          - glue:UpdateTable
          - glue:GetTable
          - glue:GetTables
          - glue:BatchCreatePartition
          - glue:CreatePartition
          - glue:DeletePartition
          - glue:BatchDeletePartition
          - glue:UpdatePartition
          - glue:GetPartition
          - glue:GetPartitions
          - glue:BatchGetPartition
          Effect: Allow
          Resource:
          - '*'
        - Action:
          - s3:GetBucketLocation
          - s3:GetObject
          - s3:ListBucket
          - s3:PutObject
          Effect: Allow
          Resource: 
          - !Join ['', ['arn:aws:s3:::', !Ref DataBucket ]]
          - !Join ['', ['arn:aws:s3:::', !Ref DataBucket, '/*']]
        
      PolicyName: redshift_demo_policy
      Roles:
      - Ref: RedshiftRole
    Type: AWS::IAM::Policy    

# Glue Role and Policy

  GlueExecutionRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - glue.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: /
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
  
  GluePolicy:
    Properties:
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Action:
          - s3:GetBucketLocation
          - s3:GetObject
          - s3:ListBucket
          - s3:HeadObject
          Effect: Allow
          Resource:
          - !Join ['', ['arn:aws:s3:::', !Ref ScriptBucket] ]
          - !Join ['', ['arn:aws:s3:::', !Ref ScriptBucket, '/*'] ]
          - !Join ['', ['arn:aws:s3:::', !Ref DataBucket] ]
          - !Join ['', ['arn:aws:s3:::', !Ref DataBucket, '/*'] ]
        - Action:
          - secretsmanager:*
          Effect: Allow
          Resource: !Ref RedshiftSecret
      PolicyName: glue_demo_policy
      Roles:
      - Ref: GlueExecutionRole
    Type: AWS::IAM::Policy

# Lambda Function and Role
  
  SetupFunction:
    Type: AWS::Lambda::Function
    Properties:
      Role: !GetAtt 'LambdaExecutionRole.Arn'
      FunctionName: !Join ['-', ['copy_scripts', !Ref 'AWS::StackName']]
      MemorySize: 2048
      Runtime: python2.7
      Timeout: 900
      Handler: index.handler
      Code:
        ZipFile: 
          Fn::Sub:
          - |-
           import json
           import boto3
           import os
           import logging
           import cfnresponse

           LOGGER = logging.getLogger()
           LOGGER.setLevel(logging.INFO)

           SCRIPT_SOURCE_BUCKET = '${ArtifactsBucket}'
           SCRIPT_SOURCE_FOLDER = 'artifacts/redshift-based-etl'
           DEST_BUCKET= '${ScriptBucket}'
           UNLOAD_DEST_BUCKET = '${DataBucket}'
           ROLE_ARN = '${Role}'
           CLUSTER_ENDPOINT = '${Endpoint}'
           CLUSTER_ID = CLUSTER_ENDPOINT.split('.')[0]
           ENGINE = 'redshift'
           PORT = '5439'
           SECRET = '${Secret}'

           def handler(event, context):

               # Get CloudFormation-specific parameters, if they exist
               cfn_stack_id = event.get('StackId')
               cfn_request_type = event.get('RequestType')

               
               # Was the function triggered by a CloudFormation resource deletion?
               # If so, delete all objects in dest bucket and return
               if cfn_stack_id and cfn_request_type == 'Delete':
                   s3 = boto3.resource('s3')

                   try:
                       bucket = s3.Bucket(DEST_BUCKET)
                       bucket.objects.delete()
                       bucket = s3.Bucket(UNLOAD_DEST_BUCKET)
                       bucket.objects.delete()
                       message = 'Deleted data'
                   except botocore.exceptions.ClientError as e:
                       # If a client error is thrown, then check that it was a 404 error.
                       # If it was a 404 error, then the bucket does not exist.
                       error_code = int(e.response['Error']['Code'])
                       if error_code == 404:
                           message = 'bucket does not exist.'
                    
                   cfnresponse.send(event, context, cfnresponse.SUCCESS, {
                           'Message': message
                            },
                            context.log_stream_name)

                   return {
                        'statusCode': 200,
                        'body': json.dumps(message)
                    }
                    
               s3 = boto3.client('s3') 
               #copy sql files with iam role and bucket replace
               prefix = SCRIPT_SOURCE_FOLDER + '/SQL'
               sqlfiles = s3.list_objects_v2(Bucket=SCRIPT_SOURCE_BUCKET, Prefix = prefix)['Contents']
               for file in sqlfiles[1:]:
                   key = file['Key']
                   obj = s3.get_object(Bucket=SCRIPT_SOURCE_BUCKET, Key=key)['Body'].read().decode('utf-8').replace('rolearn', ROLE_ARN).replace('bucket', UNLOAD_DEST_BUCKET)
                   putkey = 'sql/' + key.split('/')[-1]
                   response = s3.put_object(Bucket=DEST_BUCKET, Key=putkey, Body=obj)
    
               #copy script files directly
               prefix = SCRIPT_SOURCE_FOLDER + '/Python'
               scriptfiles = s3.list_objects_v2(Bucket=SCRIPT_SOURCE_BUCKET, Prefix = prefix)['Contents']
               for file in scriptfiles[1:]:
                   key = file['Key']
                   putkey = 'python/' + key.split('/')[-1]
                   response = s3.copy_object(Bucket=DEST_BUCKET,Key=putkey, CopySource={'Bucket': SCRIPT_SOURCE_BUCKET,'Key':key})
                   
               #update Secrets Manager secret with host and port
               sm = boto3.client('secretsmanager')
               sec = json.loads(sm.get_secret_value(SecretId=SECRET)['SecretString'])
               sec['host'] = CLUSTER_ENDPOINT
               sec['port'] = PORT
               sec['engine'] = ENGINE
               sec['dbClusterIdentifier'] = CLUSTER_ID
               newsec = json.dumps(sec)
               response = sm.update_secret(SecretId=SECRET, SecretString=newsec)
                

               # Send a response to CloudFormation pre-signed URL
               cfnresponse.send(event, context, cfnresponse.SUCCESS, {
                   'Message': 'Ingested Data'
                   }, 
                   context.log_stream_name)
                
               return {
                   'statusCode': 200,
                   'body': json.dumps('Copied Files')
               }

          - {
            ScriptBucket : !Ref ScriptBucket,
            DataBucket : !Ref DataBucket,
            Role : !GetAtt RedshiftRole.Arn,
            Endpoint: !GetAtt RedshiftCluster.Endpoint.Address,
            Secret: !Ref RedshiftSecret
            } 

  
  LambdaExecutionRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: /
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AWSLambdaExecute
        - arn:aws:iam::aws:policy/AmazonS3FullAccess
        - arn:aws:iam::aws:policy/SecretsManagerReadWrite

  GlueRedshiftConnection:
      Type: "AWS::Glue::Connection"
      Properties:
          ConnectionInput: 
            Description: String
            ConnectionType: JDBC
            PhysicalConnectionRequirements: 
                # AvailabilityZone: !Ref AvailabilityZone
                SecurityGroupIdList: 
                  - !Ref RedshiftSG
                SubnetId: !Ref SubnetAPrivate
            ConnectionProperties: 
                  "JDBC_CONNECTION_URL":
                      Fn::Join:
                      - ""
                      - - "jdbc:redshift://"
                        - Fn::Join:
                          - ":"
                          - - Fn::GetAtt:
                              - RedshiftCluster
                              - Endpoint.Address
                            - Fn::GetAtt:
                              - RedshiftCluster
                              - Endpoint.Port
                        - "/"
                        - !Sub "${RedshiftDBName}"
                  "USERNAME": !Join ['', ['{{resolve:secretsmanager:', !Ref RedshiftSecret, ':SecretString:username}}' ]]
                  "PASSWORD": !Join ['', ['{{resolve:secretsmanager:', !Ref RedshiftSecret, ':SecretString:password}}' ]]
            Name: GlueRedshiftConnection
          CatalogId: !Ref AWS::AccountId
  
  EnvSetup:
    Type: 'Custom::EnvSetup'
    DependsOn:
      - LambdaExecutionRole
      - DataBucket
      - ScriptBucket
    Properties:
      ServiceToken: !GetAtt 
        - SetupFunction
        - Arn

  QueryRedshift:
    Type: "AWS::Glue::Job"
    Properties:
      Role: !Ref GlueExecutionRole
      Name: "QueryRedshift"
      Connections:
        Connections:
        - !Ref GlueRedshiftConnection 
      Command: 
        Name : pythonshell
        PythonVersion: 3
        ScriptLocation: !Sub "s3://${ScriptBucket}/python/rs_query.py" 
      DefaultArguments: {
          "--s3_output_path": !Sub "s3://${DataBucket}/out/queries",
          "--extra-py-files" : !Sub "s3://${ScriptBucket}/python/redshift_module-0.1-py3.6.egg",
          "--db": !Sub "${RedshiftDBName}",
          "--db_creds": !Sub "${SecretsManagerSecret}",
          "--bucket": !Sub "${ScriptBucket}",
          "--redshift_role": !GetAtt [ RedshiftRole, Arn ]
      }
      MaxCapacity: 1
      ExecutionProperty:
        MaxConcurrentRuns: 20

  AWSGlueJobRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - glue.amazonaws.com
            Action:
              - sts:AssumeRole
      Policies:
        - PolicyName: root
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - "s3:GetObject"
                  - "s3:PutObject"
                  - "s3:ListBucket"
                  - "s3:DeleteObject"
                Resource: 
                  - !Sub "arn:aws:s3:::${DataBucket}"
                  - !Sub "arn:aws:s3:::${DataBucket}/*"
                  - !Sub "arn:aws:s3:::${ArtifactsBucket}"
                  - !Sub "arn:aws:s3:::${ArtifactsBucket}/*"
                  - !Sub "arn:aws:s3:::${ScriptBucket}"
                  - !Sub "arn:aws:s3:::${ScriptBucket}/*"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
      Path: "/"

  ProcessAmazonData:
    Type: "AWS::Glue::Job"
    Properties:
      GlueVersion: 2.0
      Role: !Ref AWSGlueJobRole
      Name: "ProcessAmazonData"
      Command: {
        "Name" : "glueetl",
        "ScriptLocation": !Sub "s3://${ArtifactsBucket}/scripts/process_amazon_data.py"
      }
      DefaultArguments: {
          "--s3_output_path": !Sub "s3://${DataBucket}/out/amazon/"
      }
      MaxRetries: 0
      AllocatedCapacity: 5

  ProcessNetflixData:
    Type: "AWS::Glue::Job"
    Properties:
      Role: !Ref AWSGlueJobRole
      Name: "ProcessNetflixData"
      GlueVersion: 2.0

      Command: {
        "Name" : "glueetl",
        "ScriptLocation": !Sub "s3://${ArtifactsBucket}/scripts/process_netflix_data.py"
      }
      DefaultArguments: {
          "--s3_output_path": !Sub "s3://${DataBucket}/out/netflix/"
      }
      WorkerType: G.1X
      NumberOfWorkers: 10
      MaxRetries: 0


  JoinAndProcessData:
    Type: "AWS::Glue::Job"
    Properties:
      GlueVersion: 2.0
      Role: !Ref AWSGlueJobRole
      Name: "JoinAndProcessData"
      Command: {
        "Name" : "glueetl",
        "ScriptLocation": !Sub "s3://${ArtifactsBucket}/scripts/join_data.py"
      }
      DefaultArguments: {
          "--s3_output_path": !Sub "s3://${DataBucket}/out/"
      }
      MaxRetries: 0
      WorkerType: G.1X
      NumberOfWorkers: 25

  GlueJobStateMachine:
    Type: AWS::StepFunctions::StateMachine
    Properties:
      DefinitionString:
        Fn::Sub:
          - |
              {
                "StartAt": "MyParallelState",
                "States": {
                  "MyParallelState": {
                "Type": "Parallel",
                "Next": "JoinAndProcessData",
                "Branches": [
                  {
                    "StartAt": "ProcessAmazonData",
                    "States": {
                      "ProcessAmazonData": {
                        "Type": "Task",
                        "Resource": "arn:aws:states:::glue:startJobRun.sync",
                         "Parameters":{
                      "JobName":"ProcessAmazonData"
                    },
                        
                        "End": true
                      }
                    }
                  },
                  {
                    "StartAt": "ProcessNetflixData",
                    "States": {
                      "ProcessNetflixData": {
                        "Type": "Task",
                         "Resource": "arn:aws:states:::glue:startJobRun.sync",
                         "Parameters":{
                      "JobName":"ProcessNetflixData"
                    },
                        "End": true
                      }
                    }
                  }
                ]}
                ,
                 
                  "JoinAndProcessData": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::glue:startJobRun.sync",
                    "Parameters":{
                      "JobName":"JoinAndProcessData"
                    },
                    "Next": "CreateTables"
                   },
               "CreateTables": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::glue:startJobRun.sync",
                    "Parameters":{
                      "JobName":"QueryRedshift",
                      "Arguments":{
                        "--file": "sql/create_schemas.sql"
                      }
                    },
                      "Next": "LoadRedshift"
                   },
                   
                  "LoadRedshift": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::glue:startJobRun.sync",
                    "Parameters":{
                      "JobName":"QueryRedshift",
                      "Arguments":{
                        "--file": "sql/load_redshift.sql"
                      }
                    },
                      "Next": "RunQueries"
                   },

                  "RunQueries": {
                    "Type": "Task",
                    "Resource": "arn:aws:states:::glue:startJobRun.sync",
                    "Parameters":{
                      "JobName":"QueryRedshift",
                      "Arguments":{
                        "--db":"${Database}", 
                        "--db_creds":"${DBInfo}",
                        "--bucket": "${Bucket}",
                        "--file": "sql/business_queries.sql"
                      }
                    },
                      "End": true
                   }

                }
              }

          - { Database: !Ref RedshiftDBName, DBInfo: !Ref SecretsManagerSecret, Bucket: !Ref ScriptBucket }

        
      RoleArn: !GetAtt [ StatesExecutionRole, Arn ]
      
  StatesExecutionRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service:
                - !Sub states.${AWS::Region}.amazonaws.com
            Action: "sts:AssumeRole"
      Path: "/"

  StatePolicy:
    Type: AWS::IAM::Policy 
    Properties:
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          -
            Effect: "Allow"
            Action:
              - "glue:StartJobRun"
              - "glue:GetJobRun"
              - "glue:GetJobRuns"
              - "glue:BatchStopJobRun"
            Resource: "*"
          -
            Effect: Allow
            Action:
              - "sns:publish"
            Resource: !Ref FailNotificationSNS
      PolicyName: statepolicy
      Roles:
      - Ref: StatesExecutionRole

  
  FailNotificationSNS:  
    Type: AWS::SNS::Topic


Outputs:
  SecretsManagerSecret:
    Description: 'Redshift authentication data'
    Value: !Ref RedshiftSecret
  GlueRedshiftConnection:
    Description: 'JDBC connection to Redshift for Glue'
    Value: !Ref GlueRedshiftConnection
  StateMachine:
    Description: 'State Machine that performs ETL jobs and queries Redshift'
    Value: !Ref GlueJobStateMachine
  Bucket:
    Description: 'Bucket with transformed data and Redshift query results'
    Value: !Ref DataBucket