I'm getting TypeError when i try to convert RDD of Price object to Pair RDD
.
Sample code:
priceRDD = pppConformInDF.rdd.map(lambda row: Price(row.vyge_id, row.strm_typ_cd, row.sfb_nm, row.txn_dt, row.vfa_extra_am, '2'))
priceKeyValueRDD = priceRDD.map(lambda price: (",".join([price.vyge_id, price.strm_typ_cd, price.sfb_nm]), list(price)))
How do i convert the RDD of Price object to Key value RDD, the value is the Price object and Key is combination of some of the fields in Price object.
Error log:
priceKeyValueRDD = priceRDD.map(lambda price: (",".join([price.vyge_id, price.strm_typ_cd, price.sfb_nm]), list(price)))
TypeError: 'Price' object is not iterable
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.handlePythonException(PythonRunner.scala:298)
at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:438)
at org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRunner.scala:421)
at org.apache.spark.api.python.BasePythonRunner$ReaderIterator.hasNext(PythonRunner.scala:252)
at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28)
at scala.collection.generic.Growable$class.$plus$plus$eq(Growable.scala:59)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:104)
at scala.collection.mutable.ArrayBuffer.$plus$plus$eq(ArrayBuffer.scala:48)
at scala.collection.TraversableOnce$class.to(TraversableOnce.scala:310)
at org.apache.spark.InterruptibleIterator.to(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce$class.toBuffer(TraversableOnce.scala:302)
at org.apache.spark.InterruptibleIterator.toBuffer(InterruptibleIterator.scala:28)
at scala.collection.TraversableOnce$class.toArray(TraversableOnce.scala:289)
at org.apache.spark.InterruptibleIterator.toArray(InterruptibleIterator.scala:28)
at org.apache.spark.api.python.PythonRDD$$anonfun$3.apply(PythonRDD.scala:149)
at org.apache.spark.api.python.PythonRDD$$anonfun$3.apply(PythonRDD.scala:149)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2074)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.base/java.lang.Thread.run(Unknown Source)
The error your got is because of list(price)
operation, as price cannot convert to a list.
Just remove list
, then use groupByKey()
to aggregate prices with the same key into a list.
priceRDD = pppConformInDF.rdd.map(lambda row: Price(row.vyge_id, row.strm_typ_cd, row.sfb_nm, row.txn_dt, row.vfa_extra_am, '2'))
priceKeyValueRDD = priceRDD.map(lambda price: (",".join([price.vyge_id, price.strm_typ_cd, price.sfb_nm]), price))
priceGroupedRDD = priceKeyValueRDD.groupByKey() # returns something like [(key->iterable of prices)]