标签: elasticsearch-py

我正在尝试为我的数据添加时间戳,使用elasticsearch-py批量索引,然后使用kibana显示数据.

我的数据显示在kibana中,但我的时间戳没有被使用.当我在配置索引模式后进入"发现"选项卡时,我得到0结果(是的,我尝试调整搜索时间).

这是我的批量索引json的样子:

{'index': 
         {'_timestamp': u'2015-08-11 14:18:26', 
          '_type': 'webapp_fingerprint', 
          '_id': u'webapp_id_redacted_2015_08_13_12_39_34',
          '_index': 'webapp_index'
         }
}

****JSON DATA HERE***

这将被elasticsearch接受并将被导入Kibana,但_timestamp字段实际上不会被索引(当在"时间字段名称"下配置索引模式时,它确实显示在下拉列表中).

我也试过像这样格式化metaFields:

{'index': {
           '_type': 'webapp_fingerprint', 
           '_id': u'webapp_id_redacted_2015_08_13_12_50_04', 
           '_index': 'webapp_index'
          }, 
           'source': {
                      '_timestamp': {
                                     'path': u'2015-08-11 14:18:26',
                                     'enabled': True, 
                                     'format': 'YYYY-MM-DD HH:mm:ss'
                                    }
                     }
}

这也行不通.

最后,我尝试在索引中包含_timestamp字段并应用格式,但我在弹性搜索时出错.

{'index': {
           '_timestamp': {
                          'path': u'2015-08-11 14:18:26', 
                          'enabled': True, 
                          'format': 'YYYY-MM-DD HH:mm:ss'
                         }, 
           '_type': 'webapp_fingerprint', 
           '_id': u'webapp_id_redacted_2015_08_13_12_55_53', 
           '_index': 'webapp_index'
          }
}

错误是:

elasticsearch.exceptions.TransportError: 
TransportError(500,u'IllegalArgumentException[Malformed action/metadata 
line [1], expected a simple value for field [_timestamp] …

Run Code Online (Sandbox Code Playgroud)

我正在使用ElasticSearch批量Python API,它是否同时提供同步和异步api？

根据docs，这应该足够了：

"settings": {
    "index.mapping.ignore_malformed": true 
  }

但是如何在 python 包装器上实现这一点呢？我当前的代码如下所示：

from elasticsearch_dsl import Index

index = Index('my_index', my_conn)
index.settings(
     number_of_shards=ES_NUMBER_OF_SHARDS,
     number_of_replicas=ES_NUMBER_OF_REPLICAS
)
index.create()

我正在尝试使用从脚本调用的Python客户端https://github.com/elastic/elasticsearch-py（也在容器中运行）为容器化的Elasticsearch db编制索引。
通过查看现有的代码片段，看来这docker-compose是用于我的目的的有用工具。我的目录结构是

• docker-compose.yml
• indexer
  • Dockerfile
  • indexer.py
  • requirements.txt
• elasticsearch
  • Dockerfile

我的docker-compose.yml读物

version: '3'

services:
  elasticsearch:
    build: elasticsearch/
    ports: 
      - 9200:9200
    networks:
      - deploy_network
    container_name: elasticsearch

  indexer:
    build: indexer/
    depends_on:
      - elasticsearch
    networks:
      - deploy_network
    container_name: indexer

networks:
  deploy_network:
    driver: bridge

indexer.py 读

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

es = Elasticsearch(hosts=[{"host":'elasticsearch'}]) # what should I put here?

actions = [
    {
    '_index' : 'test',
    '_type' : 'content',
    '_id' : str(item['id']),
    '_source' : …

使用官方ElasticSearch Python 库（文档）

我创建一个索引：

doc = {
    "something": "123a",
    "somethingelse": "456b",
    "timestamp": datetime.now(),
    "history": []
}
es.index(index="someindex", doc_type="somedoctype", id="someid", body=doc)

我想每次都将项目附加到历史记录中，而不是覆盖它们：

es.update(index="someindex", doc_type="somedoctype", id="someid",
          body={"doc": {
                "history": {
                    "123abc": "abc", "456def": "def", "timestamp": datetime.now()
                }
               }
          })

我必须在第二个代码片段中更改什么才能将其附加到历史记录数组/列表中，而不是每次都覆盖它？

我有一个跨多个索引的搜索查询。要根据文档启用此功能，我需要提供一个以逗号分隔的索引列表。

但是当我尝试这样做时： es.search(index='index1,index2',body=body)

我没有得到任何结果： {u'hits': {u'hits': [], u'total': 0, u'max_score': None}, u'_shards': {u'successful': 10, u'failed': 0, u'skipped': 0, u'total': 10}, u'took': 1, u'timed_out': False}

但是， index='_all' 可以搜索所有索引。我在这里做错了什么还是这个功能有问题？谢谢。

首先，我想让大家知道我知道ElasticSearch Scroll API如何工作的基本工作逻辑。要使用Scroll API，首先，我们需要使用一些滚动值（如1m ）调用search方法，然后它将返回一个_scroll_id，该_scroll_id将用于 Scroll 上的下一个连续调用，直到所有文档在循环中返回。但问题是我只想在多线程的基础上使用相同的进程，而不是串行。例如：

if index_name is not None and doc_type is not None and body is not None:
   es = init_es()
   page = es.search(index_name,doc_type, scroll = '30s',size = 10, body = body)
   sid = page['_scroll_id']
   scroll_size = page['hits']['total']

   # Start scrolling
   while (scroll_size > …