AWSTemplateFormatVersion: "2010-09-09"
Parameters:
  datalakeAccountid:
    Type: String        
    Default: 838011049881
Resources:
  executeQueryFunction:
    Type: AWS::Lambda::Function
    Properties:
      Role: !GetAtt 'executeQueryRole.Arn'
      Runtime: "python2.7"
      Timeout: 300
      Handler: "index.handler"
      Code: 
        ZipFile: |
          import boto3
          import cfnresponse
          import time
          client = boto3.client('athena')
          def handler(event, context):
              if event['RequestType'] != 'Delete':
                try:
                    query = event['ResourceProperties']['QueryString']
                    queryoutput = event['ResourceProperties']['queryoutput']
                    response = client.start_query_execution(QueryString=query, ResultConfiguration={'OutputLocation': queryoutput})
                except Exception as e:
                    print(e)
                    cfnresponse.send(event, context, cfnresponse.SUCCESS, {}, "CustomResourcePhysicalID")
              cfnresponse.send(event, context, cfnresponse.SUCCESS, {}, "CustomResourcePhysicalID")
              return "Success"

  executeCreateTableQuery:
    Type: Custom::executeAthenaQuery
    DependsOn: executeCreateDatabaseQuery
    Properties:
      ServiceToken: !GetAtt executeQueryFunction.Arn
      queryoutput: !Join ["",[ "s3://", !Ref ResultBucket, "/queryresults" ] ]
      QueryString: >
                    CREATE EXTERNAL TABLE `newdb.events`(
                        `globaleventid` int, 
                        `day` int, 
                        `monthyear` int, 
                        `year` int, 
                        `fractiondate` float, 
                        `actor1code` string, 
                        `actor1name` string, 
                        `actor1countrycode` string, 
                        `actor1knowngroupcode` string, 
                        `actor1ethniccode` string, 
                        `actor1religion1code` string, 
                        `actor1religion2code` string, 
                        `actor1type1code` string, 
                        `actor1type2code` string, 
                        `actor1type3code` string, 
                        `actor2code` string, 
                        `actor2name` string, 
                        `actor2countrycode` string, 
                        `actor2knowngroupcode` string, 
                        `actor2ethniccode` string, 
                        `actor2religion1code` string, 
                        `actor2religion2code` string, 
                        `actor2type1code` string, 
                        `actor2type2code` string, 
                        `actor2type3code` string, 
                        `isrootevent` boolean, 
                        `eventcode` string, 
                        `eventbasecode` string, 
                        `eventrootcode` string, 
                        `quadclass` int, 
                        `goldsteinscale` float, 
                        `nummentions` int, 
                        `numsources` int, 
                        `numarticles` int, 
                        `avgtone` float, 
                        `actor1geo_type` int, 
                        `actor1geo_fullname` string, 
                        `actor1geo_countrycode` string, 
                        `actor1geo_adm1code` string, 
                        `actor1geo_lat` float, 
                        `actor1geo_long` float, 
                        `actor1geo_featureid` int, 
                        `actor2geo_type` int, 
                        `actor2geo_fullname` string, 
                        `actor2geo_countrycode` string, 
                        `actor2geo_adm1code` string, 
                        `actor2geo_lat` float, 
                        `actor2geo_long` float, 
                        `actor2geo_featureid` int, 
                        `actiongeo_type` int, 
                        `actiongeo_fullname` string, 
                        `actiongeo_countrycode` string, 
                        `actiongeo_adm1code` string, 
                        `actiongeo_lat` float, 
                        `actiongeo_long` float, 
                        `actiongeo_featureid` int, 
                        `dateadded` int, 
                        `sourceurl` string)
                      ROW FORMAT DELIMITED 
                        FIELDS TERMINATED BY '\t' 
                      STORED AS INPUTFORMAT 
                        'org.apache.hadoop.mapred.TextInputFormat' 
                      OUTPUTFORMAT 
                        'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
                      LOCATION
                        's3://gdelt-open-data/events'
                      TBLPROPERTIES (
                        'transient_lastDdlTime'='1558729743')

  executeCreateDatabaseQuery:
    Type: Custom::executeAthenaQuery
    Properties:
      ServiceToken: !GetAtt executeQueryFunction.Arn
      queryoutput:  !Join ["",[ "s3://", !Ref ResultBucket, "/queryresults" ] ]
      QueryString: "CREATE DATABASE newdb;"

  executeCreateTable2Query:
    Type: Custom::executeAthenaQuery
    DependsOn: executeCreateTableQuery
    Properties:
      ServiceToken: !GetAtt executeQueryFunction.Arn
      queryoutput: !Join ["",[ "s3://", !Ref ResultBucket, "/queryresults"  ]]
      QueryString: !Sub |
                    CREATE TABLE newdb.eventsparquet
                    WITH (
                      format='PARQUET',
                      external_location='s3://${ResultBucket}/gdeltevents'
                    ) AS
                    SELECT * FROM "newdb"."events"

  ResultBucket:
    Type: AWS::S3::Bucket

  executeQueryRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service:
            - lambda.amazonaws.com
          Action:
          - sts:AssumeRole
      Path: /
      Policies:
      - PolicyName: root
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
          - Action:
            - s3:*

            - athena:*
            - glue:*
            - logs:CreateLogGroup
            - logs:CreateLogStream
            - logs:PutLogEvents
            Resource: '*'
            Effect: Allow

  CopyLambdasFunction:
      Type: AWS::Lambda::Function
      Properties:
          Role: !GetAtt 'executeQueryRole.Arn'
          Runtime: "python2.7"

          Timeout: 13
          Handler: "index.lambda_handler"
          Environment:
            Variables:
              s3destinationBucket: !Ref 'ResultBucket'
          Code: 
            ZipFile: |
              import cfnresponse
              import boto3
              import os
              def lambda_handler(event, context):
                s3 = boto3.client('s3')
                destinationbucket = os.environ['s3destinationBucket']
                sourceBucket = 'lfcarocomdemo'
                objectKey = 'lakeformationcrossaccount/countries/CAMEO.country.txt'
                objectKey2='lakeformationcrossaccount/eventcodes/CAMEO.eventcodes.txt'
                objectKey3='lakeformationcrossaccount/groups/CAMEO.knowngroup.txt'
                objectKey4='lakeformationcrossaccount/types/CAMEO.type.txt'
                print(event)
                if event['RequestType'] == "Delete" :
                  object_names = ['objectKey','objectKey2','objectKey3','objectKey4']
                  objlist = [{'Key': obj} for obj in object_names]

                  s3.delete_objects(Bucket=destinationbucket, Delete={'Objects':objlist})

                  cfnresponse.send(event, context, cfnresponse.SUCCESS, {})
                  return ('hello2')
                if event['RequestType'] == 'Create':
                  s3.copy({ 'Bucket': sourceBucket, 'Key': objectKey}, destinationbucket, objectKey)
                  s3.copy({ 'Bucket': sourceBucket, 'Key': objectKey2}, destinationbucket, objectKey2)
                  s3.copy({ 'Bucket': sourceBucket, 'Key': objectKey3}, destinationbucket, objectKey3)
                  s3.copy({ 'Bucket': sourceBucket, 'Key': objectKey4}, destinationbucket, objectKey4)
                  cfnresponse.send(event, context, cfnresponse.SUCCESS, {})   

  MyCustomResource: 
    Type: "Custom::TestLambdaCrossStackRef"
    DependsOn: ['CopyLambdasFunction']
    Properties: 
      ServiceToken:
        !GetAtt 'CopyLambdasFunction.Arn'
  
  lakeformationpolicy: 
    Type: AWS::S3::BucketPolicy
    Properties: 
      Bucket: 
        Ref: "ResultBucket"
      PolicyDocument: 
        Statement: 
          - 
            Action: 
              - "s3:GetBucketLocation"
              - "s3:GetObject"
              - "s3:ListBucket"
              - "s3:ListBucketMultipartUploads"
              - "s3:ListMultipartUploadParts"
              - "s3:AbortMultipartUpload"
              - "s3:PutObject"              
            Effect: "Allow"
            Resource: 
              - !Join [ "", [ "arn:aws:s3:::",Ref: "ResultBucket", "/*" ] ]
              - !Join [ "", [ "arn:aws:s3:::",Ref: "ResultBucket" ] ]
            Principal: 
              AWS: !Join [ "", [ "arn:aws:iam::",Ref: "datalakeAccountid", ":root" ] ]


  executeCreateTable3Query:
    Type: Custom::executeAthenaQuery
    DependsOn: 
      - CopyLambdasFunction
      - executeCreateDatabaseQuery
    Properties:
      ServiceToken: !GetAtt executeQueryFunction.Arn
      queryoutput: !Join ["",[ "s3://", !Ref ResultBucket, "/queryresults" ] ]
      QueryString: !Sub |
                  
                    CREATE EXTERNAL TABLE `newdb.countries`(
                      `code` string, 
                      `country` string)
                    ROW FORMAT DELIMITED 
                      FIELDS TERMINATED BY '\t' 
                    STORED AS INPUTFORMAT 
                      'org.apache.hadoop.mapred.TextInputFormat' 
                    OUTPUTFORMAT 
                      'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
                    LOCATION
                      's3://${ResultBucket}/lakeformationcrossaccount/countries'
                    TBLPROPERTIES (
                      'skip.header.line.count'='1', 
                      'transient_lastDdlTime'='1559938070')

  executeCreateTable4Query:
    Type: Custom::executeAthenaQuery
    DependsOn: 
      - CopyLambdasFunction
      - executeCreateDatabaseQuery
    Properties:
      ServiceToken: !GetAtt executeQueryFunction.Arn
      queryoutput:  !Join ["",[ "s3://", !Ref ResultBucket, "/queryresults" ] ]
      QueryString: !Sub |
                    CREATE EXTERNAL TABLE `newdb.eventcodes`(
                      `code` string, 
                      `description` string)
                    ROW FORMAT DELIMITED 
                      FIELDS TERMINATED BY '\t' 
                    STORED AS INPUTFORMAT 
                      'org.apache.hadoop.mapred.TextInputFormat' 
                    OUTPUTFORMAT 
                      'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
                    LOCATION
                      's3://${ResultBucket}/eventcodes/CAMEO.eventcodes.txt'
                    TBLPROPERTIES (
                      'skip.header.line.count'='1', 
                      'transient_lastDdlTime'='1559938024')

  executeCreateTable5Query:
    Type: Custom::executeAthenaQuery
    DependsOn: 
      - CopyLambdasFunction
      - executeCreateDatabaseQuery
    Properties:
      ServiceToken: !GetAtt executeQueryFunction.Arn
      queryoutput: !Join ["",[ "s3://", !Ref ResultBucket, "/queryresults" ]] 
      QueryString: !Sub |
                    CREATE EXTERNAL TABLE `newdb.groups`(
                      `group` string, 
                      `description` string)
                    ROW FORMAT DELIMITED 
                      FIELDS TERMINATED BY '\t' 
                    STORED AS INPUTFORMAT 
                      'org.apache.hadoop.mapred.TextInputFormat' 
                    OUTPUTFORMAT 
                      'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
                    LOCATION
                      's3://${ResultBucket}/groups/CAMEO.knowngroup.txt'
                    TBLPROPERTIES (
                      'skip.header.line.count'='1', 
                      'transient_lastDdlTime'='1559938063')

  executeCreateTable6Query:
    Type: Custom::executeAthenaQuery
    DependsOn: 
      - CopyLambdasFunction
      - executeCreateDatabaseQuery
    Properties:
      ServiceToken: !GetAtt executeQueryFunction.Arn
      queryoutput:  !Join ["",[ "s3://", !Ref ResultBucket, "/queryresults" ] ]
      QueryString: !Sub |
                    CREATE EXTERNAL TABLE `newdb.types`(
                      `type` string, 
                      `description` string)
                    ROW FORMAT DELIMITED 
                      FIELDS TERMINATED BY '\t' 
                    STORED AS INPUTFORMAT 
                      'org.apache.hadoop.mapred.TextInputFormat' 
                    OUTPUTFORMAT 
                      'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
                    LOCATION
                      's3://${ResultBucket}/types/CAMEO.type.txt'
                    TBLPROPERTIES (
                      'skip.header.line.count'='1', 
                      'transient_lastDdlTime'='1559938044')

