Elasticsearch 预处理以删除空字段作为摄取的一部分答案

【问题标题】：Elasticsearch pre-processing to remove null fields as part of ingestElasticsearch 预处理以删除空字段作为摄取的一部分
【发布时间】：2020-04-30 01:23:14
【问题描述】：

我有一个用例，其中我调用的 API 检索数据以放入 elasticsearch 正在返回空值。

我需要编写一个摄取管道，该管道使用处理器在将其写入弹性搜索之前删除所有空字段。处理器可能会也可能不会使用无痛脚本。

这是我目前从 API 获得的示例负载

{
    "master_desc": "TESTING PART",
    "date_added": "2019-10-24T09:30:03",
    "master_no": {
      "master_no": 18460110,
      "barcode": "NLSKYTEST1-1",
      "external_key": null,
      "umid": null
    }
  }

理想情况下，管道应该将文档插入为 -

{
    "master_desc": "TESTING PART",
    "date_added": "2019-10-24T09:30:03",
    "master_no": {
      "master_no": 18460110,
      "barcode": "NLSKYTEST1-1"
    }
}

注意，这些字段是动态的，所以我不能编写一个处理器来检查一组定义的字段是否存在空值。

谢谢！

【问题讨论】：

stackoverflow.com/questions/60538475/… 这可能对你有帮助

标签： elasticsearch elasticsearch-painless

【解决方案1】：

Null 字段未编入索引，也不可搜索。我已在管道下方编写以删除此类字段。请在所有场景中使用之前对其进行测试。使用此管道发布文档后，您将无法使用“存在”搜索空字段

管道：

PUT _ingest/pipeline/remove_null_fields
{
  "description": "Remove any null field",
  "processors": [
    {
      "script": {
        "source": """
         // return list of field with null values
          def loopAllFields(def x){
            def ret=[];
            if(x instanceof Map){
              for (entry in x.entrySet()) {
                if (entry.getKey().indexOf("_")==0) { 
                  continue;
                }
                def val=entry.getValue();
                if( val instanceof HashMap ||
                    val instanceof Map || 
                    val instanceof ArrayList)
                {
                  def list=[];
                  if(val instanceof ArrayList)
                  {
                    def index=0;
                    // Call for each object in arraylist
                    for(v in val)
                    {
                      list=loopAllFields(v);
                      for(item in list)
                      {
                          ret.add(entry.getKey()+"["+index+"]."+ item);
                      }
                      index++;
                    }
                  }
                  else
                  {
                    list =loopAllFields(val);
                  }
                  if(list.size()==val.size())
                  {
                    ret.add(entry.getKey());
                  }
                  else{
                        for(item in list)
                        {
                          ret.add(entry.getKey()+"."+ item);
                        }
                  }
                }

                if(val==null)
                {
                  ret.add(entry.getKey());
                }
              }
            }
            return ret;
          }


        /* remove fields from source, recursively deletes fields which part of other fields */
        def removeField(def ctx, def fieldname)
        {
            def pos=fieldname.indexOf(".");
            if(pos>0)
            {
              def str=fieldname.substring(0,pos);
              if(str.indexOf('[')>0 && str.indexOf(']')>0)
              {
                def s=str.substring(0,str.indexOf('['));
                def i=str.substring(str.indexOf('[')+1,str.length()-1);
                removeField(ctx[s][Integer.parseInt(i)],fieldname.substring(pos+1,fieldname.length())); 
              }
              else
              {
                  if(ctx[str] instanceof Map)
                  {
                    removeField(ctx[str],fieldname.substring(pos+1,fieldname.length()));
                  }
              }
            }else{

              ctx.remove(fieldname);
            } 
            return ctx;
        }

          def list=[];
          list=loopAllFields(ctx);
          for(item in list)
          {
             removeField(ctx,item);   
          }
"""
      }
    }
  ]
}

发布文件：

POST index8/_doc?pipeline=remove_null_fields
{
    "master_desc": "TESTING PART",
    "ddd":null,
    "date_added": "2019-10-24T09:30:03",
    "master_no": {
      "master_no": 18460110,
      "barcode": "NLSKYTEST1-1",
      "external_key": null,
      "umid": null
    }
}

结果：

"hits" : [
      {
        "_index" : "index8",
        "_type" : "_doc",
        "_id" : "06XAyXEBAWHHnYGOSa_M",
        "_score" : 1.0,
        "_source" : {
          "date_added" : "2019-10-24T09:30:03",
          "master_no" : {
            "master_no" : 18460110,
            "barcode" : "NLSKYTEST1-1"
          },
          "master_desc" : "TESTING PART"
        }
      }
    ]

【讨论】：

谢谢@jaspreet，上面的脚本是用无痛编写的吗？只是想确认一下，因为elastic 6.0及以上只支持painless，我不是很熟悉
@raja 它是用无痛编写的。请在对实际数据执行之前进行测试。我只检查了几个场景
所以脚本几乎可以工作了。然而，它并没有消除空对象、空数组或空值。已经发布了我在上面看到的更详细的解释
脚本看起来一样，你确定你更新了吗？
@raja 是的。只需要一点点改变。我添加了这些行 if(list.size()==val.size()) { ret.add(entry.getKey()); } else{ for(列表中的项目) { ret.add(entry.getKey()+"."+ item); }

【解决方案2】：

@Jaspreet，所以脚本几乎可以工作。然而，它并没有消除空对象、空数组或空值。这是我试图索引的文档 -

{
    "master_desc": "TESTING PART",
    "date_added": "2019-10-24T09:30:03",
    "master_no": {
      "master_no": 18460110,
      "barcode": "NLSKYTEST1-1",
      "external_key": null,
      "umid": null
    },
    "remote_sync_state": "",
    "lib_title_footage": [],
    "prj_no": {
       "prj_no": null,
       "prj_desc": null,
  }

以上返回-

{
    "master_desc": "TESTING PART",
    "date_added": "2019-10-24T09:30:03",
    "master_no": {
      "master_no": 18460110,
      "barcode": "NLSKYTEST1-1"
    },
    "remote_sync_state": "",
    "lib_title_footage": [ ],
    "prj_no": { }

我尝试更新脚本以检查这些模式的条件，但不幸的是出现了编译错误。

【讨论】：