Clojure 中的 RDD 和向量

Question

给定一个函数来过滤掉大于从给定数据集 (RDD) 的子集确定的最大日期的日期，因此使用确定的最大日期来检查给定向量是否包含大于确定的最大日期我尝试了以下方法：

(defn future-rows
  "input = { :col val :qty-col val}
   :col = Date column reference"
   [ row input ]
   (let [{:keys [ col qty-col ]} input
     get-qty-max-date (-> (:rdd @scope-rdd)
                       (f/map #(if-not (or (s/blank? (get % qty-col)) (not (pos? (read-string (get % qty-col)))))
                                 (get % col) false))
                       (f/reduce #(if (pos? (compare % %2)) %1 %2)))]
(when-not (pos? (compare (get row col) get-qty-max-date)) row)))

这里row是一个vector。我面临的挑战是 get-qty-max-date 是 RDD 类型。如何以when-not形式进行比较？

注意：这个想法是 future-rows 函数将用作谓词

给定一个 RDD：

[[" " "2009/12/02"] ["4" "2005/02/08"] ["0" "2014/12/02"] ["5" "2005/08/01"] ["2" "2007/09/02"]]

当future-rows用作谓词时，期望的输出是：

[["4" "2005/02/08"] ["5" "2005/08/01"] ["2" "2007/09/02"]]

上面函数的输入是input { :col 1 :qty-col 0 } 确定的最大日期是 2007/09/02。因此，较大的日期 2009/12/02 和 2014/12/02 将从数据集中删除。

如果有任何其他方法可以做到这一点，我将不胜感激。

假设我们有一个主要功能可以做到这一点

(defn remove-rows [xctx input]
  (f/filter (:rdd xctx) #(future-rows row { :col 1 :qty-col 0 }))

将产生所需的输出

谢谢！

Answer 1

我猜你正在寻找这样的东西：

(defn not-empty-and-positive?
  [qty-col]
  (f/fn
   [row]
   (let [x (get row qty-col)]
     (not (or (s/blank? x) (neg? (read-string x)))))))


(defn get-max-date
  [col qty-col]
    (-> (:rdd @scope-rdd)
        (f/filter (not-empty-and-positive? qty-col))
        (f/map (f/fn [row] (get row col)))
        (.top 1)
        (first)))


(defn is-past?
    [col qty-col]
    (let [max-date (get-max-date col qty-col)]
      (f/fn [row] (neg? (compare (get row col) max-date)))))


(let [{:keys [ col qty-col ]} input
      not-empty-and-positive? (not-empty-and-positive? qty-col)
      is-past? (is-past? col qty-col)]
  (-> (f/filter rdd not-empty-and-positive?) (f/filter is-past? ) (f/collect)))

Clojure 中的 RDD 和向量

RDDs and vectors in clojure

clojure

apache-spark

rdd

flambo