关于ElasticSearch Bulk的用法

Background

刚接触ElasticSearch不久，公司让我帮忙去导一下数据，刚开始数量并不是很大，我都是用Elasticsearch python的api接口，一条一条数据去往新的index里插. 但是马上又有了千万级别的数据需要操作，如果还是老办法，会特别慢，而且不够稳定。于是去查了一下资料。关于Elasticsearch的python api文档很少，中文的就更少了，官方稳定又不是很详细，只是大概得到一个信息，可以用bulk() api去操作大量的数据。

Solution：

我需要解决的问题有两个：

1. 查询多个Index中的内容，然后将满足条件的数据写到新的一个Index中：
   这个问题比较简单，elasticsearch 的helpers module提供了一个完美的方法来做这件事：reindex()
   elasticsearch.helpers.reindex(client, source_index, target_index, query=None,target_client=None, chunk_size=500, scroll=u'5m', scan_kwargs={}, bulk_kwargs={})

这个方法的参数，提供了source_index(可以是List), target_index, query以及 scroll_size 和 scroll的保存时间,所以直接跑就可以了。

2. 批量更新现有Index中的所有数据，给每个document增加一个field并赋值：
   官方文档中的api 签名是这样的：
   elasticsearch.helpers.bulk(client, actions, stats_only=False, **kwargs)
   我一直没搞明白actions是什么类型的参数，以为是个函数类行的参数，后来看了一下源码，发现其实是一个List, 而且是要被操作的document的集合,官方文档上显示是要满足这个样子，跟search()返回的结果格式一样：

{ '_index': 'index-name', '_type': 'document', '_id': 42, '_parent': 5, '_ttl': '1d', '_source': { "title": "Hello World!", "body": "..." }}

但是又说：The bulk()
api accepts index, create, delete, and update actions. Use the _op_type field to specify an action (_op_type defaults to index):

{ '_op_type': 'delete', '_index': 'index-name', '_type': 'document', '_id': 42,}{ '_op_type': 'update', '_index': 'index-name', '_type': 'document', '_id': 42, 'doc': {'question': 'The life, universe and everything.'}}

我在自己的数据上加了"_op_type":"update", 然后运行一直出错：

TransportError(400, u'action_request_validation_exception',u'Validation Failed: 1: script or doc is missing

直到我尝试着删掉"_op_type"这个字段，终于运行成功了。以下是我的代码:

def queryInES( esinstance):
    search_body={"query":{"match_all":{}}}
    page = esinstance.search(index='my_index', body=search_body, search_type='scan', doc_type='Tweet', scroll='5s', size=1000)
    sid=page['_scroll_id']
    scroll_size = page['hits']['hits']

    while(scroll_size>0):
        pg = es.scroll(scroll_id=sid, scroll='30s')
        scroll_size = len(pg['hits']['hits'])
        print "scroll size: " + str(scroll_size)
        sid = pg['_scroll_id']
        data=pg['hits']['hits']
        ... ...

        for i in range(0, scroll_size):
            data[i]['_source']['attributes']['senti']={"label":label, "score": score, "confidence": confidence}

    helpers.bulk(client=esinstance, actions=data)