zipWithUniqueId() в flambo using clojure

Question

Я хочу создать 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]) 

я получил ошибку Arity, так как F/карта используется в качестве (f/map rdd fn) Требуется использовать zipWithUniqueId() в апача искра, но я теряюсь на том, как реализовать это, и я не могу найти эквивалентную функцию в flambo. Любые предложения и помощь приветствуются.

Apache-spark zip with Index

Map implementation in flambo

Благодаря

Answer 1

Вы можете просто позвонить zipWithIndex с последующим map использованием untuple:

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

Вы можете использовать .zipWithUniqueId точно так же, но результат будет другим от того, что вы expec т. zipWithUniqueId будет генерировать пары, но поле индекса не будет упорядочено.

Следует также использовать zip, но, насколько я могу судить, он не работает с бесконечным диапазоном.

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

Всякий раз, когда вы используете zip вы должны быть осторожны, хотя Вообще говоря РДД следует рассматривать как неупорядоченные коллекции, если есть перетасовка участия.