6

(Question credit: Fernando Abrao.)

I hear about the performance benefits of transducers in Clojure, but I'm not sure how to use them.

Say I have a qos/device-qos-range function that returns sequence of maps, some of which contain a decimal :samplevalue, like so:

[
  { :samplevalue 1.3, ... },
  { :othervalue -27.7, ... },
  { :samplevalue 7.5, ... },
  { :samplevalue 1.9, ... },
]

I'd like to see how many :samplevalues fall into each integer bin, like so:

(frequencies
  (reduce #(if (not (nil? (:samplevalue %2)))
             (conj %1 (.intValue (:samplevalue %2))))
          []
          (qos/device-qos-range origem device qos alvo inicio fim)))

;; => {1 2, 7 1}

How can I turn this into a fast version with transducers that eliminates intermediate data structures (such as the one returned by reduce)? Bonus points for code that can take advantage of multiple cores to do parallel processing.

4

1 回答 1

6

(Answer credit: Renzo Borgatti (@reborg).)

首先，让我们设置一些示例数据，稍后我们将使用这些数据进行性能测试。该向量包含 500k 个具有相同键的映射。值重叠 1/5 的时间。

(def data 
 (mapv hash-map 
       (repeat :samplevalue) 
       (concat (range 1e5)
               (range 1e5)
               (range 1e5)
               (range 1e5)
               (range 1e5))))

现在让我们使用传感器进行转换。请注意，此解决方案不是并行的。我将你缩短.intValue为 just int，它做同样的事情。此外，:samplevalue从每个地图有条件地获取可以缩短为 just (keep :samplevalue sequence)，相当于(remove nil? (map :samplevalue sequence)). 我们将使用Criterium进行基准测试。

(require '[criterium.core :refer [quick-bench]])
(quick-bench
  (transduce
    (comp
      (keep :samplevalue)
      (map int))
    (completing #(assoc! %1 %2 (inc (get %1 %2 0))) persistent!)
    (transient {})
    data))
;; My execution time mean: 405 ms

请注意，我们frequencies这次不是作为外部步骤调用。相反，我们将其融入到操作中。就像frequencies我们所做的一样，我们已经在瞬态哈希图上完成了操作，以获得额外的性能。我们通过使用瞬态哈希图作为种子和completing调用persistent!它的最终值来做到这一点。

我们可以使这个平行。为了获得最佳性能，我们使用可变的 JavaConcurrentHashMap而不是不可变的 Clojure 数据结构。

(require '[clojure.core.reducers :as r])
(import '[java.util HashMap Collections Map]
        'java.util.concurrent.atomic.AtomicInteger
        'java.util.concurrent.ConcurrentHashMap)

(quick-bench
  (let [concurrency-level (.availableProcessors (Runtime/getRuntime))
        m (ConcurrentHashMap. (quot (count data) 2) 0.75 concurrency-level)
        combinef (fn ([] m) ([_ _]))  ; just return `m` from the combine step
        rf (fn [^Map m k]
             (let [^AtomicInteger v (or (.get m k) (.putIfAbsent m k (AtomicInteger. 1)))]
               (when v (.incrementAndGet v))
               m))
        reducef ((comp (keep :samplevalue) (map int)) rf)]
    (r/fold combinef reducef data)
    (into {} m)))
;; My execution time mean: 70 ms

这里我们使用foldfromclojure.core.reducers库来实现并行。请注意，在并行上下文中，使用的任何转换器都需要是无状态的。另请注意， aConcurrentHashMap不支持nil用作键或值；幸运的是，我们不需要在这里这样做。

最后将输出转换为不可变的 Clojure 哈希图。您可以删除该步骤并仅使用 ConcurrentHashMap 实例来获得额外的加速 - 在我的机器上，删除该into步骤使整个过程fold大约需要 26 毫秒。

编辑 2017-11-20：用户 @clojuremostly 正确地指出，此答案的早期版本调用了初始化并发哈希映射实例quick-bench的块内部let，这意味着基准在所有运行中都使用了相同的实例。我把电话移到quick-bench了let街区外。它对结果没有显着影响。

于 2017-11-17T20:01:04.507 回答