AWS日記20 (Amazon Elasticsearch Service)

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

# システム構成図
![20201031.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/586707/bc7ba83f-4d5d-3f63-a96e-06304ee49b8c.jpeg)



# 準備
[AWS SAM の準備をします](https://qiita.com/tanaka_takurou/items/cae7c19fc3aaf0031abb)

[Amazon Elasticsearch Serviceの資料]
[Amazon Elasticsearch Service](https://aws.amazon.com/jp/elasticsearch-service/)


# AWS SAM テンプレート作成

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

[参考資料]
[AWS SAM テンプレートを作成する](https://docs.aws.amazon.com/ja_jp/codedeploy/latest/userguide/tutorial-lambda-sam-template.html)

<details><summary>template.yml</summary><div>

```yaml: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,'/'] ]
```
</div></details>

### Amazon Elasticsearch Serviceの設定は以下の部分
```yaml
  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リクエスト を使う
```golang
	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リクエスト を使う
```golang
	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からもロードできるため今後試していこうと思います。

<details><summary>参考資料</summary><div>
[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)