网站建设与维护 pdf,wordpress安装上传文件,郑州大旗网站制作公司,wordpress 500一片空白ElasticSearch是广受欢迎的NoSQL数据库#xff0c;其分布式架构提供了极佳的数据空间的水平扩展能力#xff0c;同时保障了数据的可靠性#xff1b;反向索引技术使得数据检索和查询速度非常快。更多功能参见官网介绍 https://www.elastic.co/cn/elasticsearch/ 下面简单罗列…ElasticSearch是广受欢迎的NoSQL数据库其分布式架构提供了极佳的数据空间的水平扩展能力同时保障了数据的可靠性反向索引技术使得数据检索和查询速度非常快。更多功能参见官网介绍 https://www.elastic.co/cn/elasticsearch/ 下面简单罗列了通过Python访问ES的方法。
注本文不是Elasticsearch的入门介绍需要有ES基本知识。
Python - ElasticSearch 接口
Elastic提供的Python ElasticSearch原生接口源代码托管在Github上。项目链接和文档链接如下 https://github.com/elastic/elasticsearch-py https://www.elastic.co/guide/en/elasticsearch/client/python-api/7.17/examples.html#examples 下面是常见操作示例
建立ES连接 from elasticsearch import Elasticsearch, helpers
from elasticsearch.exceptions import ConnectionError, ConnectionTimeout, TransportError
...try :#es Elasticsearch(es_server, retry_on_timeoutTrue)es Elasticsearch(es_server, http_auth(es_user, es_pass), timeout30, max_retries10, retry_on_timeoutTrue) print(Connection failed, exit ...)sys.exit(1)
创建ES数据
doc {author: author_name,text: Interesting content...,timestamp: datetime.now(),
}
res es.index(indextest-index, id1, bodydoc)
获取ES数据
res es.get(indextest-index, id1)
通过查询获取ES数据 query{match_all:{}}try :result es.search(indexindex, queryquery, size10000)except([ConnectionError, ConnectionTimeout, TransportError]):print(Connection failed, exit ...)sys.exit(1)data[]for item in result[hits][hits] :data.push(item[_source])更新ES数据
doc {author: author_name,text: Interesting modified content...,timestamp: datetime.now(),
}
res es.update(indextest-index, id1, bodydoc)
删除ES数据
es.delete(indextest-index, id1)
ElasticSearch-DSL python接口
原生ES python接口在查询时需要编写复杂的DSL查询语句Elastic提供的ElasticSearch-DSL库极大地简化了查询语法方便编写查询语句。相关项目和文档的URL https://github.com/elastic/elasticsearch-dsl-py https://elasticsearch-dsl.readthedocs.io/en/latest/ 示例代码如下
from elasticsearch import Elasticsearch
from elasticsearch_dsl import Searchclient Elasticsearch()s Search(usingclient, indexmy-index) \.filter(term, categorysearch) \.query(match, titlepython) \.exclude(match, descriptionbeta)s.aggs.bucket(per_tag, terms, fieldtags) \.metric(max_lines, max, fieldlines)response s.execute()for hit in response:print(hit.meta.score, hit.title)for tag in response.aggregations.per_tag.buckets:print(tag.key, tag.max_lines.value)
ElasticSearch - Pandas 接口
Pandas是流行的大数据处理Python库Elastic提供了Pandas DataFrame的接口 可以直接将索引数据表中的数据放到 pandas 的 dataframe 中非常方便。相关项目和文档URL如下 https://github.com/elastic/eland https://eland.readthedocs.io/en/latest/reference/dataframe.html 注意返回的并不是原生Pandas DataFrame而是Elastic自己的实现但并没有实现所有DataFrame的功能。
示例代码如下
import eland as ed# Connecting to an Elasticsearch instance running on localhost:9200
df ed.DataFrame(localhost:9200, es_index_patternflights)也可以先建立 ES 连接
# Connecting to an Elastic Cloud instance
from elasticsearch import Elasticsearches Elasticsearch(localhost:9200,http_auth(elastic, password)
)
df ed.DataFrame(es, es_index_patternflights)第三方 ElasticSearch - Pandas 接口
eland虽然可以方便将 Elastic 中的数据转换为 dataframe但没有提供将 dataframe 保存到 Elastic的接口。这时我们需要使用第三方的接口。es_pandas是开源的 ES Pandas接口可以直接将ES查询得到的数据以Pandas DataFrame的方式返回也可将 dataframe 保存到 Elastic 中。 https://github.com/fuyb1992/es_pandas 初始化与ES的连接
import pandas as pd
from es_pandas import es_pandas...
epcon None
try :epcon es_pandas(esurl)
except Exception as e:logger.error(Initializa DB connection failed! Error[{}].format(str(e)))
从ES表中获取数据返回格式为Pandas DataFrame
# 从ES表中获取数据返回DataFrame
try: if query is None:data epcon.to_pandas(dbname, infer_dtypeTrue, show_progressFalse)else:data epcon.to_pandas(dbname, infer_dtypeTrue, show_progressFalse, query_rulequery)
except exceptions.NotFoundError:logger.debug(Not found data. Params: dbname[{}] query[{}].format(dbname, query))
将Pandas DataFrame中的数据写入ES表中
# 将DataFrame中的数据写入ES表中
ret True
try:epcon.to_es(df, dbname, use_indexTrue, _op_typecreate, thread_count2, chunk_size10000, show_progressFalse)
except ConnectionError:ret Falselogger.error(Save data failed! Params: dbname[{}] data[{}],, connection error!.format(dbname, df))
将Pandas DataFrame中的数据更新到ES表中
# 将DataFrame中的更新到ES表中
ret True
try:epcon.to_es(df, dbname, use_indexTrue, _op_typeupdate, thread_count2, chunk_size10000, show_progressFalse)
except ConnectionError:ret Falselogger.error(Update data failed! Params: dbname[{}] data[{}],, connection error!.format(dbname, df))
将Pandas DataFrame中的数据从ES表中删除
# 将DataFrame中的数据从ES表中删除
ret True
try:epcon.to_es(df, dbname, use_indexTrue, _op_typedelete, thread_count2, chunk_size10000, show_progressFalse)
except ConnectionError:ret Falselogger.error(Delete data failed! Params: dbname[{}] data[{}],, connection error!.format(dbname, df))
