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

Question

我正在根据 UDAF example 实现 UDAF 。更新阶段如下所示:

    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)，并将前一个缓冲区中的数据复制到其中。必须创建新 map 并复制所有元素，这看起来是一种可怕的浪费。所以我尝试了(在我的例子中，我有一张类型略有不同的 map ):

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)

是否可以更新现有 map ？更新此 map 的最佳方法是什么？

Answer 1

Isn't it possible to update the existing Map?

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

如果需要性能，则应避免将 UserDefinedAggregateFunction 与非原子类型一起使用。