Concatenating array objects in elasticsearch transform aggregations

[kbirk (Kevin Birk)]

I'm currently using an Elasticsearch transform in order to "merge" similar documents. It works very well, however I've run into an issue attempting to concatenate fields that contain an array of objects.

For example,

If I have two documents in the form:

{
    "_source": {
        "name": "xxx",
        "values": [
            { 
                "label": "xxx-a",
                "otherA": 123,
                "otherB": "abc"
            }
        ]
    }
}

{
    "_source": {
        "name": "xxx",
        "values": [
            { 
                "label": "xxx-b",
                "otherA": 456,
                "otherB": "def"
            }
        ]
    }
}

With the following mappings:

{
    "properties": {
        "name": {
            "type": "keyword"
        },
        "values": {
            "properties": {
                "label": {
                    "type": "keyword"
                },
                "otherA": {
                    "type": "integer"
                },
                "otherB": {
                    "type": "keyword"
                }
            }
        }
    }
}

My ideal result of the transform would be a single document in the form:

{
    "name": "xxx",
    "values": [
        { 
            "label": "xxx-a",
            "otherA": 123,
            "otherB": "abc"
        },
        { 
            "label": "xxx-b",
            "otherA": 456,
            "otherB": "def"
        }
    ]
}

My currrent transform pivot is:

{
  "pivot": {
    "group_by": {
      "name": {
        "terms": {
          "field": "name"
        }
      }
    },
    "aggregations": {
      "temporary.valuesLabel": {
        "terms": {
          "field": "values.label"
        }
      },
      "temporary.valuesOtherA": {
        "terms": {
          "field": "values.otherA"
        }
      },
      "temporary.valuesOtherB": {
        "terms": {
          "field": "values.otherB"
        }
      }
    }
  }
}

I use a pipeline with the following script to attempt to re-assemble the objects:

ctx.values = !ctx.temporary.valuesLabel.keySet().isEmpty() ? ctx.temporary.valuesLabel.keySet().stream().map(label -> ['label': label]).collect(Collectors.toList()) : [];

List valuesOtherA = !ctx.temporary.valuesOtherA.keySet().isEmpty() ? ctx.temporary.valuesOtherA.keySet().stream().collect(Collectors.toList()) : [];
List valuesOtherB = !ctx.temporary.valuesOtherB.keySet().isEmpty() ? ctx.temporary.valuesOtherB.keySet().stream().collect(Collectors.toList()) : [];
for (int i=0; i<ctx.values.length; i++) {
  ctx.values[i].otherA = Int.parseInt(valuesOtherA.get(i));
  ctx.values[i].otherB = valuesOtherB.get(i);
}

This almost works, except for the issue that when I do the individual terms aggs for each field in the values array, the terms agg can result with "misaligned" buckets, resulting in "mixed" objects".

Is there a way to have all the bucket indices get a fixed ordering? Is there a better way to achieve what I am trying to do?

Thank you for your time.

[Hendrik_Muhs]

I think you can do this easier with a scripted_metric. An advent post from 2 years ago discusses a similar case, unfortunately in german, but just use your favorite online translator:

Dec 5th, 2019 [DE] Weihnachtswünsche zusammenfassen mit Transforms

Weihnachtswünsche zusammenfassen mit Transforms Zu Weihnachten gibt es die schöne Tradition des Wunschzettels, Kinder können hier Ihre Wünsche aufschreiben, aber auch Sorgen und andere Dinge dem Weihnachtsmann mitteilen. In Deutschland gibt es mehrere Weihnachtspostämter, die jährlich hunderttausende Briefe beantworten, noch viel mehr Wunschzettel bleiben wahrscheinlich innerhalb der Familie und landen bei den Eltern oder Großeltern. Stellen wir uns vor, dass auch beim Weihnachtsmann Fachkräfte…

In short you need something like:

"values": {
  "scripted_metric": {
    "init_script": "state.docs = []",
    "map_script": "state.docs.add(new HashMap(params['_source']['values']))",
    "combine_script": "return state.docs",
    "reduce_script": "def docs = []; for (s in states) {for (d in s) { docs.add(d);}}return docs"
  }
}

You probably need to tweak the map part a bit to e.g. not fail if value is missing in a doc.

[kbirk (Kevin Birk)]

Thank you very much for the quick response. The scripted_metric works perfectly!

I needed to also remove duplicate entries (which existed in the data) so my final aggregation was:

"values": {
    "scripted_metric": {
        "init_script": "state.docs = []",
        "map_script": "def values = params['_source']['values'] != null ? new ArrayList(params['_source']['values']) : new ArrayList(); for (v in values) { state.docs.add(new HashMap(v)); }",
        "combine_script": "return state.docs",
        "reduce_script": "def dedupe = new HashSet(); def docs = []; for (s in states) {for (d in s) { if (!dedupe.contains(d)) { dedupe.add(d); docs.add(d); } } } return docs"
    }
}

Cheers