在UDAF的每个更新步骤上创建一个新的累加器

问题描述

我正在根据 UDAF示例. update阶段看起来像这样:

I am implementing an UDAF according to UDAF example. the update phase there looks like this:

    public void update(MutableAggregationBuffer buffer, Row input) {
    if (!input.isNullAt(0)) {
        String inputKey = input.getString(0);
        Map<String, Long> inputValues = input.<String, Long>getJavaMap(1);
        Map<String, Map<String, Long>> newData = new HashMap<>();

        if (!buffer.isNullAt(0)) {
            Map<String, Map<String, Long>> currData = buffer.<String, Map<String, Long>>getJavaMap(0);
            newData.putAll(currData);
        }
        newData.put(inputKey, inputValues);
        buffer.update(0, newData);
    }
}

您可以看到，在每个步骤上都会创建一个新的HashMap(newData)，并将先前缓冲区中的数据复制到其中.看起来很浪费，不得不创建新的Maps并复制所有元素.因此，我尝试了一下(在我的情况下，我的地图类型稍有不同):

You can see that on every step a new HashMap is created (newData) and the data from the previous buffer is copied into it. It looks like an awful waste, having to create new Maps and copying all the elements. So I tried (in my case I have a map with a slightly different types):

bufferJavaMap = buffer.<String, Integer>getJavaMap(0);
bufferJavaMap.put("aaaa", 1);
buffer.update(0, bufferJavaMap);

我收到以下错误:

java.lang.UnsupportedOperationException
   at java.util.AbstractMap.put(AbstractMap.java:209)
   at dns.MergeMapUDAF.update(MergeMapUDAF.java:84)

不可能更新现有的地图吗?更新此地图的最佳方法是什么?

Isn't it possible to update the existing Map? what is the best method update this Map?

推荐答案

这是不可能的，但是问题比您所确定的问题更为复杂. Spark在get和update上都制作了结构的完整副本，因此即使删除显式副本也无法解决问题.

It is not possible, but the problem is more complex than the one identified in your. Spark makes a full copy of the structure on both get and update so even removing explicit copy wouldn't resolve the problem.

如果需要性能，则应避免将UserDefinedAggregateFunction用于非原子类型.

If performance is required, you should avoid using UserDefinedAggregateFunction with non-atomic types.

这篇关于在UDAF的每个更新步骤上创建一个新的累加器的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！