使用clojure的flambo中的zipWithUniqueId()

Question

我想要创建一个rdd，这样每一行都有一个索引。我尝试了以下方法

给出一个rdd：

["a" "b" "c"] 

(defn make-row-index [input]
  (let [{:keys [col]} input]
    (swap! @rdd assoc :rdd (-> (:rdd xctx)
                          (f/map #(vector %1 %2 ) (range))))))

期望产出：

 (["a" 0] ["b" 1] ["c" 2])

我遇到了一个错误，因为f/map被使用，因为(f/map rdd fn)想在apache中使用zipWithUniqueId()，但是我不知道如何实现它，并且在flambo中找不到等效的函数。如有任何建议和帮助，我们将不胜感激。

Apache-spark压缩和索引

flambo中的地图实现

谢谢

Answer 1

您可以简单地调用zipWithIndex，然后使用untuple调用map

(def rdd (f/parallelize sc ["a" "b" "c"]))
(f/map (.zipWithIndex rdd) f/untuple)

您可以完全以相同的方式使用.zipWithUniqueId，但结果将与您预期的不同。zipWithUniqueId将生成对，但索引字段不会被排序。

使用zip也是可能的，但据我所知，它不能在无限范围内工作。

(def idx (f/parallelize sc (range (f/count rdd))))
(f/map (.zip rdd idx) f/untuple)

无论何时使用zip，您都应该小心，不过一般来说，如果涉及到洗牌，RDD应该被视为无序集合。