はじめに

今回は Amazon Elasticsearch Service を試します。
DynamoDBからデータをロードする方式を試します。
[Lambda関数・SAMテンプレート]
(https://github.com/tanaka-takurou/serverless-elastic-search-page-go)

システム構成図

準備

AWS SAM の準備をします

[Amazon Elasticsearch Serviceの資料]
Amazon Elasticsearch Service

AWS SAM テンプレート作成

AWS SAM テンプレートで API-Gateway , Lambda, DynamoDB, Amazon Elasticsearch Serviceの設定をします。

[参考資料]
AWS SAM テンプレートを作成する

template.yml

template.yml


AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: Serverless Elastic Search Service Page

Parameters:
  ApplicationName:
    Type: String
    Default: 'ServerlessElasticSearchServicePage'
  FrontPageApiStageName:
    Type: String
    Default: 'ProdStage'
  DataTableName:
    Type: String
    Default: 'search_target'
  ESDomainName:
    Type: String
    Default: 'es-sample'
    AllowedPattern: "[a-z][a-z0-9\\-]+"
  ESIndexName:
    Type: String
    Default: 'messages'
  ESTypeName:
    Type: String
    Default: 'message'

Resources:
  FrontPageApi:
    Type: AWS::Serverless::Api
    Properties:
      Name: ServerlessElasticSearchServicePageApi
      EndpointConfiguration: REGIONAL
      StageName: !Ref FrontPageApiStageName
  FrontPageFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: ServerlessElasticSearchServicePageFrontFunction
      CodeUri: bin/
      Handler: main
      MemorySize: 256
      Runtime: go1.x
      Description: 'Test Front Function'
      Policies:
      Environment:
        Variables:
          REGION: !Ref 'AWS::Region'
          API_PATH: !Join [ '', [ '/', !Ref FrontPageApiStageName, '/api'] ]
      Events:
        FrontPageApi:
          Type: Api
          Properties:
            Path: '/'
            Method: get
            RestApiId: !Ref FrontPageApi
  ElasticSearchServiceLoadFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: ElasticSearchServiceLoadFunction
      CodeUri: api/load/bin/
      Handler: main
      MemorySize: 256
      Runtime: go1.x
      Description: 'ElasticSearchService Service Load Function'
      Policies:
      Environment:
        Variables:
          REGION: !Ref 'AWS::Region'
          DOMAIN: !GetAtt ElasticSearchServiceDomain.DomainEndpoint
          ES_INDEX_NAME: !Ref ESIndexName
          ES_TYPE_NAME: !Ref ESTypeName
      Events:
        FrontPageApi:
          Type: Api
          Properties:
            Path: '/api'
            Method: post
            RestApiId: !Ref FrontPageApi
  ElasticSearchServiceSaveFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: ElasticSearchServiceSaveFunction
      CodeUri: api/save/bin/
      Handler: main
      MemorySize: 256
      Runtime: go1.x
      Description: 'ElasticSearchService Service Save Function'
      Policies:
      Environment:
        Variables:
          REGION: !Ref 'AWS::Region'
          DOMAIN: !GetAtt ElasticSearchServiceDomain.DomainEndpoint
          ES_INDEX_NAME: !Ref ESIndexName
          ES_TYPE_NAME: !Ref ESTypeName
      Events:
        Stream:
          Type: DynamoDB
          Properties:
            Stream: !GetAtt DataTable.StreamArn
            BatchSize: 100
            StartingPosition: LATEST
  DataTable:
    Type: AWS::DynamoDB::Table
    Properties:
      AttributeDefinitions:
      - AttributeName: "id"
        AttributeType: "N"
      KeySchema:
      - AttributeName: "id"
        KeyType: "HASH"
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5
      SSESpecification:
        SSEEnabled: True
      TableName: !Ref DataTableName
      StreamSpecification:
        StreamViewType: "NEW_IMAGE"
  ElasticsearchDomain:
    Type: AWS::Elasticsearch::Domain
    Properties:
      DomainName: !Ref ESDomainName
      ElasticsearchClusterConfig:
        DedicatedMasterEnabled: "false"
        InstanceCount: "1"
        ZoneAwarenessEnabled: "false"
        InstanceType: "t2.micro.elasticsearch"
      EBSOptions:
        EBSEnabled: true
        VolumeSize: 10
        VolumeType: "gp2"
      SnapshotOptions:
        AutomatedSnapshotStartHour: "0"
      AccessPolicies:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              AWS: "*"
            Action: "es:ESHttp*"
            Resource: "*"
      AdvancedOptions:
        rest.action.multi.allow_explicit_index: "true"

Outputs:
  APIURI:
    Description: "URI"
    Value: !Join [ '', [ 'https://', !Ref FrontPageApi, '.execute-api.',!Ref 'AWS::Region','.amazonaws.com/',!Ref FrontPageApiStageName,'/'] ]

Amazon Elasticsearch Serviceの設定は以下の部分

  ElasticsearchDomain:
    Type: AWS::Elasticsearch::Domain
    Properties:
      DomainName: !Ref ESDomainName
      ElasticsearchClusterConfig:
        DedicatedMasterEnabled: "false"
        InstanceCount: "1"
        ZoneAwarenessEnabled: "false"
        InstanceType: "t2.micro.elasticsearch"
      EBSOptions:
        EBSEnabled: true
        VolumeSize: 10
        VolumeType: "gp2"
      SnapshotOptions:
        AutomatedSnapshotStartHour: "0"
      AccessPolicies:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              AWS: "*"
            Action: "es:ESHttp*"
            Resource: "*"
      AdvancedOptions:
        rest.action.multi.allow_explicit_index: "true"

Lambda関数作成

※ Lambda関数は aws-lambda-go を利用しました。

Elasticsearchのインデックスを追加・更新するには POSTリクエストを使う

	jsonStr := `{"` + name + `":"` + value + `"}`
	req, err := http.NewRequest(
		"POST",
		os.Getenv("DOMAIN") + "/" + os.Getenv("ES_INDEX_NAME") + "/" + os.Getenv("ES_TYPE_NAME") + "/",
		bytes.NewBuffer([]byte(jsonStr)),
	)
	if err != nil {
		return err
	}
    req.Header.Set("Content-Type", "application/json")
	client := &http.Client{}
	res, err := client.Do(req)
	if err != nil {
		return err
	}
	defer res.Body.Close()

Elasticsearchのインデックスを検索するには GETリクエストを使う

	var d interface{}
	res, err := http.Get(os.Getenv("DOMAIN") + "/" + os.Getenv("ES_INDEX_NAME") + "/" + os.Getenv("ES_TYPE_NAME") + "/_search?q=" + word)
	if err != nil {
		return d, err
	}
	defer res.Body.Close()

	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		return d, err
	}

	err = json.Unmarshal(body, &d)
	if err != nil {
		return d, err
	}
	return d, nil

終わりに

Amazon Elasticsearch Serviceは、完全マネージド型サービスであり Elasticsearchを手軽に利用できます。
今回はDynamoDBからデータをロードしましたが、Amazon Kinesis Data StreamsやAmazon CloudWatchからもロードできるため今後試していこうと思います。

参考資料

[AWSエキスパート養成読本](https://www.amazon.co.jp/dp/4774179922) [はじめての Elasticsearch](https://qiita.com/nskydiving/items/1c2dc4e0b9c98d164329) [Amazon Elasticsearch Serviceを触ってみた](https://qiita.com/takayuki-miura0203/items/c2ad9527d818fd20ca93) [Amazon Elasticsearch Serviceについてまとめてみる](https://qiita.com/corn0214/items/ed777d7bfa3d5353fd5a) [[AWS]Amazon Elasticsearch Serviceでよく使うコマンド集(Amazon ES/Elasticsearch)](https://qiita.com/kenboo/items/0146b081f165b9e18736) [CloudFormationでDynamoDB streamを設定する](https://qiita.com/hir00/items/b3011d1e80fb1e94a711)