Search code examples
pysparkrdd

pyspark - TypeError: 'Price' object is not iterable


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)

Solution

  • 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)]