elasticsearch摄取管道重命名动态字段

Question

我有一个相当大的文档，我正在将其引入 elasticsearch（70-80 个属性）。并且可能会定期出现新的领域。我可以使用重命名运算符有效地重命名每个字段，但我想知道是否有更有效的方法可以使用脚本动态重命名每个字段。

每个字段的当前格式可以是全部小写，下划线分隔，也可以是大小写组合，下划线分隔。我需要将每个字段重命名如下 -

（例如源字段）TV_field_1 -->（需要重命名为）tvField1

（例如源字段）feature_title_no -->（需要重命名为）featureTitleNo

谢谢

Answer 1

由于重命名处理器不支持替换模式，编写了自定义脚本

PUT _ingest/pipeline/snakeCaseToCamelCase
{
  "description": "Convert snake case to camel case",
  "processors": [
    {
      "script": {
        "source": """
            // Iterate through all keys and create list of fields with snake case
            def loopAllFields(def x){
              def ret=[];
              if(x instanceof Map){
                for (entry in x.entrySet()) {
                  if (entry.getKey().indexOf("_")==0) { 
                    continue;
                  }
                  // Get value
                  def val=entry.getValue();
                  if(entry.getKey().indexOf("_")>-1)
                  {
                      ret.add(entry.getKey());
                  }
                  // If further object
                  if(val instanceof Map|| val instanceof HashMap)
                  {
                    def list =loopAllFields(val);
                    for(item in list)
                    {
                      ret.add(entry.getKey()+"."+ item);
                    }
                  }
                  // If array
                  if(val instanceof ArrayList)
                  {
                    for(v in val)
                    {
                       def list =loopAllFields(v);
                       for(item in list)
                       {
                          ret.add(entry.getKey()+"."+ item);
                       }
                    }
                  }
                }
              }
              return ret;
            }

            // Create a camel case field and delete snake case field
            def renameField(def ctx, def fieldName)
            {
               def str=splitText(fieldName,'.');
               if(str.length<=1)
               {
                  def newField=snakeToCamel(fieldName);
                  ctx[newField]=ctx[fieldName];
                  ctx.remove(fieldName);
               }
               else
               {
                   if(ctx[str[0]] instanceof ArrayList)
                   {
                       def fld=combineArray(str);
                       for(v in ctx[str[0]])
                       {
                          renameField(v,fld);
                       }
                   }
                   if(ctx[str[0]] instanceof Map)
                   {
                      def fld=combineArray(str);
                      renameField(ctx[str[0]],fld); 
                    }
               }
               return 1;
            }

            def combineArray(def str)
            {
              def fld="";
              for(int i=1;i<str.length;i++)
              {
                if(fld=="")
                {
                   fld+=str[i];
                }
                else
                {
                   fld+="."+str[i];
                }
              }
              return fld;
            }

            // Convert field name from snake case to camel case
            def snakeToCamel(def s){
              //def str=/_/.split(s);
              def str=splitText(s,'_');
              def fieldName="";
              for(int i=0;i<str.length;i++)
              {
                 if(i==0)
                 {
                   fieldName+=str[i].toLowerCase();
                 }
                 else{
                    if(str[i].length()==1)
                    {
                      fieldName+=str[i].substring(0,1).toUpperCase();
                    }else{
                      fieldName+=str[i].substring(0,1).toUpperCase()+str[i].substring(1,str[i].length());
                    }
                 }
              }
              return fieldName;
            }

            def splitText(def str, def seperator)
            {
              def pos= str.indexOf(seperator);
              def ret=[];
              while(pos>0)
              {
                 def split=str.substring(0,pos);
                 def rest= str.substring(pos+1,str.length());
                 ret.add(split);
                 pos=rest.indexOf(seperator);
                 if(pos==-1)
                 {
                   ret.add(rest);
                 }
                 str=rest;
              }
              return ret;
            }

            def fields=loopAllFields(ctx);
            fields.sort((s1, s2) -> s2.length() - s1.length());
            for(field in fields)
            {
              renameField(ctx,field);
            }

"""
      }
    }
  ]
}