Elasticsearch:使用 Python 进行 Bulk insert 及 Scan
2021年9月8日 | by mebius
在本文章中,我将展示如何使用 Python 来对索引进行 Bulk写入。在单个 BulkAPI 调用中执行多个索引或删除操作。 这减少了开销并且可以大大提高索引速度。
在今天的展示中,我将使用 Jupyter 来进行演示。如果你对 Jupyter 还不是很熟的话,请参阅我之前的文章 “Elasticsearch:使用 Jupyter Notebook 创建 tgcodePython 应用导入 CSV 文件”。
首先,我们来创建一个叫做 bulk 的 notebook。
jupyter notebook
我们输入如下的代码。它创建一个连接到 Elasticsearch 的实例:
from datetime import datetime
from elasticsearch import Elasticsearch
from elasticsearch import helpers
import time
es = Elasticsearch('localhost:9200')
我们接着输入如下的代码:
'''
Bulk inserts snippet with time calculation to insert.
'''
# source code: https://stackoverflow.com/questions/20288770/how-to-use-bulk-api-to-store-the-keywords-in-es-by-using-python
actions = [
{
"_index": "index8",
"_type": "_doc",
"_id": j,
"_source": {
"any":"data" + str(j),
"timestamp": datetime.now()}
}
for j in range(0, 2000)
]
上面的代码生成一个数组。数组中的项如下:
[{'_index': 'index8',
'_type': '_doc',
'_id': 0,
'_source': {'any': 'data0',
'timestamp': datetime.datetime(2021, 9, 8, 9, 32, 39, 247186)}},
...
如上所示,我们创建了2千个数据。在使用上面的时候,我们必须注意到的是,在 action 的tgcode定义中,我们可以定义 _op_type 字段。我们可以参考链接https://elasticsearch-py.readthedocs.io/en/master/helpers.html。我们可以像如下的方式这样添加 _op_type 字段以实现不同的操作:
{
'_op_type': 'delete',
'_index': 'index-name',
'_id': 42,
}
{
'_op_type': 'update',
'_index': 'index-name',
'_id': 42,
'doc': {'question': 'The life, universe and everything.'}
}
在默认的情况下,它的操作是 index。接下来,我们来计算一下批量导入这些数据所需要的时间到底tgcode是多少。
我们打入如下的代码:
st = time.time()
helpers.bulk(es, actions)
end = time.time()
print("total time:", end-st)
上面显示整个写入2千个数据所需的时间是 0.52 秒左右。我们可以看到是非常地高效。
接下来,我们使用另外一种方法来导入数据。我们通过每次写入一个数据的方法来进行写入。这样需要请求1万次才可以把1万个数据写入:
'''
Insertion iteratively
'''
st = time.time()
for j in range(0,2000):
doc = {
"any":"data" + str(j),
"timestamp": datetime.now()
}
es.index(index="test8", doc_type="_doc", id=j, body=doc)
end = time.time()
print("total time", end-st)
在 Kibana 中,我们发现 test8 索引的 count 是一直在增加,但是写入的速度是非常慢的:
等写完 2000 个数据,我们再看一下总的写入时间:
从上面,我们可以看出来总共需要 193 秒的时间。这个差距是非常之大的。
接下来,我们使用 helper 中的 scan 来获取大量的数据:
'''
Scanning over large data
'''
#source code: https://stackoverflow.com/questions/47722238/python-elasticsearch-helpers-scan-example
results = helpers.scan(es, index="chapter8",doc_type="doc", query={"query": {"match_all": {}}})
for item in results:
print(item['_id'], item['_source'])
我们发现这个 scan 的速度是非常之快的。几乎在咋眼之间就完成了。
文章来源于互联网:Elasticsearch:使用 Python 进行 Bulk insert 及 Scan
相关推荐: 使用 match_only_text 在日志数据集上节省 10% 的磁盘空间
Elasticsearch 7.14 引入了 match_only_text,这是一种新的字段类型,可在日志记录用例中直接替代文本字段类型,磁盘占用空间更少,从而降低成本。 Elasticsearch 对日志分析很有吸引力,因为它能够为日志消息建立索引。想要计…