2
  • 我在 mongoDB 中有很多文档。Mongo-connector 将这些数据插入到 elasticsearch。有没有办法，在插入到 ES 之前，我们可以在文档中添加额外的字段，然后插入到 elasticsearch 中？mongo-connector 有什么办法可以做到以上几点吗？

更新

根据您的更新 3 ，我创建了类似这样的映射，对吗？

PUT my_index2
{
 "mappings":{
  "my_type2": {
  "transform": {
  "script": {
    "inline": "if (ctx._source.geopoint.alt) ctx._source.geopoint.remove('alt')",
    "lang": "groovy"
  }
},
"properties": {
  "geopoint": {
    "type": "geo_point"
  }
 }
}
}
}

错误

这是我尝试插入映射时不断遇到的错误

{
   "error": {
  "root_cause": [
     {
        "type": "script_parse_exception",
        "reason": "Value must be of type String: [script]"
     }
  ],
  "type": "mapper_parsing_exception",
  "reason": "Failed to parse mapping [my_type2]: Value must be of type String: [script]",
  "caused_by": {
     "type": "script_parse_exception",
     "reason": "Value must be of type String: [script]"
  }
   },
   "status": 400
}

更新 2

现在映射被插入并得到确认为真。但是，当尝试插入 json 数据时，如下面的抛出错误。

PUT my_index2/my_type2/1
{
 "geopoint": {
        "lon": 48.845877,
        "lat": 8.821861,
        "alt": 0.0
        }
}

UPDATE2 错误

{
   "error": {
  "root_cause": [
     {
        "type": "mapper_parsing_exception",
        "reason": "failed to parse"
     }
  ],
  "type": "mapper_parsing_exception",
  "reason": "failed to parse",
  "caused_by": {
     "type": "illegal_argument_exception",
     "reason": "failed to execute script",
     "caused_by": {
        "type": "script_exception",
        "reason": "scripts of type [inline], operation [mapping] and lang [groovy] are disabled"
     }
  }
  },
  "status": 400
}

更新 2 的错误 1

添加 script.inline:true 后，尝试插入数据但出现以下错误。

{
   "error": {
  "root_cause": [
     {
        "type": "parse_exception",
        "reason": "field must be either [lat], [lon] or [geohash]"
     }
  ],
  "type": "mapper_parsing_exception",
  "reason": "failed to parse",
  "caused_by": {
     "type": "parse_exception",
     "reason": "field must be either [lat], [lon] or [geohash]"
  }
   },
   "status": 400
}
4

1 回答 1

4

mongo-connector 旨在将 Mongo 数据库与另一个目标系统同步，例如 ES、Solr 或另一个 Mongo DB。同步意味着 1:1 复制，因此我不知道 mongo-connector 在复制期间无法丰富文档（这也不是它的意图）。

但是，在 ES 5 中，我们很快将能够使用摄取节点，我们将能够在其中定义处理管道，其目标是在文档被索引之前丰富文档。

更新

可能有一种方法可以修改formatters.py文件。

在transform_value我会添加一个案例来处理Geopoint：

    if isinstance(value, dict):
        return self.format_document(value)
    elif isinstance(value, list):
        return [self.transform_value(v) for v in value]

    # handle Geopoint class
    elif isinstance(value, Geopoint):
        return self.format.document({'lat': value['lat'], 'lon': value['lon']})

    ...

更新 2

让我们通过修改transform_element函数来尝试另一种方法（第 104 行）：

def transform_element(self, key, value):
    try:
        # add these next two lines
        if key == 'GeoPoint':
            value = {'lat': value['lat'], 'lon': value['lon']}
        # do not modify the initial code below
        new_value = self.transform_value(value)
        yield key, new_value
    except ValueError as e:
        LOG.warn("Invalid value for key: %s as %s"
                 % (key, str(e)))

更新 3

您可能会尝试的另一件事是添加一个transform. 我之前没有提到它的原因是它在 ES 2.0 中已被弃用，但在 ES 5.0 中，您将拥有摄取节点，并且您可以在摄取时使用remove处理器来处理它

您可以像这样定义映射：

PUT my_index2
{
  "mappings": {
    "my_type2": {
      "transform": {
        "script": "ctx._source.geopoint.remove('alt'); ctx._source.geopoint.remove('valid')"
      },
      "properties": {
        "geopoint": {
          "type": "geo_point"
        }
      }
    }
  }
}

注意：确保通过添加并重新启动 ES 节点script.inline: true来启用动态脚本。elasticsearch.yml

将会发生的是该alt字段在存储中仍然可见，_source但不会被索引，因此不会发生错误。

使用 ES 5，您只需创建一个带有remove处理器的管道，如下所示：

PUT _ingest/pipeline/geo-pipeline
{
  "description" : "remove unsupported altitude field",
  "processors" : [
    {
      "remove" : {
        "field": "geopoint.alt"
      }
    }
  ]
}
于 2016-04-22T03:55:41.717 回答