flink-refarch-infrastructure.yml

AWSTemplateFormatVersion: 2010-09-09

Description: Reference architecture for real-time stream processing with Apache Flink on Amazon EMR, Amazon Kinesis, and Amazon Elasticsearch Service.

Parameters:
  KeyName:
    ConstraintDescription: must be the name of an existing EC2 KeyPair.
    Description: Name of an existing EC2 KeyPair to enable SSH access to the instance
    Type: AWS::EC2::KeyPair::KeyName

  ClientIpAddress:
    AllowedPattern: (\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3})/(\d{1,2})
    ConstraintDescription: must be a valid IP CIDR range of the form x.x.x.x/x.
    Description: IP address range that is able to access the EC2 instance and that gets full read and write access to the Elasticsearch Cluster and the Kibana dashboard
    MaxLength: 18
    MinLength: 9
    Type: String

  ArtifactBucketName:
    Description: Name of the bucket containing the runtime artifacts (see output section of the flink-refarch-build-artifacts template)
    Type: String


Outputs:
  EmrMasterNode:
    Description: EMR master node SSH connection
    Value: !Sub ssh -C -D 8157 hadoop@${FlinkCluster.MasterPublicDNS}

  StartFlinkDemon:
    Description: Start Flink Demon
    Value: flink-yarn-session -n 2 -s 2 -jm 768 -tm 1024 -d

  FlinkCommand:
    Description: Command to invoke Flink job on the EMR cluster
    Value: !Sub flink run -p 4 flink-taxi-stream-processor-1.3.jar --region ${AWS::Region} --stream ${KinesisStream} --es-endpoint https://${ElasticSearch.DomainEndpoint} --checkpoint s3://${FlinkDataBucket}/checkpoint-data

  ProducerInstance:
    Description: SSH connection to the instance that ingests trip events into the Kinesis stream
    Value: !Sub ssh -C ec2-user@${ProducerInstance.PublicDnsName}

  ProducerCommand:
    Description: Command to populate Kinesis stream with taxi trip events
    Value: !Sub java -jar kinesis-taxi-stream-producer-1.3.jar -stream ${KinesisStream} -region ${AWS::Region} -speedup 6480
  
  KibanaDashboardURL:
    Description: URL to access the Kibana dasboards
    Value: !Sub https://${ElasticSearch.DomainEndpoint}/_plugin/kibana/app/kibana#/dashboard/Taxi-Trips-Dashboard


Resources:
  AttachGateway:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      InternetGatewayId: !Ref InternetGateway
      VpcId: !Ref VPC

  SshSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: !Sub allow ssh access from ${ClientIpAddress}
      SecurityGroupIngress:
      - CidrIp: !Ref ClientIpAddress
        IpProtocol: tcp
        FromPort: 22
        ToPort: 22
      VpcId: !Ref VPC

  InternetGateway:
    Type: AWS::EC2::InternetGateway
    Properties:
      Tags:
      - Key: Application
        Value: !Ref AWS::StackId

  Route:
    Type: AWS::EC2::Route
    Properties:
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId: !Ref InternetGateway
      RouteTableId: !Ref RouteTable
    DependsOn: AttachGateway

  RouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      Tags:
      - Key: Application
        Value: !Ref AWS::StackId
      VpcId: !Ref VPC

  Subnet:
    Type: AWS::EC2::Subnet
    Properties:
      CidrBlock: 10.0.0.0/24
      Tags:
      - Key: Application
        Value: !Ref AWS::StackId
      VpcId: !Ref VPC

  SubnetRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref RouteTable
      SubnetId: !Ref Subnet

  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.0.0.0/16
      Tags:
      - Key: Application
        Value: !Ref AWS::StackId
      EnableDnsHostnames: true

  RolePolicy:
    Type: AWS::IAM::Policy
    Properties:
      PolicyName: !Sub "${AWS::StackName}-kinesis-cw-s3-es-access"
      Roles:
        - Ref: ProducerRole
        - Ref: EmrRole
      PolicyDocument:
        !Sub |
          {
            "Version": "2012-10-17",
            "Statement": [
              {
                  "Effect": "Allow",
                  "Action": [
                      "kinesis:DescribeStream",
                      "kinesis:ListShards",
                      "kinesis:GetShardIterator",
                      "kinesis:GetRecords",
                      "kinesis:PutRecord",
                      "kinesis:PutRecords"
                  ],
                  "Resource": "${KinesisStream.Arn}"
              },
              {
                  "Effect": "Allow",
                  "Action": [
                      "cloudwatch:PutMetricData"
                  ],
                  "Resource": "*"
              },
              {
                  "Effect": "Allow",
                  "Action": "es:ESHttp*",
                  "Resource": "${ElasticSearch.DomainArn}/*"
              },
              {
                  "Effect": "Allow",
                  "Action": "cloudformation:DescribeStacks",
                  "Resource": "arn:aws:cloudformation:${AWS::Region}:${AWS::AccountId}:stack/${AWS::StackName}/*"
              },
              {
                  "Effect": "Allow",
                  "Action": [
                      "s3:GetObject",
                      "s3:ListBucket"
                  ],
                  "Resource": [
                    "arn:aws:s3:::${ArtifactBucketName}",
                    "arn:aws:s3:::${ArtifactBucketName}/*",
                    "arn:aws:s3:::aws-bigdata-blog",
                    "arn:aws:s3:::aws-bigdata-blog/*"
                  ]  
              }
            ]
          }

  EmrRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          -
            Effect: Allow
            Principal:
              Service:
                - ec2.amazonaws.com
            Action:
              - sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role

  EmrInstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties:
      Roles:
        - Ref: EmrRole

  ProducerRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          -
            Effect: Allow
            Principal:
              Service:
                - ec2.amazonaws.com
            Action:
              - sts:AssumeRole

  ProducerInstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties:
      Roles:
        - Ref: ProducerRole

  KinesisStream:
    Type: AWS::Kinesis::Stream
    Properties:
      ShardCount: 32

  ElasticSearch:
    Type: AWS::Elasticsearch::Domain
    Properties:
      AccessPolicies:
        !Sub |
          {
            "Version": "2012-10-17",
            "Statement": [
              {
                "Effect": "Allow",
                "Principal": {
                  "AWS": "*"
                },
                "Action": [
                  "es:ESHttpGet",
                  "es:ESHttpPut",
                  "es:ESHttpPost",
                  "es:ESHttpHead"
                ],
                "Resource": "*",
                "Condition": {
                  "IpAddress": {
                    "aws:SourceIp": [
                      "${ClientIpAddress}",
                      "${ProducerInstanceIp}"
                    ]
                  }
                }
              }
            ]
          }
      ElasticsearchClusterConfig:
        InstanceCount: 2
        InstanceType: m3.medium.elasticsearch
        ZoneAwarenessEnabled: true
      ElasticsearchVersion: 5.5


  FlinkCluster:
    Type: AWS::EMR::Cluster
    Properties:
      Name: !Sub ${AWS::StackName}-cluster
      Applications:
        - Name: Hadoop
        - Name: Ganglia
        - Name: Flink
        - Name: ZooKeeper
      Instances:
        MasterInstanceGroup:
          InstanceCount: 1
          InstanceType: m4.large
          Market: ON_DEMAND
          Name: Master
        CoreInstanceGroup:
          InstanceCount: 2
          InstanceType: c4.large
          Market: ON_DEMAND
          Name: Core
        Ec2KeyName: !Ref KeyName
        Ec2SubnetId: !Ref Subnet
        AdditionalMasterSecurityGroups:
          - !Ref SshSecurityGroup
      JobFlowRole : !Ref EmrInstanceProfile
      LogUri: !Sub s3://${FlinkDataBucket}/emr-logs
      ServiceRole : EMR_DefaultRole
      ReleaseLabel: emr-5.16.0
      VisibleToAllUsers: true
      Configurations:
        - Classification: flink-conf
          ConfigurationProperties:
            state.checkpoints.dir: !Sub s3://${FlinkDataBucket}/checkpoint-metadata
            high-availability: zookeeper
            high-availability.zookeeper.quorum: "%{hiera('hadoop::zk')}"
            high-availability.zookeeper.storageDir: hdfs:///flink/recovery
        - Classification: hdfs-site
          ConfigurationProperties:
            dfs.replication: 2


  FlinkDataBucket:
    Type: AWS::S3::Bucket

  ProducerInstanceIp:
    Type: AWS::EC2::EIP
    DependsOn: AttachGateway
    Properties:
      Domain: vpc

  ProducerInstanceIpAssiciation:
    Type: AWS::EC2::EIPAssociation
    Properties:
      InstanceId: !Ref ProducerInstance
      AllocationId: !GetAtt ProducerInstanceIp.AllocationId


  ProducerInstance:
    Type: AWS::EC2::Instance
    DependsOn: AttachGateway
    Properties:
      ImageId:
        Fn::FindInMap:
        - AWSRegionArch2AMI
        - !Ref AWS::Region
        - HVM64
      Monitoring: true
      InstanceType: m5.2xlarge
      IamInstanceProfile: !Ref ProducerInstanceProfile
      KeyName: !Ref KeyName
      NetworkInterfaces:
      - AssociatePublicIpAddress: true
        DeleteOnTermination: true
        DeviceIndex: 0
        GroupSet:
        - !Ref SshSecurityGroup
        SubnetId: !Ref Subnet
      Tags:
      - Key: Application
        Value: !Ref AWS::StackId
      - Key: Name
        Value: !Sub ${AWS::StackName}-producer-instance
      UserData:
        Fn::Base64:
          !Sub |
            #!/bin/bash -xe

            yum install -y java-1.8.0-openjdk-devel.x86_64 jq tmux
            alternatives --set java /usr/lib/jvm/jre-1.8.0-openjdk.x86_64/bin/java

            aws cloudformation wait stack-create-complete --stack-name '${AWS::StackName}' --region ${AWS::Region}

            curl -s https://raw.githubusercontent.com/awslabs/flink-stream-processing-refarch/master/elasticsearch/es-index.json | curl -s -XPUT https://${ElasticSearch.DomainEndpoint}/taxi-dashboard -d @-
            curl -s -XPUT https://${ElasticSearch.DomainEndpoint}/.kibana/index-pattern/taxi-dashboard -d '{"title" : "taxi-dashboard",  "timeFieldName": "timestamp"}'
            curl -s -XPUT https://${ElasticSearch.DomainEndpoint}/.kibana/config/5.5.0 -d '{"defaultIndex" : "taxi-dashboard"}'
            curl -s https://raw.githubusercontent.com/awslabs/flink-stream-processing-refarch/master/elasticsearch/kibana-visualizations-and-dashboard.json | jq -rc '.[]' | while read -r item; do jq -rn "$item | ._source" | curl -s -XPUT https://${ElasticSearch.DomainEndpoint}/.kibana/$(jq -rn "$item | [._type,._id] | join(\"/\")") -d @-; done


  FlinkNotProcessingFromKinesisAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmDescription: Alarm if Flink stops reading from the Kinesis stream or falls behind processing the available data
      Namespace: AWS/Kinesis
      MetricName: GetRecords.IteratorAgeMilliseconds
      Dimensions:
        - Name: StreamName
          Value: !Sub ${KinesisStream}
      Statistic: Average
      Period: 60
      EvaluationPeriods: 1
      Threshold: 60000
      ComparisonOperator: GreaterThanThreshold
      TreatMissingData: breaching

  FlinkDashboard:
    Type: AWS::CloudWatch::Dashboard
    Properties:
      DashboardName: !Sub ${AWS::StackName}-dashboard
      DashboardBody: !Sub |
        {
            "widgets": [
                {
                    "type": "metric",
                    "x": 0,
                    "y": 0,
                    "width": 21,
                    "height": 6,
                    "properties": {
                        "view": "timeSeries",
                        "stacked": false,
                        "metrics": [
                            [ "AWS/Kinesis", "IncomingRecords", "StreamName", "${KinesisStream}", { "period": 60, "stat": "Sum" } ],
                            [ ".", "WriteProvisionedThroughputExceeded", ".", ".", { "period": 60, "yAxis": "right" } ]
                        ],
                        "region": "${AWS::Region}",
                        "period": 60,
                        "yAxis": {
                            "right": {
                                "min": 0,
                                "max": 100
                            }
                        }
                    }
                },
                {
                    "type": "metric",
                    "x": 0,
                    "y": 6,
                    "width": 21,
                    "height": 6,
                    "properties": {
                        "view": "timeSeries",
                        "stacked": false,
                        "metrics": [
                            [ "AWS/Kinesis", "GetRecords.Records", "StreamName", "${KinesisStream}", { "period": 60, "stat":"Sum" } ],
                            [ ".", "GetRecords.IteratorAgeMilliseconds", ".", ".", { "period": 60, "yAxis":"right" } ]
                        ],
                        "region": "${AWS::Region}",
                        "period": 60,
                        "yAxis": {
                            "left": {
                                "min": 0
                            },
                            "right": {
                                "min": 0
                            }
                        }
                    }
                },
                {
                    "type": "metric",
                    "x": 0,
                    "y": 12,
                    "width": 21,
                    "height": 6,
                    "properties": {
                        "view": "timeSeries",
                        "stacked": false,
                        "metrics": [
                            [ "AWS/ES", "SearchableDocuments", "DomainName", "${ElasticSearch}", "ClientId", "${AWS::AccountId}", { "period": 60 } ],
                            [ ".", "CPUUtilization", ".", ".", ".", ".", { "period": 60, "yAxis": "right" } ]
                        ],
                        "region": "${AWS::Region}",
                        "period": 60,
                        "yAxis": {
                            "left": {
                                "min": 0
                            },
                            "right": {
                                "min": 0,
                                "max": 100
                            }
                        }
                    }
                }
            ]
        }

Mappings:
  AWSRegionArch2AMI:
    ap-northeast-1:
      HVM64: ami-374db956
    ap-northeast-2:
      HVM64: ami-2b408b45
    ap-south-1:
      HVM64: ami-ffbdd790
    ap-southeast-1:
      HVM64: ami-a59b49c6
    ap-southeast-2:
      HVM64: ami-dc361ebf
    cn-north-1:
      HVM64: ami-8e6aa0e3
    eu-central-1:
      HVM64: ami-ea26ce85
    eu-west-1:
      HVM64: ami-f9dd458a
    sa-east-1:
      HVM64: ami-6dd04501
    us-east-1:
      HVM64: ami-6869aa05
    us-west-1:
      HVM64: ami-31490d51
    us-west-2:
      HVM64: ami-7172b611