【问题标题】:value join is not a member of org.apache.spark.rdd.RDD值连接不是 org.apache.spark.rdd.RDD 的成员
【发布时间】:2015-03-25 20:12:59
【问题描述】:

我收到此错误:

value join is not a member of 
    org.apache.spark.rdd.RDD[(Long, (Int, (Long, String, Array[_0])))
        forSome { type _0 <: (String, Double) }]

我发现的唯一建议是import org.apache.spark.SparkContext._我已经在这样做了。

我做错了什么?

编辑:更改代码以消除forSome(即,当对象具有org.apache.spark.rdd.RDD[(Long, (Int, (Long, String, Array[(String, Double)]))) 类型时)解决了问题。 这是 Spark 中的错误吗?

【问题讨论】:

  • joinorg.apache.spark.rdd.PairRDDFunctions 的成员。您忘记包含导致错误的代码行。
  • @DanielDarabos:代码是foo.join(bar)foo 的类型在错误消息中 - 它显然是 Pair 的子类型。
  • wdym?你有一个forSome 类型并且join 有效吗?
  • 对不起,我忽略了错误信息中类型的后半部分。这是个好问题。
  • 你的意思是支持者? :)

标签: scala apache-spark


【解决方案1】:

joinorg.apache.spark.rdd.PairRDDFunctions 的成员。那么为什么隐式类没有触发呢?

scala> val s = Seq[(Long, (Int, (Long, String, Array[_0]))) forSome { type _0 <: (String, Double) }]()
scala> val r = sc.parallelize(s)
scala> r.join(r) // Gives your error message.
scala> val p = new org.apache.spark.rdd.PairRDDFunctions(r)
<console>:25: error: no type parameters for constructor PairRDDFunctions: (self: org.apache.spark.rdd.RDD[(K, V)])(implicit kt: scala.reflect.ClassTag[K], implicit vt: scala.reflect.ClassTag[V], implicit ord: Ordering[K])org.apache.spark.rdd.PairRDDFunctions[K,V] exist so that it can be applied to arguments (org.apache.spark.rdd.RDD[(Long, (Int, (Long, String, Array[_0]))) forSome { type _0 <: (String, Double) }])
 --- because ---
argument expression's type is not compatible with formal parameter type;
 found   : org.apache.spark.rdd.RDD[(Long, (Int, (Long, String, Array[_0]))) forSome { type _0 <: (String, Double) }]
 required: org.apache.spark.rdd.RDD[(?K, ?V)]
Note: (Long, (Int, (Long, String, Array[_0]))) forSome { type _0 <: (String, Double) } >: (?K, ?V), but class RDD is invariant in type T.
You may wish to define T as -T instead. (SLS 4.5)
       val p = new org.apache.spark.rdd.PairRDDFunctions(r)
               ^
<console>:25: error: type mismatch;
 found   : org.apache.spark.rdd.RDD[(Long, (Int, (Long, String, Array[_0]))) forSome { type _0 <: (String, Double) }]
 required: org.apache.spark.rdd.RDD[(K, V)]
       val p = new org.apache.spark.rdd.PairRDDFunctions(r)

我确信错误信息对其他人来说是清楚的,但为了我自己的缓慢自我,让我们试着理解它。 PairRDDFunctions 有两个类型参数,KV。您的 forSome 是针对整对的,因此不能将其拆分为单独的 KV 类型。没有KV 可以让RDD[(K, V)] 等于您的RDD 类型。

但是,您可以让 forSome 仅应用于密钥,而不是整个密钥对。现在加入就可以了,因为这种类型可以分为KV

scala> val s2 = Seq[(Long, (Int, (Long, String, Array[_0])) forSome { type _0 <: (String, Double) })]()
scala> val r2 = sc.parallelize(2s)
scala> r2.join(r2)
res0: org.apache.spark.rdd.RDD[(Long, ((Int, (Long, String, Array[_0])) forSome { type _0 <: (String, Double) }, (Int, (Long, String, Array[_0])) forSome { type _0 <: (String, Double) }))] = MapPartitionsRDD[5] at join at <console>:26

【讨论】:

  • 谢谢!我想知道是否可以为 existing 序列在内部模式化 forSome
  • 我已经设法用s2.map { case (k, v) =&gt; k -&gt; v } 做相反的事情。 (不知道我在做什么......)
  • 啊,我觉得这行得通:s.map(_._1).zip(s.map(_._2))(压缩 RDD 也有同样的效果!)
  • 再想一想,在运行时将 CPU 周期花在这方面会很糟糕。你能用演员表吗?
  • 我解决了这个问题,但首先没有创建隐式类。
【解决方案2】:

考虑将 2 个 Spark RDD 连接在一起..

说,rdd1.first 的形式是(Int, Int, Float) = (1,957,299.98)rdd2.first 类似于(Int, Int) = (25876,1),连接应该发生在两个RDD 的第一个字段上。

scala> rdd1.join(rdd2) --- 导致错误:**:错误: 值连接不是 org.apache.spark.rdd.RDD[(Int, Int, 浮动)]

原因


两个 RDD 都应该是键值对的形式。

这里,rdd2 -- 形式为 (1,957,299.98) -- 不遵守此规则。而 rdd1 -- 形式为 (25876,1) -- 遵守。

分辨率


将第一个RDD的输出从(1,957,299.98)转换为(1,(957,299.98))形式的Key-Value对,再加入rdd2,如下图:

scala> val rdd1KV = rdd1.map(x=>(x.split(",")(1).toInt,(x.split(",")(2).toInt,x.split(",")(4).toFloat))) -- modified RDD

scala> rdd1KV.join(rdd2) -- join successful :)
res**: (Int, (Int, Float)) = (1,(957,299.98))

顺便说一句,join是org.apache.spark.rdd.PairRDDFunctions的成员。因此,请确保将其导入到 Eclipse 或 IDE 中,无论您想在何处运行代码。

文章也在我的博客上:

https://tips-to-code.blogspot.com/2018/08/apache-spark-error-resolution-value.html

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 1970-01-01
    • 2017-11-19
    • 2016-02-15
    • 1970-01-01
    • 2016-05-20
    • 2018-03-27
    • 2020-01-20
    相关资源
    最近更新 更多