在clojure中实现flambo mapValues函数

Question

我有一个clojure函数，它使用flambov0.60函数api对示例数据集进行一些分析。我注意到，当我使用(get rdd 2)而不是在rdd集合中获取第二个元素时，它将获得rdd集合的第一个元素的第二个字符。我的假设是clojure将rdd集合的每一行视为一个整体字符串，而不是一个向量，以使我能够获得集合中的第二个元素。我正在考虑使用map-value函数将映射的值转换为一个向量，然后我可以获得第二个元素，我尝试了以下方法：

(defn split-on-tab-transformation [xctx input]
 (assoc xctx :rdd (-> (:rdd xctx)
                   (spark/map (spark/fn [row] (s/split row #"\t")))
                   (spark/map-values vec)))) 

不幸的是，我遇到了一个错误：java.lang.IllegalArgumentException: No matching method found: mapValues for class org.apache.spark.api.java.JavaRDD...

这是代码，返回rdd中的第一个集合：(假设我删除了上面函数中的(spark/map-values vec) )

(defn get-distinct-column-val
 "input = {:col val}"
  [ xctx input ]
  (let [rdds (-> (:rdd xctx)
           (f/map (f/fn [row] row))
           f/first)]
(clojure.pprint/pprint rdds)))

输出：

[2.00000 770127      200939.000000   \t6094\tBENTONVILLE, AR DPS\t22.500000\t5.000000\t2.500000\t5.000000\t0.000000\t0.000000\t0.000000\t0.000000\t0.000000\t1\tStore Tab\t0.000000\t4.50\t3.83\t5.00\t0.000000\t0.000000\t0.000000\t0.000000\t19.150000]

如果我试图获得第二个元素770127

(defn get-distinct-column-val
 "input = {:col val}"
  [ xctx input ]
  (let [rdds (-> (:rdd xctx)
           (f/map (f/fn [row] row))
           f/first)]
   (clojure.pprint/pprint (get rdds 1)))

我得到：

[\.]

地图值的Flambo文档

我是新来的克洛尔，我很感谢你的帮助。谢谢

Answer 1

首先，map-values (或Spark中的mapValues )是一个仅在PairRDD上的有效转换(例如，类似于这个[:foo [1 2 3]] )。具有类似于此值的RDD可以解释为某种映射，其中第一个元素是键，第二个是值。

如果您有这样的RDD，mapValues会在不更改键的情况下转换值。在这种情况下，您应该使用第二个映射，尽管它似乎过时了，因为clojure.string/split已经返回了一个向量。

使用map-values的一个简单示例

(let [pairs [(ft/tuple :foo 1) (ft/tuple :bar 2)]
      rdd (f/parallelize-pairs sc pairs) ;; Note parallelize-pairs -> PairRDD
      result (-> rdd       
          (f/map-values inc) ;; Map values
          (f/collect))]
  (assert (= result [(ft/tuple :foo 2) (ft/tuple :bar 3)])))

从您的描述来看，您使用的是输入RDD，而不是从split-on-tab-transformation返回的RDD。如果我不得不猜测您正在尝试使用原始的xctx，而不是从split-on-tab-transformation返回的那个。由于Clojure maps是不可变的，所以assoc不更改传递的参数，get-distinct-column-val接收RDD[String]而不是RDD[Array[String]]

根据命名约定，我假设您希望为数组中的单个位置获取不同的值。为了清晰起见，我删除了您代码中未使用的部分。首先让我们创建虚拟数据：

(spit "data.txt"
      (str "Mazda RX4\t21\t6\t160\n"
           "Mazda RX4 Wag\t21\t6\t160\n"
           "Datsun 710\t22.8\t4\t108\n"))

添加函数的重写版本

(defn split-on-tab-transformation [xctx]
   (assoc xctx :rdd (-> (:rdd xctx)
                        (f/map #(clojure.string/split % #"\t")))))

(defn get-distinct-column-val
  [xctx col]
    (-> (:rdd xctx)
      (f/map #(get % col))
        (f/distinct)))

和结果

(assert
 (= #{"Mazda RX4 Wag" "Datsun 710" "Mazda RX4"}
    (-> {:sc sc :rdd (f/text-file sc "data.txt")}
      (split-on-tab-transformation)
      (get-distinct-column-val 0)
      (f/collect)
      (set))))