Amazon AWS - S3 到 ElasticSearch (Python Lambda)答案

【问题标题】：Amazon AWS - S3 to ElasticSearch (Python Lambda)Amazon AWS - S3 到 ElasticSearch (Python Lambda)
【发布时间】：2017-01-01 17:19:48
【问题描述】：

我想将数据从 S3 目录复制到 Amazon ElasticSearch 服务。我试过关注guide，但不幸的是我正在寻找的部分丢失了。我不知道 lambda 函数本身应该是什么样子（指南中有关此的所有信息都是：“将您的应用程序源代码放在 eslambda 文件夹中。”）。我希望 ES 自动索引文件。

目前我正在尝试

for record in event['Records']:
    bucket = record['s3']['bucket']['name']
    key = urllib.unquote_plus(record['s3']['object']['key'])
    index_name = event.get('index_name', key.split('/')[0])
    object = s3_client.Object(bucket, key)

    data = object.get()['Body'].read()

    helpers.bulk(es, data, chunk_size=100)

但我得到了一个巨大的错误说明 elasticsearch.exceptions.RequestError: TransportError(400, u'action_request_validation_exception', u'Validation Failed: 1: index is missing;2: type is missing;3: index is missing;4: type is missing;5: index is missing;6: type is missing;7: ...

谁能向我解释一下，我该如何设置，以便我的数据从 S3 移动到 ES 并自动映射和自动索引？显然这是可能的，如参考文献here 和here 中所述。

【问题讨论】：

标签： python amazon-web-services amazon-s3 aws-lambda amazon-elasticsearch

【解决方案1】：

我写了一个脚本从 S3 下载一个 csv 文件，然后将数据传输到 ES。

使用 boto3 制作 S3 客户端并从 S3 下载文件
制作了一个 ES 客户端来连接到 Elasticsearch。
打开 csv 文件并使用 elasticsearch 中的 helpers 模块将 csv 文件内容插入到弹性搜索中。

main.py

import boto3
from elasticsearch import helpers, Elasticsearch
import csv
import os
from config import *


#S3
Downloaded_Filename=os.path.basename(Prefix)
s3 = boto3.client('s3', aws_access_key_id=awsaccesskey,aws_secret_access_key=awssecretkey,region_name=awsregion)
s3.download_file(Bucket,Prefix,Downloaded_Filename)

#ES
ES_index = Downloaded_Filename.split(".")[0]
ES_client = Elasticsearch([ES_host],http_auth=(ES_user, ES_password),port=ES_port)

#S3 to ES
with open(Downloaded_Filename) as f:
    reader = csv.DictReader(f)
    helpers.bulk(ES_client, reader, index=ES_index, doc_type='my-type')

config.py

awsaccesskey = ""
awssecretkey = ""
awsregion = "us-east-1"
Bucket=""
Prefix=''
ES_host = "localhost"
ES_port = "9200"
ES_user = "elastic"
ES_password = "changeme"

【讨论】：

感谢您的帮助。考虑编辑您的帖子以突出/解释您的解决方案的重要部分以及它如何/为什么解决 OP 的问题。所以不鼓励只使用代码的答案。随着未来的访问者从您的帖子中学到一些东西，让他们深入了解自己的编码问题，高质量的答案会随着时间的推移而受到好评。仅代码转储迎合了“免费编码服务”的氛围，而 SO 不是。人们不太可能阅读代码转储来寻找见解，而无需您指出他们应该关注什么以及为什么。

【解决方案2】：

虽然可以在 Elasticsearch 中自动分配映射，但不会自动生成索引。您必须在 POST 请求中指定索引名称和类型。如果该索引不存在，则 Elasticsearch 将自动创建索引。

根据您的错误，您似乎没有通过索引和类型。

例如，下面是一个简单的 POST 请求如何将记录添加到索引 MyIndex 并键入 MyType，这将首先创建索引并在它不存在时键入。

curl -XPOST 'example.com:9200/MyIndex/MyType/' \ 
    -d '{"name":"john", "tags" : ["red", "blue"]}'

【讨论】：