值reduceByKey不是org.apache.spark.rdd.RDD的成员

lemanli

很可悲。我的Spark版本是2.1.1,Scala版本是2.11

import org.apache.spark.SparkContext._
import com.mufu.wcsa.component.dimension.{DimensionKey, KeyTrait}
import com.mufu.wcsa.log.LogRecord
import org.apache.spark.rdd.RDD

object PV {

//
  def stat[C <: LogRecord,K <:DimensionKey](statTrait: KeyTrait[C ,K],logRecords: RDD[C]): RDD[(K,Int)] = {
    val t = logRecords.map(record =>(statTrait.getKey(record),1)).reduceByKey((x,y) => x + y)

我得到这个错误

at 1502387780429
[ERROR] /Users/lemanli/work/project/newcma/wcsa/wcsa_my/wcsavistor/src/main/scala/com/mufu/wcsa/component/stat/PV.scala:25: error: value reduceByKey is not a member of org.apache.spark.rdd.RDD[(K, Int)]
[ERROR]     val t = logRecords.map(record =>(statTrait.getKey(record),1)).reduceByKey((x,y) => x + y)

有一个特质

trait KeyTrait[C <: LogRecord,K <: DimensionKey]{
  def getKey(c:C):K
}

编译,谢谢。

 def stat[C <: LogRecord,K <:DimensionKey : ClassTag : Ordering](statTrait: KeyTrait[C ,K],logRecords: RDD[C]): RDD[(K,Int)] = {
    val t = logRecords.map(record =>(statTrait.getKey(record),1)).reduceByKey((x,y) => x + y)

键需要覆盖Ordering [T]。

  object ClientStat extends KeyTrait[DetailLogRecord, ClientStat] {
      implicit val c

lientStatSorting = new Ordering[ClientStat] {
    override def compare(x: ClientStat, y: ClientStat): Int = x.key.compare(y.key)
  }

      def getKey(detailLogRecord: DetailLogRecord): ClientStat = new ClientStat(detailLogRecord)
    }
乔·克

这来自一般使用对rdd函数。reduceByKey方法实际上是PairRDDFunctions该类的方法,它具有以下隐式转换RDD

implicit def rddToPairRDDFunctions[K, V](rdd: RDD[(K, V)])
    (implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null): PairRDDFunctions[K, V]

因此,它需要几个隐式类型类。通常,当使用简单的具体类型时,这些已经在范围内。但是您应该能够修改您的方法,使其也需要那些相同的隐式:

def stat[C <: LogRecord,K <:DimensionKey](statTrait: KeyTrait[C ,K],logRecords: RDD[C])(implicit kt: ClassTag[K], ord: Ordering[K])

或使用更新的语法:

def stat[C <: LogRecord,K <:DimensionKey : ClassTag : Ordering](statTrait: KeyTrait[C ,K],logRecords: RDD[C])

本文收集自互联网,转载请注明来源。

如有侵权,请联系 [email protected] 删除。

编辑于
0

我来说两句

0 条评论
登录 后参与评论

相关文章

值toDS不是org.apache.spark.rdd.RDD的成员

值联接不是org.apache.spark.rdd.RDD的成员

值toDF不是org.apache.spark.rdd.RDD的成员

值 collectAsMap 不是 org.apache.spark.rdd.RDD 的成员

值查找不是org.apache.spark.rdd.RDD的成员

sortBy不是org.apache.spark.rdd.RDD的成员

值联接不是org.apache.spark.rdd.RDD [(Long,T)]的成员

Scala:出现错误 - mapPartitionsWithIndex 不是 org.apache.spark.rdd.RDD[Int] 的成员

saveAsTextFile不是Array [String] spark RDD的成员

Apache Spark RDD拆分“ |”

Apache Spark RDD替代

SparkStreaming和Kafka:值reduceByKey不是org.apache.spark.streaming.dstream.DStream [Any]的成员

Scala-如何过滤RDD org.apache.spark.rdd.RDD [String]]

Spark RDD.aggregate与RDD.reduceByKey?

Apache Spark: reading RDD from Spark Cluster

Apache Spark:按键将RDD对拆分为多个RDD以保存值

Spark:值reduceByKey不是成员

Apache Spark-使用2个RDD:RDD的补充

无法从现有 RDD 创建 RDD - Apache Spark

Spark MLlib如何将org.apache.spark.rdd.RDD [Array [Double]]转换为Array [Double]

等效于Apache Spark RDD中的getLines

Apache Spark-如何压缩多个RDD

Apache Spark:指向父RDD的引用指针

apache spark-从RDD迭代跳过

如何减少RDD在Apache Spark中的工作

使用Scala将org.apache.spark.mllib.linalg.Vector RDD转换为Spark中的DataFrame

值avro不是org.apache.spark.sql.DataFrameReader的成员

Zeppelin java.lang.NoClassDefFoundError:无法初始化类org.apache.spark.rdd.RDDOperationScope $

如何将org.apache.spark.ml.linalg.Vector的RDD转换为数据集?