CODEWITHSUNDEEP
pysparkapache-spark-sqldatabricks
Shasu
Shasu

Reputation: 501

How to solve "aggregateByKey is not a member of org.apache.spark.sql.Dataset" in Spark?

I am trying this example:

https://backtobazics.com/big-data/spark/apache-spark-aggregatebykey-example/

But instead of an RDD, I am using a dataframe.

I tried the following:

val aggrRDD = student_df.map(r => (r.getString(0), (r.getString(1), r.getInt(2))))
                       .aggregateByKey(zeroVal)(seqOp, combOp)

which is a part of a this code snippet :



val student_df = sc.parallelize(Array(
    ("Joseph", "Maths", 83), ("Joseph", "Physics", 74), ("Joseph", "Chemistry", 91), ("Joseph", "Biology", 82), 
    ("Jimmy", "Maths", 69), ("Jimmy", "Physics", 62), ("Jimmy", "Chemistry", 97), ("Jimmy", "Biology", 80), 
    ("Tina", "Maths", 78), ("Tina", "Physics", 73), ("Tina", "Chemistry", 68), ("Tina", "Biology", 87), 
    ("Thomas", "Maths", 87), ("Thomas", "Physics", 93), ("Thomas", "Chemistry", 91), ("Thomas", "Biology", 74), 
    ("Cory", "Maths", 56), ("Cory", "Physics", 65), ("Cory", "Chemistry", 71), ("Cory", "Biology", 68), 
    ("Jackeline", "Maths", 86), ("Jackeline", "Physics", 62), ("Jackeline", "Chemistry", 75), ("Jackeline", "Biology", 83), 
    ("Juan", "Maths", 63), ("Juan", "Physics", 69), ("Juan", "Chemistry", 64), ("Juan", "Biology", 60)), 3).toDF("student", "subject", "marks")


def seqOp = (accumulator: Int, element: (String, Int)) => 
    if(accumulator > element._2) accumulator else element._2

def combOp = (accumulator1: Int, accumulator2: Int) => 
    if(accumulator1 > accumulator2) accumulator1 else accumulator2


val zeroVal = 0

val aggrRDD = student_df.map(r => (r.getString(0), (r.getString(1), r.getInt(2))))
                       .aggregateByKey(zeroVal)(seqOp, combOp)

That gives this error :

error: value aggregateByKey is not a member of org.apache.spark.sql.Dataset[(String, (String, Int))]

A possible cause might be that a semicolon is missing before value aggregateByKey?

What am I doing wrong here? How do I work with dataframes or datasets on this?

Upvotes: 0

Views: 354

Answers (1)

dytyniak
dytyniak

Reputation: 394

Try to call rdd after student_df and before map:

val aggrRDD = student_df.rdd.map(r => (r.getString(0), (r.getString(1), r.getInt(2))))
          .aggregateByKey(zeroVal)(seqOp, combOp)

Upvotes: 2

Related Questions