Elasticsearch：使用 Python 进行 Bulk insert 及 Scan

在本文章中，我将展示如何使用 Python 来对索引进行 Bulk 写入。在单个 Bulk API 调用中执行多个索引或删除操作。这减少了开销并且可以大大提高索引速度。

在今天的展示中，我将使用 Jupyter 来进行演示。如果你对 Jupyter 还不是很熟的话，请参阅我之前的文章 “Elasticsearch：使用 Jupyter Notebook 创建 Python 应用导入 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 的定义中，我们可以定义 _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。接下来，我们来计算一下批量导入这些数据所需要的时间到底是多少。

我们打入如下的代码：

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相关推荐

bulkwrite 批量插入_SQL SERVER 使用BULK Insert将txt文件中的数据批量插入表中（1）
1/首先建立数据表 CREATE TABLE BasicMsg ( RecvTime FLOAT NOT NULL , --接收时间,不存在时间相同的数据 AA INT NOT NULL, --24位 ...
navicat使用查询向表中插入一行记录_SQL--每日一解------Bulk Insert 快速插入
SQL INSERT INTO 语句 INSERT INTO 语句用于向表中插入新记录. SQL INSERT INTO 语法 INSERT INTO 语句可以有两种编写形式. ① 第一种形式无需指定 ...
SSIS常用的包—大量插入任务（Bulk Insert task）
大量插入任务允许像BULK INSERT语句或者bcp.exe命令行工具一样从txt文件(也叫做平面文件)中插入数据.这个task工具箱中的Control Flow Items中,它不会产生数据流.这 ...
sql server 2005 T-SQL BULK INSERT (Transact-SQL)
以用户指定的格式将数据文件导入数据库表或视图. Transact-SQL 语法约定语法 BULK INSERT [ database_name . [ schema_name ] . | schem ...
azure 导入 bak_如何使用BULK INSERT在本地和Azure中导入数据
azure 导入 bak 介绍 (Introduction) BULK INSERT is a popular method to import data from a local file to S ...
用BULK INSERT命令导入数据详解
转载而来.来源已经不清楚了. 如果你从事与数据库相关的工作,有可能会涉及到将数据从外部数据文件插入倒SQL Server的操作.本文将为大家演示如何利用BULK INSERT命令来导入数据,并讲解怎样 ...
Elasticsearch：如何使用 Elasticsearch 和 Python 构建面部识别系统
你是否曾经尝试在图像中搜索目标? Elasticsearch 可以帮助你存储,分析和搜索图像或视频中的目标. 在本快速教程中,我们将向你展示如何构建一个使用 Python 进行面部识别的系统. 了解有 ...
MyBatis批量插入（sqlserver BULK INSERT）
MyBatis批量插入: 1. foreach方式 2.sqlsession + sqlsession.flushStatements方式: // SqlSession sqlSessi ...
使用BULK INSERT高效导入大量数据到SQL Server数据库
源数据 (文本文件) 下载了大量的股票历史数据, 都是文本格式的: 每个文件第一行包含股票代码, 股票名称, 数据类型. 第二行是数据列的名称: 数据表在数据库中新建了一个数据表TestStock, ...
BULK INSERT如何将大量数据高效地导入SQL Server
转载自:http://database.51cto.com/art/201108/282631.htm BULK INSERT如何将大量数据高效地导入SQL Server 本文我们详细介绍了BULK ...

Elasticsearch：使用 Python 进行 Bulk insert 及 Scan

Elasticsearch：使用 Python 进行 Bulk insert 及 Scan相关推荐

最新文章

热门文章